feat: Wrap Stream for termination handling

Author: carlreid

src/StreamMaster.Streams/Broadcasters/SourceBroadcaster.cs

@@ -30,8 +30,10 @@ public class SourceBroadcaster(ILogger<ISourceBroadcaster> logger, StreamConnect
 
     /// <inheritdoc/>
     public StreamHandlerMetrics Metrics => StreamMetricsRecorder.Metrics;
+
     public bool IsMultiView { get; set; }
     public CancellationToken CancellationToken { get; } = cancellationToken;
+
     /// <inheritdoc/>
     public ConcurrentDictionary<string, IStreamDataToClients> ChannelBroadcasters { get; } = new();
 
@@ -72,6 +74,7 @@ public bool RemoveChannelBroadcaster(string Id)
     {
         return ChannelBroadcasters.TryRemove(Id, out _);
     }
+
     public async Task<long> SetSourceMultiViewStreamAsync(IChannelBroadcaster channelBroadcaster, CancellationToken cancellationToken)
     {
         logger.LogInformation("Setting source stream {Name} to {StreamName}", Name, SMStreamInfo.Name);
@@ -269,6 +272,7 @@ private async Task RunPipelineAsync(Stream sourceStream, string name, int buffer
     public async Task StopAsync()
     {
         await _stopLock.WaitAsync().ConfigureAwait(false);
+        Task? taskToAwait = null;
         try
         {
             if (Interlocked.CompareExchange(ref _isStopped, 1, 0) == 0)
@@ -277,12 +281,29 @@ public async Task StopAsync()
                 {
                     _cancellationTokenSource?.Cancel();
                 }
+                taskToAwait = _streamingTask;
             }
         }
         finally
         {
             _stopLock.Release();
         }
+
+        if (taskToAwait != null)
+        {
+            try
+            {
+                await taskToAwait.ConfigureAwait(false);
+            }
+            catch (OperationCanceledException)
+            {
+                logger.LogDebug("Task was already cancelled");
+            }
+            catch (Exception ex)
+            {
+                logger.LogError(ex, "Error during SourceBroadcaster streaming task completion wait.");
+            }
+        }
     }
 
     /// <inheritdoc/>