Merge pull request #1340 from aspnet/release/2.1

Return stream method error to client (#1331)

src/Microsoft.AspNetCore.SignalR.Core/HubEndPoint.cs

@@ -400,6 +400,11 @@ namespace Microsoft.AspNetCore.SignalR
                     await SendMessageAsync(connection, new StreamItemMessage(invocationId, enumerator.Current));
                 }
             }
+            catch (ChannelClosedException ex)
+            {
+                // If the channel closes from an exception in the streaming method, grab the innerException for the error from the streaming method
+                error = ex.InnerException == null ? ex.Message : ex.InnerException.Message;
+            }
             catch (Exception ex)
             {
                 // If the streaming method was canceled we don't want to send a HubException message - this is not an error case

test/Microsoft.AspNetCore.SignalR.Tests/HubEndpointTestUtils/Hubs.cs

@@ -3,6 +3,7 @@
 
 using System;
 using System.Collections.Generic;
+using System.Reactive.Linq;
 using System.Threading;
 using System.Threading.Channels;
 using System.Threading.Tasks;
@@ -533,6 +534,11 @@ namespace Microsoft.AspNetCore.SignalR.Tests.HubEndpointTestUtils
             return Channel.CreateUnbounded<string>().Reader;
         }
 
+        public IObservable<int> ThrowStream()
+        {
+            return Observable.Throw<int>(new Exception("Exception from observable"));
+        }
+
         private class CountingObservable : IObservable<string>
         {
             private int _count;

test/Microsoft.AspNetCore.SignalR.Tests/HubEndpointTests.cs

@@ -1386,6 +1386,31 @@ namespace Microsoft.AspNetCore.SignalR.Tests
             }
         }
 
+        [Fact]
+        public async Task ReceiveCorrectErrorFromStreamThrowing()
+        {
+            var serviceProvider = HubEndPointTestUtils.CreateServiceProvider();
+            var endPoint = serviceProvider.GetService<HubEndPoint<StreamingHub>>();
+
+            using (var client = new TestClient())
+            {
+                var endPointLifetime = endPoint.OnConnectedAsync(client.Connection);
+
+                await client.Connected.OrTimeout();
+
+                var messages = await client.StreamAsync(nameof(StreamingHub.ThrowStream));
+
+                Assert.Equal(1, messages.Count);
+                var completion = messages[0] as CompletionMessage;
+                Assert.NotNull(completion);
+                Assert.Equal("Exception from observable", completion.Error);
+
+                client.Dispose();
+
+                await endPointLifetime.OrTimeout();
+            }
+        }
+
         public static IEnumerable<object[]> StreamingMethodAndHubProtocols
         {
             get

test/Microsoft.AspNetCore.SignalR.Tests/Microsoft.AspNetCore.SignalR.Tests.csproj

@@ -36,6 +36,7 @@
     <PackageReference Include="Microsoft.Extensions.Logging.Console" Version="$(MicrosoftExtensionsLoggingConsolePackageVersion)" />
     <PackageReference Include="Microsoft.Extensions.Logging.Debug" Version="$(MicrosoftExtensionsLoggingDebugPackageVersion)" />
     <PackageReference Include="Microsoft.Extensions.Logging" Version="$(MicrosoftExtensionsLoggingPackageVersion)" />
+    <PackageReference Include="System.Reactive.Linq" Version="$(SystemReactiveLinqPackageVersion)" />
   </ItemGroup>
 
 </Project>