Use IAsyncEnumerable returned by ChannelReader.ReadAllAsync()

halter73 · halter73 · commit 0d00933f9350 · 2019-02-21T19:09:55.000-08:00
diff --git a/src/SignalR/samples/SignalRSamples/SignalRSamples.csproj b/src/SignalR/samples/SignalRSamples/SignalRSamples.csproj
@@ -2,6 +2,7 @@
 
   <PropertyGroup>
     <TargetFramework>netcoreapp3.0</TargetFramework>
+    <LangVersion>8.0</LangVersion>
   </PropertyGroup>
 
   <ItemGroup>
diff --git a/src/SignalR/server/Core/src/Internal/AsyncEnumerableAdapters.cs b/src/SignalR/server/Core/src/Internal/AsyncEnumerableAdapters.cs
@@ -17,9 +17,9 @@ public static IAsyncEnumerable<object> MakeCancelableAsyncEnumerable<T>(IAsyncEn
             return new CancelableAsyncEnumerable<T>(asyncEnumerable, cancellationToken);
         }
 
-        public static IAsyncEnumerable<object> GetAsyncEnumerableFromChannel<T>(ChannelReader<T> channel, CancellationToken cancellationToken = default)
+        public static IAsyncEnumerable<object> MakeCancelableAsyncEnumerableFromChannel<T>(ChannelReader<T> channel, CancellationToken cancellationToken = default)
         {
-            return new ChannelAsyncEnumerable<T>(channel, cancellationToken);
+            return MakeCancelableAsyncEnumerable(channel.ReadAllAsync(), cancellationToken);
         }
 
         /// <summary>Converts an IAsyncEnumerable of T to an IAsyncEnumerable of object.</summary>
@@ -36,15 +36,19 @@ public CancelableAsyncEnumerable(IAsyncEnumerable<T> asyncEnumerable, Cancellati
 
             public IAsyncEnumerator<object> GetAsyncEnumerator(CancellationToken cancellationToken = default)
             {
+                // Assume that this will be iterated through with await foreach which always passes a default token.
+                // Instead use the token from the ctor.
                 Debug.Assert(cancellationToken == default);
-                return new CancelableAsyncEnumerator(_asyncEnumerable.GetAsyncEnumerator(_cancellationToken));
+
+                var enumeratorOfT = _asyncEnumerable.GetAsyncEnumerator(_cancellationToken);
+                return enumeratorOfT as IAsyncEnumerator<object> ?? new BoxedAsyncEnumerator(enumeratorOfT);
             }
 
-            private class CancelableAsyncEnumerator : IAsyncEnumerator<object>
+            private class BoxedAsyncEnumerator : IAsyncEnumerator<object>
             {
                 private IAsyncEnumerator<T> _asyncEnumerator;
 
-                public CancelableAsyncEnumerator(IAsyncEnumerator<T> asyncEnumerator)
+                public BoxedAsyncEnumerator(IAsyncEnumerator<T> asyncEnumerator)
                 {
                     _asyncEnumerator = asyncEnumerator;
                 }
@@ -62,74 +66,5 @@ public ValueTask DisposeAsync()
                 }
             }
         }
-
-        /// <summary>Provides an IAsyncEnumerable of object for the data in a channel.</summary>
-        private class ChannelAsyncEnumerable<T> : IAsyncEnumerable<object>
-        {
-            private readonly ChannelReader<T> _channel;
-            private readonly CancellationToken _cancellationToken;
-
-            public ChannelAsyncEnumerable(ChannelReader<T> channel, CancellationToken cancellationToken)
-            {
-                _channel = channel;
-                _cancellationToken = cancellationToken;
-            }
-
-            public IAsyncEnumerator<object> GetAsyncEnumerator(CancellationToken cancellationToken = default)
-            {
-                Debug.Assert(cancellationToken == default);
-                return new ChannelAsyncEnumerator(_channel, _cancellationToken);
-            }
-
-            private class ChannelAsyncEnumerator : IAsyncEnumerator<object>
-            {
-                /// <summary>The channel being enumerated.</summary>
-                private readonly ChannelReader<T> _channel;
-                /// <summary>Cancellation token used to cancel the enumeration.</summary>
-                private readonly CancellationToken _cancellationToken;
-                /// <summary>The current element of the enumeration.</summary>
-                private T _current;
-
-                public ChannelAsyncEnumerator(ChannelReader<T> channel, CancellationToken cancellationToken)
-                {
-                    _channel = channel;
-                    _cancellationToken = cancellationToken;
-                }
-
-                public object Current => _current;
-
-                public ValueTask<bool> MoveNextAsync()
-                {
-                    var result = _channel.ReadAsync(_cancellationToken);
-
-                    if (result.IsCompletedSuccessfully)
-                    {
-                        _current = result.Result;
-                        return new ValueTask<bool>(true);
-                    }
-
-                    return new ValueTask<bool>(MoveNextAsyncAwaited(result));
-                }
-
-                private async Task<bool> MoveNextAsyncAwaited(ValueTask<T> channelReadTask)
-                {
-                    try
-                    {
-                        _current = await channelReadTask;
-                    }
-                    catch (ChannelClosedException ex) when (ex.InnerException == null)
-                    {
-                        return false;
-                    }
-
-                    return true;
-                }
-
-                public ValueTask DisposeAsync()
-                {
-                    return default;
-                }
-            }
-        }
     }
 }
diff --git a/src/SignalR/server/Core/src/Internal/DefaultHubDispatcher.cs b/src/SignalR/server/Core/src/Internal/DefaultHubDispatcher.cs
@@ -397,7 +397,7 @@ private ValueTask CleanupInvocation(HubConnectionContext connection, HubMethodIn
             return scope.DisposeAsync();
         }
 
-         private async Task StreamResultsAsync(string invocationId, HubConnectionContext connection, IAsyncEnumerable<object> enumerable, IServiceScope scope,
+        private async Task StreamResultsAsync(string invocationId, HubConnectionContext connection, IAsyncEnumerable<object> enumerable, IServiceScope scope,
             IHubActivator<THub> hubActivator, THub hub, CancellationTokenSource streamCts, HubMethodInvocationMessage hubMethodInvocationMessage)
         {
             string error = null;
diff --git a/src/SignalR/server/Core/src/Internal/HubMethodDescriptor.cs b/src/SignalR/server/Core/src/Internal/HubMethodDescriptor.cs
@@ -19,12 +19,12 @@ internal class HubMethodDescriptor
             .GetRuntimeMethods()
             .Single(m => m.Name.Equals(nameof(AsyncEnumerableAdapters.MakeCancelableAsyncEnumerable)) && m.IsGenericMethod);
 
-        private static readonly MethodInfo GetAsyncEnumerableFromChannelMethod = typeof(AsyncEnumerableAdapters)
+        private static readonly MethodInfo MakeCancelableAsyncEnumerableFromChannelMethod = typeof(AsyncEnumerableAdapters)
             .GetRuntimeMethods()
-            .Single(m => m.Name.Equals(nameof(AsyncEnumerableAdapters.GetAsyncEnumerableFromChannel)) && m.IsGenericMethod);
+            .Single(m => m.Name.Equals(nameof(AsyncEnumerableAdapters.MakeCancelableAsyncEnumerableFromChannel)) && m.IsGenericMethod);
 
-        private readonly MethodInfo _convertToEnumerableMethodInfo;
-        private Func<object, CancellationToken, IAsyncEnumerable<object>> _convertToEnumerable;
+        private readonly MethodInfo _makeCancelableEnumerableMethodInfo;
+        private Func<object, CancellationToken, IAsyncEnumerable<object>> _makeCancelableEnumerable;
 
         public HubMethodDescriptor(ObjectMethodExecutor methodExecutor, IEnumerable<IAuthorizeData> policies)
         {
@@ -46,14 +46,14 @@ public HubMethodDescriptor(ObjectMethodExecutor methodExecutor, IEnumerable<IAut
                 if (openReturnType == typeof(IAsyncEnumerable<>))
                 {
                     StreamReturnType = returnType.GetGenericArguments()[0];
-                    _convertToEnumerableMethodInfo = MakeCancelableAsyncEnumerableMethod;
+                    _makeCancelableEnumerableMethodInfo = MakeCancelableAsyncEnumerableMethod;
                     break;
                 }
 
                 if (openReturnType == typeof(ChannelReader<>))
                 {
                     StreamReturnType = returnType.GetGenericArguments()[0];
-                    _convertToEnumerableMethodInfo = GetAsyncEnumerableFromChannelMethod;
+                    _makeCancelableEnumerableMethodInfo = MakeCancelableAsyncEnumerableFromChannelMethod;
                     break;
                 }
             }
@@ -110,19 +110,19 @@ public HubMethodDescriptor(ObjectMethodExecutor methodExecutor, IEnumerable<IAut
         public IAsyncEnumerable<object> FromReturnedStream(object stream, CancellationToken cancellationToken)
         {
             // there is the potential for compile to be called times but this has no harmful effect other than perf
-            if (_convertToEnumerable == null)
+            if (_makeCancelableEnumerable == null)
             {
-                _convertToEnumerable = CompileConvertToEnumerable(_convertToEnumerableMethodInfo, StreamReturnType);
+                _makeCancelableEnumerable = CompileConvertToEnumerable(_makeCancelableEnumerableMethodInfo, StreamReturnType);
             }
 
-            return _convertToEnumerable.Invoke(stream, cancellationToken);
+            return _makeCancelableEnumerable.Invoke(stream, cancellationToken);
         }
 
         private static Func<object, CancellationToken, IAsyncEnumerable<object>> CompileConvertToEnumerable(MethodInfo adapterMethodInfo, Type streamReturnType)
         {
             // This will call one of two adapter methods to wrap the passed in streamable value into an IAsyncEnumerable<object>:
-            // - AsyncEnumerableAdapters.GetAsyncEnumerableFromAsyncEnumerable<T>(asyncEnumerable, cancellationToken);
-            // - AsyncEnumerableAdapters.GetAsyncEnumerableFromChannel<T>(channelReader, cancellationToken);
+            // - AsyncEnumerableAdapters.MakeCancelableAsyncEnumerable<T>(asyncEnumerable, cancellationToken);
+            // - AsyncEnumerableAdapters.MakeCancelableAsyncEnumerableFromChannel<T>(channelReader, cancellationToken);
 
             var parameters = new[]
             {
diff --git a/src/SignalR/server/Core/src/Microsoft.AspNetCore.SignalR.Core.csproj b/src/SignalR/server/Core/src/Microsoft.AspNetCore.SignalR.Core.csproj
@@ -1,10 +1,11 @@
-<Project Sdk="Microsoft.NET.Sdk">
+﻿<Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
     <Description>Real-time communication framework for ASP.NET Core.</Description>
     <TargetFramework>netcoreapp3.0</TargetFramework>
     <IsAspNetCoreApp>true</IsAspNetCoreApp>
     <RootNamespace>Microsoft.AspNetCore.SignalR</RootNamespace>
+    <LangVersion>8.0</LangVersion>
   </PropertyGroup>
 
   <ItemGroup>
diff --git a/src/SignalR/server/SignalR/test/Microsoft.AspNetCore.SignalR.Tests.csproj b/src/SignalR/server/SignalR/test/Microsoft.AspNetCore.SignalR.Tests.csproj
@@ -2,6 +2,7 @@
 
   <PropertyGroup>
     <TargetFramework>netcoreapp3.0</TargetFramework>
+    <LangVersion>8.0</LangVersion>
   </PropertyGroup>
 
 

Original file line number	Diff line number	Diff line change
`@@ -397,7 +397,7 @@ private ValueTask CleanupInvocation(HubConnectionContext connection, HubMethodIn`
`397`	`397`	`return scope.DisposeAsync();`
`398`	`398`	`}`
`399`	`399`
`400`		`- private async Task StreamResultsAsync(string invocationId, HubConnectionContext connection, IAsyncEnumerable<object> enumerable, IServiceScope scope,`
	`400`	`+ private async Task StreamResultsAsync(string invocationId, HubConnectionContext connection, IAsyncEnumerable<object> enumerable, IServiceScope scope,`
`401`	`401`	`IHubActivator<THub> hubActivator, THub hub, CancellationTokenSource streamCts, HubMethodInvocationMessage hubMethodInvocationMessage)`
`402`	`402`	`{`
`403`	`403`	`string error = null;`
Original file line number	Diff line number	Diff line change
`@@ -19,12 +19,12 @@ internal class HubMethodDescriptor`
`19`	`19`	`.GetRuntimeMethods()`
`20`	`20`	`.Single(m => m.Name.Equals(nameof(AsyncEnumerableAdapters.MakeCancelableAsyncEnumerable)) && m.IsGenericMethod);`
`21`	`21`
`22`		`- private static readonly MethodInfo GetAsyncEnumerableFromChannelMethod = typeof(AsyncEnumerableAdapters)`
	`22`	`+ private static readonly MethodInfo MakeCancelableAsyncEnumerableFromChannelMethod = typeof(AsyncEnumerableAdapters)`
`23`	`23`	`.GetRuntimeMethods()`
`24`		`- .Single(m => m.Name.Equals(nameof(AsyncEnumerableAdapters.GetAsyncEnumerableFromChannel)) && m.IsGenericMethod);`
	`24`	`+ .Single(m => m.Name.Equals(nameof(AsyncEnumerableAdapters.MakeCancelableAsyncEnumerableFromChannel)) && m.IsGenericMethod);`
`25`	`25`
`26`		`- private readonly MethodInfo _convertToEnumerableMethodInfo;`
`27`		`- private Func<object, CancellationToken, IAsyncEnumerable<object>> _convertToEnumerable;`
	`26`	`+ private readonly MethodInfo _makeCancelableEnumerableMethodInfo;`
	`27`	`+ private Func<object, CancellationToken, IAsyncEnumerable<object>> _makeCancelableEnumerable;`
`28`	`28`
`29`	`29`	`public HubMethodDescriptor(ObjectMethodExecutor methodExecutor, IEnumerable<IAuthorizeData> policies)`
`30`	`30`	`{`
`@@ -46,14 +46,14 @@ public HubMethodDescriptor(ObjectMethodExecutor methodExecutor, IEnumerable<IAut`
`46`	`46`	`if (openReturnType == typeof(IAsyncEnumerable<>))`
`47`	`47`	`{`
`48`	`48`	`StreamReturnType = returnType.GetGenericArguments()[0];`
`49`		`- _convertToEnumerableMethodInfo = MakeCancelableAsyncEnumerableMethod;`
	`49`	`+ _makeCancelableEnumerableMethodInfo = MakeCancelableAsyncEnumerableMethod;`
`50`	`50`	`break;`
`51`	`51`	`}`
`52`	`52`
`53`	`53`	`if (openReturnType == typeof(ChannelReader<>))`
`54`	`54`	`{`
`55`	`55`	`StreamReturnType = returnType.GetGenericArguments()[0];`
`56`		`- _convertToEnumerableMethodInfo = GetAsyncEnumerableFromChannelMethod;`
	`56`	`+ _makeCancelableEnumerableMethodInfo = MakeCancelableAsyncEnumerableFromChannelMethod;`
`57`	`57`	`break;`
`58`	`58`	`}`
`59`	`59`	`}`
`@@ -110,19 +110,19 @@ public HubMethodDescriptor(ObjectMethodExecutor methodExecutor, IEnumerable<IAut`
`110`	`110`	`public IAsyncEnumerable<object> FromReturnedStream(object stream, CancellationToken cancellationToken)`
`111`	`111`	`{`
`112`	`112`	`// there is the potential for compile to be called times but this has no harmful effect other than perf`
`113`		`- if (_convertToEnumerable == null)`
	`113`	`+ if (_makeCancelableEnumerable == null)`
`114`	`114`	`{`
`115`		`- _convertToEnumerable = CompileConvertToEnumerable(_convertToEnumerableMethodInfo, StreamReturnType);`
	`115`	`+ _makeCancelableEnumerable = CompileConvertToEnumerable(_makeCancelableEnumerableMethodInfo, StreamReturnType);`
`116`	`116`	`}`
`117`	`117`
`118`		`- return _convertToEnumerable.Invoke(stream, cancellationToken);`
	`118`	`+ return _makeCancelableEnumerable.Invoke(stream, cancellationToken);`
`119`	`119`	`}`
`120`	`120`
`121`	`121`	`private static Func<object, CancellationToken, IAsyncEnumerable<object>> CompileConvertToEnumerable(MethodInfo adapterMethodInfo, Type streamReturnType)`
`122`	`122`	`{`
`123`	`123`	`// This will call one of two adapter methods to wrap the passed in streamable value into an IAsyncEnumerable<object>:`
`124`		`- // - AsyncEnumerableAdapters.GetAsyncEnumerableFromAsyncEnumerable<T>(asyncEnumerable, cancellationToken);`
`125`		`- // - AsyncEnumerableAdapters.GetAsyncEnumerableFromChannel<T>(channelReader, cancellationToken);`
	`124`	`+ // - AsyncEnumerableAdapters.MakeCancelableAsyncEnumerable<T>(asyncEnumerable, cancellationToken);`
	`125`	`+ // - AsyncEnumerableAdapters.MakeCancelableAsyncEnumerableFromChannel<T>(channelReader, cancellationToken);`
`126`	`126`
`127`	`127`	`var parameters = new[]`
`128`	`128`	`{`