Blazor WebAssembly & WebView JS to .NET Streaming Interop Support

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

@@ -14,8 +14,8 @@ internal sealed class RemoteJSDataStream : Stream
     {
         private readonly RemoteJSRuntime _runtime;
         private readonly long _streamId;
-        private readonly long _totalLength;
         private readonly TimeSpan _jsInteropDefaultCallTimeout;
+        private readonly long _totalLength;
         private readonly CancellationToken _streamCancellationToken;
         private readonly Stream _pipeReaderStream;
         private readonly Pipe _pipe;
@@ -71,8 +71,8 @@ private RemoteJSDataStream(
         {
             _runtime = runtime;
             _streamId = streamId;
-            _totalLength = totalLength;
             _jsInteropDefaultCallTimeout = jsInteropDefaultCallTimeout;
+            _totalLength = totalLength;
             _streamCancellationToken = cancellationToken;
 
             _lastDataReceivedTime = DateTimeOffset.UtcNow;
@@ -94,9 +94,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 +108,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 +118,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/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

@@ -286,7 +286,7 @@ private static async Task<RemoteJSDataStream> CreateRemoteJSDataStreamAsync(Test
         }
 
         private static long GetStreamId(RemoteJSDataStream stream, RemoteJSRuntime runtime) =>
-            runtime.RemoteJSDataStreamInstances.FirstOrDefault(kvp => kvp.Value == stream).Key;
+            runtime.JSDataStreamInstances.FirstOrDefault(kvp => kvp.Value == stream).Key;
 
         class TestRemoteJSRuntime : RemoteJSRuntime, IJSRuntime
         {

src/Components/Shared/src/PullFromJSDataStream.cs

@@ -0,0 +1,99 @@
+// 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 JSRuntime _runtime;
+        private readonly IJSStreamReference _jsStreamReference;
+        private readonly long _totalLength;
+        private readonly CancellationToken _streamCancellationToken;
+        private long _offset;
+
+        public static PullFromJSDataStream CreateJSDataStream(
+            JSRuntime runtime,
+            IJSStreamReference jsStreamReference,
+            long totalLength,
+            CancellationToken cancellationToken = default)
+        {
+            var jsDataStream = new PullFromJSDataStream(runtime, jsStreamReference, totalLength, cancellationToken);
+            return jsDataStream;
+        }
+
+        private PullFromJSDataStream(
+            JSRuntime 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);
+            bytesRead.CopyTo(buffer);
+
+            return bytesRead.Length;
+        }
+
+        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;
+            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,7 @@ 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';
+import { getNextChunk } from './StreamingInterop';
 
 declare var Module: EmscriptenModule;
 let started = false;
@@ -43,8 +43,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 +158,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,7 @@ 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';
+import { getNextChunk } from './StreamingInterop';
 
 let started = false;
 
@@ -23,7 +23,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,
   },
 };
 

src/Components/Web.JS/src/InputFile.ts

@@ -1,7 +1,6 @@
 export const InputFile = {
   init,
   toImageFile,
-  ensureArrayBufferReadyForSharedMemoryInterop,
   readFileData,
 };
 
@@ -11,9 +10,8 @@ interface BrowserFile {
   name: string;
   size: number;
   contentType: string;
-  readPromise: Promise<ArrayBuffer> | undefined;
-  arrayBuffer: ArrayBuffer | undefined;
   blob: Blob;
+  chunk: Uint8Array | undefined;
 }
 
 export interface InputElement extends HTMLInputElement {
@@ -43,6 +41,7 @@ function init(callbackWrapper: any, elem: InputElement): void {
         readPromise: undefined,
         arrayBuffer: undefined,
         blob: file,
+        chunk: undefined,
       };
 
       elem._blazorFilesById[result.id] = result;
@@ -83,21 +82,15 @@ async function toImageFile(elem: InputElement, fileId: number, format: string, m
     name: originalFile.name,
     size: resizedImageBlob?.size || 0,
     contentType: format,
-    readPromise: undefined,
-    arrayBuffer: undefined,
     blob: resizedImageBlob ? resizedImageBlob : originalFile.blob,
+    chunk: undefined,
   };
 
   elem._blazorFilesById[result.id] = result;
 
   return result;
 }
 
-async function ensureArrayBufferReadyForSharedMemoryInterop(elem: InputElement, fileId: number): Promise<void> {
-  const arrayBuffer = await getArrayBufferFromFileAsync(elem, fileId);
-  getFileById(elem, fileId).arrayBuffer = arrayBuffer;
-}
-
 async function readFileData(elem: InputElement, fileId: number): Promise<Blob> {
   const file = getFileById(elem, fileId);
   return file.blob;
@@ -112,23 +105,3 @@ export function getFileById(elem: InputElement, fileId: number): BrowserFile {
 
   return file;
 }
-
-function getArrayBufferFromFileAsync(elem: InputElement, fileId: number): Promise<ArrayBuffer> {
-  const file = getFileById(elem, fileId);
-
-  // On the first read, convert the FileReader into a Promise<ArrayBuffer>.
-  if (!file.readPromise) {
-    file.readPromise = new Promise(function(resolve: (buffer: ArrayBuffer) => void, reject): void {
-      const reader = new FileReader();
-      reader.onload = function(): void {
-        resolve(reader.result as ArrayBuffer);
-      };
-      reader.onerror = function(err): void {
-        reject(err);
-      };
-      reader.readAsArrayBuffer(file.blob);
-    });
-  }
-
-  return file.readPromise;
-}

src/Components/Web.JS/src/Platform/Circuits/CircuitStreamingInterop.ts

@@ -1,6 +1,7 @@
 import { HubConnection } from '@microsoft/signalr';
+import { getNextChunk } from '../../StreamingInterop';
 
-export function sendJSDataStream(connection: HubConnection, data: ArrayBufferView | Blob, streamId: string, chunkSize: number) {
+export function sendJSDataStream(connection: HubConnection, data: ArrayBufferView | Blob, streamId: number, chunkSize: number) {
     // Run the rest in the background, without delaying the completion of the call to sendJSDataStream
     // otherwise we'll deadlock (.NET can't begin reading until this completes, but it won't complete
     // because nobody's reading the pipe)
@@ -50,23 +51,3 @@ export function sendJSDataStream(connection: HubConnection, data: ArrayBufferVie
         }
     }, 0);
 };
-
-async function getNextChunk(data: ArrayBufferView | Blob, position: number, nextChunkSize: number): Promise<Uint8Array> {
-    if (data instanceof Blob) {
-        return await getChunkFromBlob(data, position, nextChunkSize);
-    } else {
-        return getChunkFromArrayBufferView(data, position, nextChunkSize);
-    }
-}
-
-async function getChunkFromBlob(data: Blob, position: number, nextChunkSize: number): Promise<Uint8Array> {
-    const chunkBlob = data.slice(position, position + nextChunkSize);
-    const arrayBuffer = await chunkBlob.arrayBuffer();
-    const nextChunkData = new Uint8Array(arrayBuffer);
-    return nextChunkData;
-}
-
-function getChunkFromArrayBufferView(data: ArrayBufferView, position: number, nextChunkSize: number) {
-    const nextChunkData = new Uint8Array(data.buffer, data.byteOffset + position, nextChunkSize);
-    return nextChunkData;
-}

src/Components/Web.JS/src/Platform/Mono/MonoPlatform.ts

@@ -488,7 +488,6 @@ function attachInteropInvoker(): void {
   const dotNetDispatcherEndInvokeJSMethodHandle = bindStaticMethod('Microsoft.AspNetCore.Components.WebAssembly', 'Microsoft.AspNetCore.Components.WebAssembly.Services.DefaultWebAssemblyJSRuntime', 'EndInvokeJS');
   const dotNetDispatcherNotifyByteArrayAvailableMethodHandle = bindStaticMethod('Microsoft.AspNetCore.Components.WebAssembly', 'Microsoft.AspNetCore.Components.WebAssembly.Services.DefaultWebAssemblyJSRuntime', 'NotifyByteArrayAvailable');
 
-
   DotNet.attachDispatcher({
     beginInvokeDotNetFromJS: (callId: number, assemblyName: string | null, methodIdentifier: string, dotNetObjectId: any | null, argsJson: string): void => {
       assertHeapIsNotLocked();

src/Components/Web.JS/src/StreamingInterop.ts

@@ -0,0 +1,19 @@
+export async function getNextChunk(data: ArrayBufferView | Blob, position: number, nextChunkSize: number): Promise<Uint8Array> {
+    if (data instanceof Blob) {
+        return await getChunkFromBlob(data, position, nextChunkSize);
+    } else {
+        return getChunkFromArrayBufferView(data, position, nextChunkSize);
+    }
+}
+
+async function getChunkFromBlob(data: Blob, position: number, nextChunkSize: number): Promise<Uint8Array> {
+    const chunkBlob = data.slice(position, position + nextChunkSize);
+    const arrayBuffer = await chunkBlob.arrayBuffer();
+    const nextChunkData = new Uint8Array(arrayBuffer);
+    return nextChunkData;
+}
+
+function getChunkFromArrayBufferView(data: ArrayBufferView, position: number, nextChunkSize: number) {
+    const nextChunkData = new Uint8Array(data.buffer, data.byteOffset + position, nextChunkSize);
+    return nextChunkData;
+}