Revert "Remove CancelableEnumerator (#10099)" (#10129)

Author: mikaelm12

src/SignalR/common/Shared/AsyncEnumerableAdapters.cs

@@ -51,10 +51,37 @@ public IAsyncEnumerator<TResult> GetAsyncEnumerator(CancellationToken cancellati
                     {
                         ((CancellationTokenSource)ctsState).Cancel();
                     }, _cts);
+
+                    return new CancelableEnumerator<TResult>(_asyncEnumerable.GetAsyncEnumerator(), registration);
                 }
 
                 return enumerator;
             }
+
+            private class CancelableEnumerator<T> : IAsyncEnumerator<T>
+            {
+                private IAsyncEnumerator<T> _asyncEnumerator;
+                private readonly CancellationTokenRegistration _cancellationTokenRegistration;
+
+                public T Current => (T)_asyncEnumerator.Current;
+
+                public CancelableEnumerator(IAsyncEnumerator<T> asyncEnumerator, CancellationTokenRegistration registration)
+                {
+                    _asyncEnumerator = asyncEnumerator;
+                    _cancellationTokenRegistration = registration;
+                }
+
+                public ValueTask<bool> MoveNextAsync()
+                {
+                    return _asyncEnumerator.MoveNextAsync();
+                }
+
+                public ValueTask DisposeAsync()
+                {
+                    _cancellationTokenRegistration.Dispose();
+                    return _asyncEnumerator.DisposeAsync();
+                }
+            }
         }
 
         /// <summary>Converts an IAsyncEnumerable of T to an IAsyncEnumerable of object.</summary>
@@ -71,6 +98,10 @@ 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);
+
                 var enumeratorOfT = _asyncEnumerable.GetAsyncEnumerator(_cancellationToken);
                 return enumeratorOfT as IAsyncEnumerator<object> ?? new BoxedAsyncEnumerator(enumeratorOfT);
             }

src/SignalR/server/Core/src/Internal/DefaultHubDispatcher.cs

@@ -403,12 +403,13 @@ private async Task StreamResultsAsync(string invocationId, HubConnectionContext
             IHubActivator<THub> hubActivator, THub hub, CancellationTokenSource streamCts, HubMethodInvocationMessage hubMethodInvocationMessage)
         {
             string error = null;
+
             try
             {
-                await foreach(var item in enumerable.WithCancellation(streamCts.Token))
+                await foreach (var streamItem in enumerable)
                 {
                     // Send the stream item
-                    await connection.WriteAsync(new StreamItemMessage(invocationId, item));
+                    await connection.WriteAsync(new StreamItemMessage(invocationId, streamItem));
                 }
             }
             catch (ChannelClosedException ex)