[C#] fix: When streaming operation failed the exception is suppressed.

dotnet/packages/Microsoft.TeamsAI/Microsoft.TeamsAI.Tests/Application/StreamingResponseTests.cs

@@ -5,6 +5,7 @@
 using Microsoft.Teams.AI.Application;
 using Microsoft.Teams.AI.Exceptions;
 using Microsoft.Teams.AI.Tests.TestUtils;
+using Moq;
 
 namespace Microsoft.Teams.AI.Tests.Application
 {
@@ -234,5 +235,27 @@ void CaptureSend(Activity[] arg)
             Assert.Equal(2, streamer.UpdatesSent());
             Assert.Single(streamer.Attachments);
         }
+
+        [Fact]
+        public async Task Test_SendActivityThrowsException_AssertThrows()
+        {
+            // Arrange
+            Activity[]? activitiesToSend = null;
+            void CaptureSend(Activity[] arg)
+            {
+                activitiesToSend = arg;
+            }
+            var adapter = new SimpleAdapter(CaptureSend);
+            var turnContextMock = new Mock<ITurnContext>();
+            turnContextMock.Setup((tc) => tc.SendActivityAsync(It.IsAny<Activity>(), It.IsAny<CancellationToken>())).ThrowsAsync(new Exception("Forbidden operation"));
+
+            // Act
+            StreamingResponse streamer = new(turnContextMock.Object);
+            Exception ex = await Assert.ThrowsAsync<TeamsAIException>(() => streamer.EndStream());
+
+
+            // Assert
+            Assert.Equal("Error occured when sending activity while streaming: Forbidden operation", ex.Message);
+        }
     }
 }

dotnet/packages/Microsoft.TeamsAI/Microsoft.TeamsAI/Application/StreamingResponse.cs

@@ -121,7 +121,7 @@ public Task EndStream()
             QueueNextChunk();
 
             // Wait for the queue to drain
-            return this._queueSync!;
+            return WaitForQueue()!;
         }
 
         /// <summary>
@@ -133,7 +133,7 @@ private void QueueActivity(Func<Activity> factory)
             this._queue.Add(factory);
 
             // If there's no sync in progress, start one
-            if (this._queueSync == null)
+            if (this._queueSync == null || this._queueSync.IsCompleted)
             {
                 this._queueSync = DrainQueue();
             }
@@ -198,25 +198,20 @@ private void QueueNextChunk()
         /// </summary>
         private async Task DrainQueue()
         {
-            await Task.Run(async () =>
+            try
             {
-                try
+                while (this._queue.Count > 0)
                 {
-                    while (this._queue.Count > 0)
-                    {
-                        // Get next activity from queue
-                        Activity activity = _queue[0]();
-                        await SendActivity(activity).ConfigureAwait(false);
-                        _queue.RemoveAt(0);
-                    }
-                }
-
-                finally
-                {
-                    // Queue is empty, mark as idle
-                    this._queueSync = null;
+                    // Get next activity from queue
+                    Activity activity = _queue[0]();
+                    await SendActivity(activity).ConfigureAwait(false);
+                    _queue.RemoveAt(0);
                 }
-            }).ConfigureAwait(false);
+            }
+            catch (Exception ex)
+            {
+                throw new TeamsAIException($"Error occured when sending activity while streaming: {ex.Message}", ex);
+            }
         }
 
         /// <summary>