Blazor WebAssembly & WebView JS to .NET Streaming Interop Support

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

@@ -40,22 +40,20 @@ public static async ValueTask<RemoteJSDataStream> CreateRemoteJSDataStreamAsync(
             RemoteJSRuntime runtime,
             IJSStreamReference jsStreamReference,
             long totalLength,
-            long maximumIncomingBytes,
+            long signalRMaximumIncomingBytes,
             TimeSpan jsInteropDefaultCallTimeout,
-            long pauseIncomingBytesThreshold = -1,
-            long resumeIncomingBytesThreshold = -1,
             CancellationToken cancellationToken = default)
         {
             // Enforce minimum 1 kb, maximum 50 kb, SignalR message size.
             // We budget 512 bytes overhead for the transfer, thus leaving at least 512 bytes for data
             // transfer per chunk with a 1 kb message size.
             // Additionally, to maintain interactivity, we put an upper limit of 50 kb on the message size.
-            var chunkSize = maximumIncomingBytes > 1024 ?
-                Math.Min(maximumIncomingBytes, 50*1024) - 512 :
+            var chunkSize = signalRMaximumIncomingBytes > 1024 ?
+                Math.Min(signalRMaximumIncomingBytes, 50*1024) - 512 :
                 throw new ArgumentException($"SignalR MaximumIncomingBytes must be at least 1 kb.");
 
             var streamId = runtime.RemoteJSDataStreamNextInstanceId++;
-            var remoteJSDataStream = new RemoteJSDataStream(runtime, streamId, totalLength, jsInteropDefaultCallTimeout, pauseIncomingBytesThreshold, resumeIncomingBytesThreshold, cancellationToken);
+            var remoteJSDataStream = new RemoteJSDataStream(runtime, streamId, totalLength, jsInteropDefaultCallTimeout, cancellationToken);
             await runtime.InvokeVoidAsync("Blazor._internal.sendJSDataStream", jsStreamReference, streamId, chunkSize);
             return remoteJSDataStream;
         }
@@ -65,8 +63,6 @@ private RemoteJSDataStream(
             long streamId,
             long totalLength,
             TimeSpan jsInteropDefaultCallTimeout,
-            long pauseIncomingBytesThreshold,
-            long resumeIncomingBytesThreshold,
             CancellationToken cancellationToken)
         {
             _runtime = runtime;
@@ -80,7 +76,7 @@ private RemoteJSDataStream(
 
             _runtime.RemoteJSDataStreamInstances.Add(_streamId, this);
 
-            _pipe = new Pipe(new PipeOptions(pauseWriterThreshold: pauseIncomingBytesThreshold, resumeWriterThreshold: resumeIncomingBytesThreshold));
+            _pipe = new Pipe();
             _pipeReaderStream = _pipe.Reader.AsStream();
             PipeReader = _pipe.Reader;
         }
@@ -94,9 +90,6 @@ private async Task<bool> ReceiveData(long chunkId, byte[] chunk, string error)
         {
             try
             {
-                _lastDataReceivedTime = DateTimeOffset.UtcNow;
-                _ = ThrowOnTimeout();
-
                 if (!string.IsNullOrEmpty(error))
                 {
                     throw new InvalidOperationException($"An error occurred while reading the remote stream: {error}");
@@ -111,7 +104,7 @@ private async Task<bool> ReceiveData(long chunkId, byte[] chunk, string error)
 
                 if (chunk.Length == 0)
                 {
-                    throw new EndOfStreamException($"The incoming data chunk cannot be empty.");
+                    throw new EndOfStreamException("The incoming data chunk cannot be empty.");
                 }
 
                 _bytesRead += chunk.Length;
@@ -121,6 +114,10 @@ private async Task<bool> ReceiveData(long chunkId, byte[] chunk, string error)
                     throw new EndOfStreamException($"The incoming data stream declared a length {_totalLength}, but {_bytesRead} bytes were sent.");
                 }
 
+                // Start timeout _after_ performing validations on data.
+                _lastDataReceivedTime = DateTimeOffset.UtcNow;
+                _ = ThrowOnTimeout();
+
                 await _pipe.Writer.WriteAsync(chunk, _streamCancellationToken);
 
                 if (_bytesRead == _totalLength)

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

@@ -157,8 +157,8 @@ public void MarkPermanentlyDisconnected()
             _clientProxy = null;
         }
 
-        protected override async Task<Stream> ReadJSDataAsStreamAsync(IJSStreamReference jsStreamReference, long totalLength, long pauseIncomingBytesThreshold = -1, long resumeIncomingBytesThreshold = -1, CancellationToken cancellationToken = default)
-            => await RemoteJSDataStream.CreateRemoteJSDataStreamAsync(this, jsStreamReference, totalLength, _maximumIncomingBytes, _options.JSInteropDefaultCallTimeout, pauseIncomingBytesThreshold, resumeIncomingBytesThreshold, cancellationToken);
+        protected override async Task<Stream> ReadJSDataAsStreamAsync(IJSStreamReference jsStreamReference, long totalLength, CancellationToken cancellationToken = default)
+            => await RemoteJSDataStream.CreateRemoteJSDataStreamAsync(this, jsStreamReference, totalLength, _maximumIncomingBytes, _options.JSInteropDefaultCallTimeout, cancellationToken);
 
         public static class Log
         {

src/Components/Server/src/Microsoft.AspNetCore.Components.Server.csproj

@@ -25,7 +25,7 @@
     <Reference Include="Microsoft.Extensions.Logging" />
     <!-- Required for S.T.J source generation -->
     <Reference Include="System.Text.Json" PrivateAssets="All" />
-    
+
     <Compile Include="$(SharedSourceRoot)ValueStopwatch\*.cs" />
 
     <Compile Include="$(SharedSourceRoot)LinkerFlags.cs" LinkBase="Shared" />

src/Components/Server/test/Circuits/RemoteJSDataStreamTest.cs

@@ -27,7 +27,7 @@ public async Task CreateRemoteJSDataStreamAsync_CreatesStream()
             var jsStreamReference = Mock.Of<IJSStreamReference>();
 
             // Act
-            var remoteJSDataStream = await RemoteJSDataStream.CreateRemoteJSDataStreamAsync(_jsRuntime, jsStreamReference, totalLength: 100, maximumIncomingBytes: 10_000, jsInteropDefaultCallTimeout: TimeSpan.FromMinutes(1), pauseIncomingBytesThreshold: 50, resumeIncomingBytesThreshold: 25, cancellationToken: CancellationToken.None).DefaultTimeout();
+            var remoteJSDataStream = await RemoteJSDataStream.CreateRemoteJSDataStreamAsync(_jsRuntime, jsStreamReference, totalLength: 100, signalRMaximumIncomingBytes: 10_000, jsInteropDefaultCallTimeout: TimeSpan.FromMinutes(1), cancellationToken: CancellationToken.None).DefaultTimeout();
 
             // Assert
             Assert.NotNull(remoteJSDataStream);
@@ -146,7 +146,7 @@ public async Task ReceiveData_ProvidedWithMoreBytesThanRemaining()
             // Arrange
             var jsRuntime = new TestRemoteJSRuntime(Options.Create(new CircuitOptions()), Options.Create(new HubOptions()), Mock.Of<ILogger<RemoteJSRuntime>>());
             var jsStreamReference = Mock.Of<IJSStreamReference>();
-            var remoteJSDataStream = await RemoteJSDataStream.CreateRemoteJSDataStreamAsync(jsRuntime, jsStreamReference, totalLength: 100, maximumIncomingBytes: 10_000, jsInteropDefaultCallTimeout: TimeSpan.FromMinutes(1), pauseIncomingBytesThreshold: 50, resumeIncomingBytesThreshold: 25, cancellationToken: CancellationToken.None);
+            var remoteJSDataStream = await RemoteJSDataStream.CreateRemoteJSDataStreamAsync(jsRuntime, jsStreamReference, totalLength: 100, signalRMaximumIncomingBytes: 10_000, jsInteropDefaultCallTimeout: TimeSpan.FromMinutes(1), cancellationToken: CancellationToken.None);
             var streamId = GetStreamId(remoteJSDataStream, jsRuntime);
             var chunk = new byte[110]; // 100 byte totalLength for stream
 
@@ -166,7 +166,7 @@ public async Task ReceiveData_ProvidedWithOutOfOrderChunk_SimulatesSignalRDiscon
             // Arrange
             var jsRuntime = new TestRemoteJSRuntime(Options.Create(new CircuitOptions()), Options.Create(new HubOptions()), Mock.Of<ILogger<RemoteJSRuntime>>());
             var jsStreamReference = Mock.Of<IJSStreamReference>();
-            var remoteJSDataStream = await RemoteJSDataStream.CreateRemoteJSDataStreamAsync(jsRuntime, jsStreamReference, totalLength: 100, maximumIncomingBytes: 10_000, jsInteropDefaultCallTimeout: TimeSpan.FromMinutes(1), pauseIncomingBytesThreshold: 50, resumeIncomingBytesThreshold: 25, cancellationToken: CancellationToken.None);
+            var remoteJSDataStream = await RemoteJSDataStream.CreateRemoteJSDataStreamAsync(jsRuntime, jsStreamReference, totalLength: 100, signalRMaximumIncomingBytes: 10_000, jsInteropDefaultCallTimeout: TimeSpan.FromMinutes(1), cancellationToken: CancellationToken.None);
             var streamId = GetStreamId(remoteJSDataStream, jsRuntime);
             var chunk = new byte[5];
 
@@ -201,10 +201,8 @@ public async Task ReceiveData_NoDataProvidedBeforeTimeout_StreamDisposed()
                 jsRuntime,
                 jsStreamReference,
                 totalLength: 15,
-                maximumIncomingBytes: 10_000,
+                signalRMaximumIncomingBytes: 10_000,
                 jsInteropDefaultCallTimeout: TimeSpan.FromSeconds(2),
-                pauseIncomingBytesThreshold: 50,
-                resumeIncomingBytesThreshold: 25,
                 cancellationToken: CancellationToken.None);
             var streamId = GetStreamId(remoteJSDataStream, jsRuntime);
             var chunk = new byte[] { 3, 5, 7 };
@@ -244,10 +242,8 @@ public async Task ReceiveData_ReceivesDataThenTimesout_StreamDisposed()
                 jsRuntime,
                 jsStreamReference,
                 totalLength: 15,
-                maximumIncomingBytes: 10_000,
+                signalRMaximumIncomingBytes: 10_000,
                 jsInteropDefaultCallTimeout: TimeSpan.FromSeconds(3),
-                pauseIncomingBytesThreshold: 50,
-                resumeIncomingBytesThreshold: 25,
                 cancellationToken: CancellationToken.None);
             var streamId = GetStreamId(remoteJSDataStream, jsRuntime);
             var chunk = new byte[] { 3, 5, 7 };
@@ -281,7 +277,7 @@ public async Task ReceiveData_ReceivesDataThenTimesout_StreamDisposed()
         private static async Task<RemoteJSDataStream> CreateRemoteJSDataStreamAsync(TestRemoteJSRuntime jsRuntime = null)
         {
             var jsStreamReference = Mock.Of<IJSStreamReference>();
-            var remoteJSDataStream = await RemoteJSDataStream.CreateRemoteJSDataStreamAsync(jsRuntime ?? _jsRuntime, jsStreamReference, totalLength: 100, maximumIncomingBytes: 10_000, jsInteropDefaultCallTimeout: TimeSpan.FromMinutes(1), pauseIncomingBytesThreshold: 50, resumeIncomingBytesThreshold: 25, cancellationToken: CancellationToken.None);
+            var remoteJSDataStream = await RemoteJSDataStream.CreateRemoteJSDataStreamAsync(jsRuntime ?? _jsRuntime, jsStreamReference, totalLength: 100, signalRMaximumIncomingBytes: 10_000, jsInteropDefaultCallTimeout: TimeSpan.FromMinutes(1), cancellationToken: CancellationToken.None);
             return remoteJSDataStream;
         }
 

src/Components/Shared/src/PullFromJSDataStream.cs

@@ -0,0 +1,113 @@
+// 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.Threading;
+using System.Threading.Tasks;
+using Microsoft.JSInterop;
+
+namespace Microsoft.AspNetCore.Components
+{
+    /// <Summary>
+    /// A stream that pulls each chunk on demand using JavaScript interop. This implementation is used for
+    /// WebAssembly and WebView applications.
+    /// </Summary>
+    internal sealed class PullFromJSDataStream : Stream
+    {
+        private readonly IJSRuntime _runtime;
+        private readonly IJSStreamReference _jsStreamReference;
+        private readonly long _totalLength;
+        private readonly CancellationToken _streamCancellationToken;
+        private long _offset;
+
+        public static PullFromJSDataStream CreateJSDataStream(
+            IJSRuntime runtime,
+            IJSStreamReference jsStreamReference,
+            long totalLength,
+            CancellationToken cancellationToken = default)
+        {
+            var jsDataStream = new PullFromJSDataStream(runtime, jsStreamReference, totalLength, cancellationToken);
+            return jsDataStream;
+        }
+
+        private PullFromJSDataStream(
+            IJSRuntime runtime,
+            IJSStreamReference jsStreamReference,
+            long totalLength,
+            CancellationToken cancellationToken)
+        {
+            _runtime = runtime;
+            _jsStreamReference = jsStreamReference;
+            _totalLength = totalLength;
+            _streamCancellationToken = cancellationToken;
+            _offset = 0;
+        }
+
+        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 => _offset;
+            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)
+            => await ReadAsync(buffer.AsMemory(offset, count), cancellationToken);
+
+        public override async ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
+        {
+            var bytesRead = await RequestDataFromJSAsync(buffer.Length);
+            ThrowIfCancellationRequested(cancellationToken);
+            bytesRead.CopyTo(buffer);
+
+            return bytesRead.Length;
+        }
+
+        private void ThrowIfCancellationRequested(CancellationToken cancellationToken)
+        {
+            if (cancellationToken.IsCancellationRequested ||
+                _streamCancellationToken.IsCancellationRequested)
+            {
+                throw new TaskCanceledException();
+            }
+        }
+
+        private async ValueTask<byte[]> RequestDataFromJSAsync(int numBytesToRead)
+        {
+            numBytesToRead = (int)Math.Min(numBytesToRead, _totalLength - _offset);
+            var bytesRead = await _runtime.InvokeAsync<byte[]>("Blazor._internal.getJSDataStreamChunk", _jsStreamReference, _offset, numBytesToRead);
+            if (bytesRead.Length != numBytesToRead)
+            {
+                throw new EndOfStreamException("Failed to read the requested number of bytes from the stream.");
+            }
+
+            _offset += bytesRead.Length;
+            if (_offset == _totalLength)
+            {
+                Dispose(true);
+            }
+            return bytesRead;
+        }
+    }
+}

src/Components/Web.JS/src/Boot.Server.ts

@@ -13,7 +13,6 @@ import { resolveOptions, CircuitStartOptions } from './Platform/Circuits/Circuit
 import { DefaultReconnectionHandler } from './Platform/Circuits/DefaultReconnectionHandler';
 import { attachRootComponentToLogicalElement } from './Rendering/Renderer';
 import { discoverComponents, discoverPersistedState, ServerComponentDescriptor } from './Services/ComponentDescriptorDiscovery';
-import { InputFile } from './InputFile';
 import { sendJSDataStream } from './Platform/Circuits/CircuitStreamingInterop';
 
 let renderingFailed = false;
@@ -29,7 +28,6 @@ async function boot(userOptions?: Partial<CircuitStartOptions>): Promise<void> {
   const options = resolveOptions(userOptions);
   const logger = new ConsoleLogger(options.logLevel);
   Blazor.defaultReconnectionHandler = new DefaultReconnectionHandler(logger);
-  Blazor._internal.InputFile = InputFile;
 
   options.reconnectionHandler = options.reconnectionHandler || Blazor.defaultReconnectionHandler;
   logger.log(LogLevel.Information, 'Starting up Blazor server-side application.');
@@ -126,7 +124,7 @@ async function initializeConnection(options: CircuitStartOptions, logger: Logger
 
   Blazor._internal.forceCloseConnection = () => connection.stop();
 
-  Blazor._internal.sendJSDataStream = (data: ArrayBufferView, streamId: string, chunkSize: number) => sendJSDataStream(connection, data, streamId, chunkSize);
+  Blazor._internal.sendJSDataStream = (data: ArrayBufferView | Blob, streamId: number, chunkSize: number) => sendJSDataStream(connection, data, streamId, chunkSize);
 
   try {
     await connection.start();

src/Components/Web.JS/src/Boot.WebAssembly.ts

@@ -14,7 +14,6 @@ import { Pointer, System_Array, System_Boolean, System_Byte, System_Int, System_
 import { WebAssemblyStartOptions } from './Platform/WebAssemblyStartOptions';
 import { WebAssemblyComponentAttacher } from './Platform/WebAssemblyComponentAttacher';
 import { discoverComponents, discoverPersistedState, WebAssemblyComponentDescriptor } from './Services/ComponentDescriptorDiscovery';
-import { WasmInputFile } from './WasmInputFile';
 
 declare var Module: EmscriptenModule;
 let started = false;
@@ -43,8 +42,6 @@ async function boot(options?: Partial<WebAssemblyStartOptions>): Promise<void> {
     }
   });
 
-  Blazor._internal.InputFile = WasmInputFile;
-
   Blazor._internal.applyHotReload = (id: string, metadataDelta: string, ilDeta: string) => {
     DotNet.invokeMethod('Microsoft.AspNetCore.Components.WebAssembly', 'ApplyHotReloadDelta', id, metadataDelta, ilDeta);
   };
@@ -160,6 +157,9 @@ function invokeJSFromDotNet(callInfo: Pointer, arg0: any, arg1: any, arg2: any):
       case DotNet.JSCallResultType.JSObjectReference:
         return DotNet.createJSObjectReference(result).__jsObjectId;
       case DotNet.JSCallResultType.JSStreamReference:
+        const streamReference = DotNet.createJSStreamReference(result);
+        const resultJson = JSON.stringify(streamReference);
+        return BINDING.js_string_to_mono_string(resultJson);
       default:
         throw new Error(`Invalid JS call result type '${resultType}'.`);
     }

src/Components/Web.JS/src/Boot.WebView.ts

@@ -5,7 +5,6 @@ import { internalFunctions as navigationManagerFunctions } from './Services/Navi
 import { setEventDispatcher } from './Rendering/Events/EventDispatcher';
 import { startIpcReceiver } from './Platform/WebView/WebViewIpcReceiver';
 import { sendBrowserEvent, sendAttachPage, sendBeginInvokeDotNetFromJS, sendEndInvokeJSFromDotNet, sendByteArray, sendLocationChanged } from './Platform/WebView/WebViewIpcSender';
-import { InputFile } from './InputFile';
 
 let started = false;
 
@@ -23,7 +22,6 @@ async function boot(): Promise<void> {
     sendByteArray: sendByteArray,
   });
 
-  Blazor._internal.InputFile = InputFile;
   navigationManagerFunctions.enableNavigationInterception();
   navigationManagerFunctions.listenForNavigationEvents(sendLocationChanged);
 

src/Components/Web.JS/src/GlobalExports.ts

@@ -8,6 +8,7 @@ import { DefaultReconnectionHandler } from './Platform/Circuits/DefaultReconnect
 import { CircuitStartOptions } from './Platform/Circuits/CircuitStartOptions';
 import { WebAssemblyStartOptions } from './Platform/WebAssemblyStartOptions';
 import { Platform, Pointer, System_String, System_Array, System_Object, System_Boolean, System_Byte, System_Int } from './Platform/Platform';
+import { getNextChunk } from './StreamingInterop';
 
 interface IBlazor {
   navigateTo: (uri: string, options: NavigationOptions) => void;
@@ -48,7 +49,8 @@ interface IBlazor {
     getLazyAssemblies?: any
     dotNetCriticalError?: any
     getSatelliteAssemblies?: any,
-    sendJSDataStream?: (data: any, streamId: string, chunkSize: number) => void,
+    sendJSDataStream?: (data: any, streamId: number, chunkSize: number) => void,
+    getJSDataStreamChunk?: (data: any, position: number, chunkSize: number) => Promise<Uint8Array>,
 
     // APIs invoked by hot reload
     applyHotReload?: (id: string, metadataDelta: string, ilDelta: string) => void,
@@ -64,6 +66,8 @@ export const Blazor: IBlazor = {
     navigationManager: navigationManagerInternalFunctions,
     domWrapper: domFunctions,
     Virtualize,
+    InputFile,
+    getJSDataStreamChunk: getNextChunk,
   },
 };