Blazor Streaming Interop | JS to DotNet

src/Components/Server/src/Circuits/CircuitHost.cs

@@ -2,6 +2,7 @@
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
 using System;
+using System.Buffers;
 using System.Collections.Generic;
 using System.Globalization;
 using System.Security.Claims;
@@ -415,6 +416,28 @@ await Renderer.Dispatcher.InvokeAsync(() =>
             }
         }
 
+        // ReceiveJSDataChunk is used in a fire-and-forget context, so it's responsible for its own
+        // error handling.
+        internal async Task<bool> ReceiveJSDataChunk(long streamId, byte[] chunk, string error)
+        {
+            AssertInitialized();
+            AssertNotDisposed();
+
+            try
+            {
+                return await RemoteJSDataStream.ReceiveData(JSRuntime, streamId, chunk, error);
+            }
+            catch (Exception ex)
+            {
+                // An error completing JS interop means that the user sent invalid data, a well-behaved
+                // client won't do this.
+                Log.ReceiveJSDataChunkException(_logger, streamId, ex);
+                await TryNotifyClientErrorAsync(Client, GetClientErrorMessage(ex, "Invalid chunk supplied to stream."));
+                UnhandledException?.Invoke(this, new UnhandledExceptionEventArgs(ex, isTerminating: false));
+                return false;
+            }
+        }
+
         // DispatchEvent is used in a fire-and-forget context, so it's responsible for its own
         // error handling.
         public async Task DispatchEvent(string eventDescriptorJson, string eventArgsJson)
@@ -638,6 +661,7 @@ private static class Log
             private static readonly Action<ILogger, long, Exception> _endInvokeJSSucceeded;
             private static readonly Action<ILogger, long, Exception> _receiveByteArraySuccess;
             private static readonly Action<ILogger, long, Exception> _receiveByteArrayException;
+            private static readonly Action<ILogger, long, Exception> _receiveJSDataChunkException;
             private static readonly Action<ILogger, Exception> _dispatchEventFailedToParseEventData;
             private static readonly Action<ILogger, string, Exception> _dispatchEventFailedToDispatchEvent;
             private static readonly Action<ILogger, string, CircuitId, Exception> _locationChange;
@@ -682,6 +706,7 @@ private static class EventIds
                 public static readonly EventId OnRenderCompletedFailed = new EventId(212, "OnRenderCompletedFailed");
                 public static readonly EventId ReceiveByteArraySucceeded = new EventId(213, "ReceiveByteArraySucceeded");
                 public static readonly EventId ReceiveByteArrayException = new EventId(214, "ReceiveByteArrayException");
+                public static readonly EventId ReceiveJSDataChunkException = new EventId(215, "ReceiveJSDataChunkException");
             }
 
             static Log()
@@ -811,6 +836,11 @@ static Log()
                     EventIds.ReceiveByteArrayException,
                     "The ReceiveByteArray call with id '{id}' failed.");
 
+                _receiveJSDataChunkException = LoggerMessage.Define<long>(
+                    LogLevel.Debug,
+                    EventIds.ReceiveJSDataChunkException,
+                    "The ReceiveJSDataChunk call with stream id '{streamId}' failed.");
+
                 _dispatchEventFailedToParseEventData = LoggerMessage.Define(
                     LogLevel.Debug,
                     EventIds.DispatchEventFailedToParseEventData,
@@ -875,6 +905,7 @@ public static void CircuitHandlerFailed(ILogger logger, CircuitHandler handler,
             public static void EndInvokeJSSucceeded(ILogger logger, long asyncCall) => _endInvokeJSSucceeded(logger, asyncCall, null);
             internal static void ReceiveByteArraySuccess(ILogger logger, long id) => _receiveByteArraySuccess(logger, id, null);
             internal static void ReceiveByteArrayException(ILogger logger, long id, Exception ex) => _receiveByteArrayException(logger, id, ex);
+            internal static void ReceiveJSDataChunkException(ILogger logger, long streamId, Exception ex) => _receiveJSDataChunkException(logger, streamId, ex);
             public static void DispatchEventFailedToParseEventData(ILogger logger, Exception ex) => _dispatchEventFailedToParseEventData(logger, ex);
             public static void DispatchEventFailedToDispatchEvent(ILogger logger, string eventHandlerId, Exception ex) => _dispatchEventFailedToDispatchEvent(logger, eventHandlerId ?? "", ex);
 

src/Components/Server/src/Circuits/RemoteJSDataStream.cs

@@ -0,0 +1,173 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
+
+using System;
+using System.Buffers;
+using System.IO;
+using System.IO.Pipelines;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.JSInterop;
+
+namespace Microsoft.AspNetCore.Components.Server.Circuits
+{
+    internal sealed class RemoteJSDataStream : Stream
+    {
+        private readonly RemoteJSRuntime _runtime;
+        private readonly long _streamId;
+        private readonly long _totalLength;
+        private readonly CancellationToken _streamCancellationToken;
+        private readonly Stream _pipeReaderStream;
+        private readonly Pipe _pipe;
+        private long _bytesRead;
+
+        public static async Task<bool> ReceiveData(RemoteJSRuntime runtime, long streamId, byte[] chunk, string error)
+        {
+            if (!runtime.RemoteJSDataStreamInstances.TryGetValue(streamId, out var instance))
+            {
+                // There is no data stream with the given identifier. It may have already been disposed.
+                // We notify JS that the stream has been cancelled/disposed.
+                return false;
+            }
+
+            return await instance.ReceiveData(chunk, error);
+        }
+
+        public static async ValueTask<RemoteJSDataStream> CreateRemoteJSDataStreamAsync(
+            RemoteJSRuntime runtime,
+            IJSDataReference jsDataReference,
+            long totalLength,
+            long maxBufferSize,
+            long maximumIncomingBytes,
+            CancellationToken cancellationToken = default)
+        {
+            var streamId = runtime.RemoteJSDataStreamNextInstanceId++;
+            var remoteJSDataStream = new RemoteJSDataStream(runtime, streamId, totalLength, maxBufferSize, cancellationToken);
+            await runtime.InvokeVoidAsync("Blazor._internal.sendJSDataStream", jsDataReference, streamId, maximumIncomingBytes);
+            return remoteJSDataStream;
+        }
+
+        private RemoteJSDataStream(
+            RemoteJSRuntime runtime,
+            long streamId,
+            long totalLength,
+            long maxBufferSize,
+            CancellationToken cancellationToken)
+        {
+            _runtime = runtime;
+            _streamId = streamId;
+            _totalLength = totalLength;
+            _streamCancellationToken = cancellationToken;
+
+            _runtime.RemoteJSDataStreamInstances.Add(_streamId, this);
+
+            _pipe = new Pipe(new PipeOptions(pauseWriterThreshold: maxBufferSize, resumeWriterThreshold: maxBufferSize / 2));
+            _pipeReaderStream = _pipe.Reader.AsStream();
+        }
+
+        // Ideally we'd have IAsyncEnumerable<ReadOnlySequence<byte>> here so we can pass through the
+        // data without having to copy it into a temporary buffer in BlazorPackHubProtocolWorker. But trying
+        // this gives strange errors; sometimes the "chunk" variable below has a negative length, even though
+        // the logic never returns a corrupted item as far as I can tell.
+        private async Task<bool> ReceiveData(byte[] chunk, string error)
+        {
+            try
+            {
+                if (!string.IsNullOrEmpty(error))
+                {
+                    throw new InvalidOperationException($"An error occurred while reading the remote stream: {error}");
+                }
+
+                if (chunk.Length == 0)
+                {
+                    throw new InvalidOperationException($"The incoming data chunk cannot be empty.");
+                }
+
+                _bytesRead += chunk.Length;
+
+                if (_bytesRead > _totalLength)
+                {
+                    throw new InvalidOperationException($"The incoming data stream declared a length {_totalLength}, but {_bytesRead} bytes were read.");
+                }
+
+                await _pipe.Writer.WriteAsync(chunk, _streamCancellationToken);
+
+                if (_bytesRead == _totalLength)
+                {
+                    await _pipe.Writer.CompleteAsync();
+                }
+
+                return true;
+            }
+            catch (Exception e)
+            {
+                await _pipe.Writer.CompleteAsync(e);
+                Dispose(true);
+                return false;
+            }
+        }
+
+        public override bool CanRead => true;
+
+        public override bool CanSeek => false;
+
+        public override bool CanWrite => false;
+
+        public override long Length => _totalLength;
+
+        public override long Position
+        {
+            get => _pipeReaderStream.Position;
+            set => throw new NotSupportedException();
+        }
+
+        public override void Flush()
+            => throw new NotSupportedException();
+
+        public override int Read(byte[] buffer, int offset, int count)
+            => throw new NotSupportedException("Synchronous reads are not supported.");
+
+        public override long Seek(long offset, SeekOrigin origin)
+            => throw new NotSupportedException();
+
+        public override void SetLength(long value)
+            => throw new NotSupportedException();
+
+        public override void Write(byte[] buffer, int offset, int count)
+            => throw new NotSupportedException();
+
+        public override async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+        {
+            var linkedCancellationToken = GetLinkedCancellationToken(_streamCancellationToken, cancellationToken);
+            return await _pipeReaderStream.ReadAsync(buffer.AsMemory(offset, count), linkedCancellationToken);
+        }
+
+        public override async ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
+        {
+            var linkedCancellationToken = GetLinkedCancellationToken(_streamCancellationToken, cancellationToken);
+            return await _pipeReaderStream.ReadAsync(buffer, linkedCancellationToken);
+        }
+
+        private static CancellationToken GetLinkedCancellationToken(CancellationToken a, CancellationToken b)
+        {
+            if (a.CanBeCanceled && b.CanBeCanceled)
+            {
+                return CancellationTokenSource.CreateLinkedTokenSource(a, b).Token;
+            }
+            else if (a.CanBeCanceled)
+            {
+                return a;
+            }
+
+            return b;
+        }
+
+        protected override void Dispose(bool disposing)
+        {
+            if (disposing)
+            {
+                _runtime.RemoteJSDataStreamInstances.Remove(_streamId);
+            }
+        }
+    }
+}

src/Components/Server/src/Circuits/RemoteJSRuntime.cs

@@ -2,7 +2,11 @@
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
 using System;
+using System.Collections.Generic;
+using System.IO;
 using System.Text.Json;
+using System.Threading;
+using System.Threading.Tasks;
 using Microsoft.AspNetCore.SignalR;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
@@ -20,6 +24,9 @@ internal class RemoteJSRuntime : JSRuntime
         private readonly long _maximumIncomingBytes;
         private int _byteArraysToBeRevivedTotalBytes;
 
+        internal int RemoteJSDataStreamNextInstanceId;
+        internal readonly Dictionary<long, RemoteJSDataStream> RemoteJSDataStreamInstances = new();
+
         public ElementReferenceContext ElementReferenceContext { get; }
 
         public bool IsInitialized => _clientProxy is not null;
@@ -140,6 +147,9 @@ public void MarkPermanentlyDisconnected()
             _clientProxy = null;
         }
 
+        protected override async Task<Stream> ReadJSDataAsStreamAsync(IJSDataReference jsDataReference, long totalLength, long maxBufferSize, CancellationToken cancellationToken)
+            => await RemoteJSDataStream.CreateRemoteJSDataStreamAsync(this, jsDataReference, totalLength, maxBufferSize, _maximumIncomingBytes, cancellationToken);
+
         public static class Log
         {
             private static readonly Action<ILogger, long, string, Exception> _beginInvokeJS =
@@ -198,7 +208,6 @@ internal static void InvokeDotNetMethodSuccess(ILogger<RemoteJSRuntime> logger,
                 {
                     _invokeInstanceDotNetMethodSuccess(logger, invocationInfo.MethodIdentifier, invocationInfo.DotNetObjectId, invocationInfo.CallId, null);
                 }
-
             }
         }
     }

src/Components/Server/src/ComponentHub.cs

@@ -2,6 +2,7 @@
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
 using System;
+using System.Buffers;
 using System.Runtime.CompilerServices;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Components.Server.Circuits;
@@ -36,7 +37,7 @@ namespace Microsoft.AspNetCore.Components.Server
     // in error cases.
     internal sealed class ComponentHub : Hub
     {
-        private static readonly object CircuitKey = new object();
+        private static readonly object CircuitKey = new();
         private readonly IServerComponentDeserializer _serverComponentSerializer;
         private readonly IDataProtectionProvider _dataProtectionProvider;
         private readonly ICircuitFactory _circuitFactory;
@@ -220,7 +221,18 @@ public async ValueTask ReceiveByteArray(int id, byte[] data)
                 return;
             }
 
-            await circuitHost.ReceiveByteArray(id, data);
+            _ = circuitHost.ReceiveByteArray(id, data);
+        }
+
+        public async ValueTask<bool> ReceiveJSDataChunk(long streamId, byte[] chunk, string error)
+        {
+            var circuitHost = await GetActiveCircuitAsync();
+            if (circuitHost == null)
+            {
+                return false;
+            }
+
+            return await circuitHost.ReceiveJSDataChunk(streamId, chunk, error);
         }
 
         public async ValueTask DispatchBrowserEvent(string eventDescriptor, string eventArgs)