diff --git a/CliWrap.Tests/EventStreamSpecs.cs b/CliWrap.Tests/EventStreamSpecs.cs index 6d280657..916be528 100644 --- a/CliWrap.Tests/EventStreamSpecs.cs +++ b/CliWrap.Tests/EventStreamSpecs.cs @@ -120,6 +120,23 @@ public async Task I_can_execute_a_command_as_a_push_based_event_stream_and_not_h await cmd.Observe().ToArray(); } + [Fact(Timeout = 15000)] + public async Task I_can_execute_a_command_as_a_push_based_event_stream_and_the_underlying_process_is_killed_if_the_subscription_is_abandoned() + { + // Arrange + var cmd = Cli.Wrap(Dummy.Program.FilePath).WithArguments(["sleep", "00:00:20"]); + + // Act: subscribe but take only the first event (StartedCommandEvent) then dispose + var startedEvent = await cmd.Observe().OfType().FirstAsync(); + + // Assert: the process is not left running in the background. + // Kill() is system-level asynchronous, so we poll until it takes effect. + while (Process.IsRunning(startedEvent.ProcessId)) + await Task.Delay(10); + + Process.IsRunning(startedEvent.ProcessId).Should().BeFalse(); + } + [Fact(Timeout = 15000)] public async Task I_can_execute_a_command_as_a_pull_based_event_stream_and_the_underlying_process_is_killed_if_the_iterator_is_abandoned() { @@ -136,7 +153,11 @@ public async Task I_can_execute_a_command_as_a_pull_based_event_stream_and_the_u break; } - // Assert: the process is not left running in the background + // Assert: the process is not left running in the background. + // Kill() is system-level asynchronous, so we poll until it takes effect. + while (Process.IsRunning(processId)) + await Task.Delay(10); + Process.IsRunning(processId).Should().BeFalse(); } } diff --git a/CliWrap/EventStream/PushEventStreamCommandExtensions.cs b/CliWrap/EventStream/PushEventStreamCommandExtensions.cs index 1e4a0dd5..69531645 100644 --- a/CliWrap/EventStream/PushEventStreamCommandExtensions.cs +++ b/CliWrap/EventStream/PushEventStreamCommandExtensions.cs @@ -31,6 +31,12 @@ CancellationToken gracefulCancellationToken ) => Observable.CreateSynchronized(observer => { + // Used to kill the process if the subscription is disposed (abandoned) + // before the command finishes executing. + var killCts = CancellationTokenSource.CreateLinkedTokenSource( + forcefulCancellationToken + ); + var stdOutPipe = PipeTarget.Merge( command.StandardOutputPipe, PipeTarget.ToDelegate( @@ -47,11 +53,12 @@ CancellationToken gracefulCancellationToken ) ); - // Execute the command with the pipes extended to push events to the observer + // Execute the command with killCts.Token as the forceful cancellation token, + // so that disposing the subscription (which cancels killCts) also kills the process. var commandTask = command .WithStandardOutputPipe(stdOutPipe) .WithStandardErrorPipe(stdErrPipe) - .ExecuteAsync(forcefulCancellationToken, gracefulCancellationToken); + .ExecuteAsync(killCts.Token, gracefulCancellationToken); observer.OnNext(new StartedCommandEvent(commandTask.ProcessId)); @@ -86,7 +93,16 @@ CancellationToken gracefulCancellationToken // doesn't get reported to the finalizer thread and crash the process. .Task.ObserveException(); - return Disposable.Null; + // Return a disposable that cancels killCts to terminate the process if the + // subscription is disposed before the command completes. + // This is also triggered on normal completion (OnCompleted/OnError dispose + // the subscription), but by then the process has already exited, so Kill() + // is a no-op. + return Disposable.Create(() => + { + killCts.Cancel(); + killCts.Dispose(); + }); }); ///