Fix streaming hub methods combined with async (#1544)

src/Microsoft.AspNetCore.SignalR.Core/Internal/DefaultHubDispatcher.cs

@@ -372,6 +372,9 @@ namespace Microsoft.AspNetCore.SignalR.Internal
 
         private static bool IsStreamed(Type resultType)
         {
+            // TODO: cache reflection for performance, on HubMethodDescriptor maybe?
+            resultType = UnwrapTask(resultType);
+
             var observableInterface = IsIObservable(resultType) ?
                 resultType :
                 resultType.GetInterfaces().FirstOrDefault(IsIObservable);
@@ -393,6 +396,9 @@ namespace Microsoft.AspNetCore.SignalR.Internal
         {
             if (result != null)
             {
+                // TODO: cache reflection for performance, on HubMethodDescriptor maybe?
+                resultType = UnwrapTask(resultType);
+
                 var observableInterface = IsIObservable(resultType) ?
                     resultType :
                     resultType.GetInterfaces().FirstOrDefault(IsIObservable);
@@ -420,6 +426,16 @@ namespace Microsoft.AspNetCore.SignalR.Internal
             }
         }
 
+        private static Type UnwrapTask(Type type)
+        {
+            if (type.IsGenericType && type.GetGenericTypeDefinition() == typeof(Task<>))
+            {
+                return type.GetGenericArguments()[0];
+            }
+
+            return type;
+        }
+
         private static bool IsIObservable(Type iface)
         {
             return iface.IsGenericType && iface.GetGenericTypeDefinition() == typeof(IObservable<>);

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

@@ -513,6 +513,12 @@ namespace Microsoft.AspNetCore.SignalR.Tests.HubEndpointTestUtils
             return new CountingObservable(count);
         }
 
+        public async Task<IObservable<string>> CounterObservableAsync(int count)
+        {
+            await Task.Yield();
+            return CounterObservable(count);
+        }
+
         public ChannelReader<string> CounterChannel(int count)
         {
             var channel = Channel.CreateUnbounded<string>();
@@ -529,6 +535,12 @@ namespace Microsoft.AspNetCore.SignalR.Tests.HubEndpointTestUtils
             return channel.Reader;
         }
 
+        public async Task<ChannelReader<string>> CounterChannelAsync(int count)
+        {
+            await Task.Yield();
+            return CounterChannel(count);
+        }
+
         public ChannelReader<string> BlockingStream()
         {
             return Channel.CreateUnbounded<string>().Reader;

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

@@ -1453,7 +1453,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
         {
             get
             {
-                foreach (var method in new[] { nameof(StreamingHub.CounterChannel), nameof(StreamingHub.CounterObservable) })
+                foreach (var method in new[] { nameof(StreamingHub.CounterChannel), nameof(StreamingHub.CounterChannelAsync), nameof(StreamingHub.CounterObservable), nameof(StreamingHub.CounterObservableAsync) })
                 {
                     foreach (var protocol in new IHubProtocol[] { new JsonHubProtocol(), new MessagePackHubProtocol() })
                     {