fix: Use shared token on underlying stream process

This way, when the stop command is called, the underlying stream process should be signaled to cancel right away too.

Author: carlreid

src/StreamMaster.Streams/Broadcasters/SourceBroadcaster.cs

@@ -106,20 +106,21 @@ public async Task<long> SetSourceStreamAsync(SMStreamInfo SMStreamInfo, Cancella
 
         this.SMStreamInfo = SMStreamInfo;
 
+        // Start a new streaming task
+        _cancellationTokenSource = new CancellationTokenSource();
+
         Stopwatch stopwatch = Stopwatch.StartNew();
         try
         {
             (Stream? stream, int processId, ProxyStreamError? error) =
-                await streamFactory.GetStream(SMStreamInfo, cancellationToken).ConfigureAwait(false);
+                await streamFactory.GetStream(SMStreamInfo, _cancellationTokenSource.Token).ConfigureAwait(false);
             stopwatch.Stop();
             if (stream == null || error != null)
             {
                 logger.LogError("Could not create source stream for channel broadcaster: {Id} {Name} {Error}", SMStreamInfo.Id, SMStreamInfo.Name, error?.Message);
                 return 0;
             }
 
-            // Start a new streaming task
-            _cancellationTokenSource = new CancellationTokenSource();
             _streamingTask = Task.Run(() => RunPipelineAsync(stream, SMStreamInfo.Name, cancellationToken: _cancellationTokenSource.Token), _cancellationTokenSource.Token);
             return stopwatch.ElapsedMilliseconds;
         }
@@ -193,6 +194,11 @@ private async Task RunPipelineAsync(Stream sourceStream, string name, int buffer
                     timeoutCts?.Dispose();
                 }
 
+                if (cancellationToken.IsCancellationRequested)
+                {
+                    break;
+                }
+
                 if (bytesRead == 0)
                 {
                     if (!hasReadData)

src/StreamMaster.Streams/Factories/CommandExecutor.cs

@@ -34,6 +34,7 @@ public GetStreamResult ExecuteCommand(CommandProfileDto commandProfile, string s
 
             using var registration = cancellationToken.Register(() =>
             {
+                logger.LogDebug("Cancellation requested for Stream process");
                 GracefullyTerminateProcess();
             });
 
@@ -46,7 +47,20 @@ public GetStreamResult ExecuteCommand(CommandProfileDto commandProfile, string s
                 return new GetStreamResult(null, -1, error);
             }
 
+            if (!RuntimeInformation.IsOSPlatform(OSPlatform.Windows))
+            {
+                try
+                {
+                    Process.Start("setpgrp", $"{_process.Id}");
+                }
+                catch (Exception ex)
+                {
+                    logger.LogWarning(ex, "Failed to set process group for {ProcessId}", _process.Id);
+                }
+            }
+
             string stderrFilePath = Path.Combine(BuildInfo.CommandErrorFolder, $"stderr_{_process.Id}.log");
+            Directory.CreateDirectory(Path.GetDirectoryName(stderrFilePath)!);
             errorWriter = new StreamWriter(stderrFilePath, append: true, Encoding.UTF8);
 
             // Clean up older logs to keep only the latest 10
@@ -64,6 +78,7 @@ public GetStreamResult ExecuteCommand(CommandProfileDto commandProfile, string s
                 }
             };
             _process.BeginErrorReadLine();
+            _process.EnableRaisingEvents = true; // Ensure Exited event is raised
             _process.Exited += Process_Exited;
 
             stopwatch.Stop();
@@ -242,6 +257,16 @@ private static void ConfigureProcess(Process process, string commandExec, string
         process.StartInfo.RedirectStandardOutput = true;
         process.StartInfo.RedirectStandardError = true;
         process.StartInfo.StandardOutputEncoding = Encoding.UTF8;
+        process.StartInfo.StandardErrorEncoding = Encoding.UTF8;
+        process.StartInfo.WindowStyle = ProcessWindowStyle.Hidden;
+
+        if (!RuntimeInformation.IsOSPlatform(OSPlatform.Windows))
+        {
+            process.StartInfo.Environment["SM_PROCESS_ID"] = process.Id.ToString();
+            process.StartInfo.Environment["SM_PROCESS_TYPE"] = "STREAM";
+        }
+
+        process.EnableRaisingEvents = true;
     }
 
     /// <summary>
@@ -273,4 +298,4 @@ public void Dispose()
         _disposed = true;
         GC.SuppressFinalize(this);
     }
-}
+}