Streaming Interop Followup Items

Author: TanayParikh

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

@@ -40,10 +40,11 @@ public static async ValueTask<RemoteJSDataStream> CreateRemoteJSDataStreamAsync(
             RemoteJSRuntime runtime,
             IJSStreamReference jsStreamReference,
             long totalLength,
-            long maxBufferSize,
             long maximumIncomingBytes,
             TimeSpan jsInteropDefaultCallTimeout,
-            CancellationToken cancellationToken = default)
+            long pauseIncomingBytesThreshold,
+            long resumeIncomingBytesThreshold,
+            CancellationToken cancellationToken)
         {
             // 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
@@ -54,7 +55,7 @@ public static async ValueTask<RemoteJSDataStream> CreateRemoteJSDataStreamAsync(
                 throw new ArgumentException($"SignalR MaximumIncomingBytes must be at least 1 kb.");
 
             var streamId = runtime.RemoteJSDataStreamNextInstanceId++;
-            var remoteJSDataStream = new RemoteJSDataStream(runtime, streamId, totalLength, maxBufferSize, jsInteropDefaultCallTimeout, cancellationToken);
+            var remoteJSDataStream = new RemoteJSDataStream(runtime, streamId, totalLength, jsInteropDefaultCallTimeout, pauseIncomingBytesThreshold, resumeIncomingBytesThreshold, cancellationToken);
             await runtime.InvokeVoidAsync("Blazor._internal.sendJSDataStream", jsStreamReference, streamId, chunkSize);
             return remoteJSDataStream;
         }
@@ -63,8 +64,9 @@ private RemoteJSDataStream(
             RemoteJSRuntime runtime,
             long streamId,
             long totalLength,
-            long maxBufferSize,
             TimeSpan jsInteropDefaultCallTimeout,
+            long pauseIncomingBytesThreshold,
+            long resumeIncomingBytesThreshold,
             CancellationToken cancellationToken)
         {
             _runtime = runtime;
@@ -78,7 +80,7 @@ private RemoteJSDataStream(
 
             _runtime.RemoteJSDataStreamInstances.Add(_streamId, this);
 
-            _pipe = new Pipe(new PipeOptions(pauseWriterThreshold: maxBufferSize, resumeWriterThreshold: maxBufferSize / 2));
+            _pipe = new Pipe(new PipeOptions(pauseWriterThreshold: pauseIncomingBytesThreshold, resumeWriterThreshold: resumeIncomingBytesThreshold));
             _pipeReaderStream = _pipe.Reader.AsStream();
         }
 

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 maxBufferSize, CancellationToken cancellationToken)
-            => await RemoteJSDataStream.CreateRemoteJSDataStreamAsync(this, jsStreamReference, totalLength, maxBufferSize, _maximumIncomingBytes, _options.JSInteropDefaultCallTimeout, cancellationToken);
+        protected override async Task<Stream> ReadJSDataAsStreamAsync(IJSStreamReference jsStreamReference, long totalLength, long pauseIncomingBytesThreshold, long resumeIncomingBytesThreshold, CancellationToken cancellationToken)
+            => await RemoteJSDataStream.CreateRemoteJSDataStreamAsync(this, jsStreamReference, totalLength, _maximumIncomingBytes, _options.JSInteropDefaultCallTimeout, pauseIncomingBytesThreshold, resumeIncomingBytesThreshold, cancellationToken);
 
         public static class Log
         {

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

@@ -4,6 +4,7 @@
 using System;
 using System.IO;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.SignalR;
 using Microsoft.Extensions.Logging;
@@ -25,7 +26,7 @@ public async void CreateRemoteJSDataStreamAsync_CreatesStream()
             var jsStreamReference = Mock.Of<IJSStreamReference>();
 
             // Act
-            var remoteJSDataStream = await RemoteJSDataStream.CreateRemoteJSDataStreamAsync(_jsRuntime, jsStreamReference, totalLength: 100, maxBufferSize: 50, maximumIncomingBytes: 10_000, jsInteropDefaultCallTimeout: TimeSpan.FromMinutes(1));
+            var remoteJSDataStream = await RemoteJSDataStream.CreateRemoteJSDataStreamAsync(_jsRuntime, jsStreamReference, totalLength: 100, maximumIncomingBytes: 10_000, jsInteropDefaultCallTimeout: TimeSpan.FromMinutes(1), pauseIncomingBytesThreshold: 50, resumeIncomingBytesThreshold: 25, cancellationToken: CancellationToken.None);
 
             // Assert
             Assert.NotNull(remoteJSDataStream);
@@ -116,7 +117,7 @@ public async void 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, maxBufferSize: 50, maximumIncomingBytes: 10_000, jsInteropDefaultCallTimeout: TimeSpan.FromMinutes(1));
+            var remoteJSDataStream = await RemoteJSDataStream.CreateRemoteJSDataStreamAsync(jsRuntime, jsStreamReference, totalLength: 100, maximumIncomingBytes: 10_000, jsInteropDefaultCallTimeout: TimeSpan.FromMinutes(1), pauseIncomingBytesThreshold: 50, resumeIncomingBytesThreshold: 25, cancellationToken: CancellationToken.None);
             var streamId = GetStreamId(remoteJSDataStream, jsRuntime);
             var chunk = new byte[110]; // 100 byte totalLength for stream
 
@@ -136,7 +137,7 @@ public async void 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, maxBufferSize: 50, maximumIncomingBytes: 10_000, jsInteropDefaultCallTimeout: TimeSpan.FromMinutes(1));
+            var remoteJSDataStream = await RemoteJSDataStream.CreateRemoteJSDataStreamAsync(jsRuntime, jsStreamReference, totalLength: 100, maximumIncomingBytes: 10_000, jsInteropDefaultCallTimeout: TimeSpan.FromMinutes(1), pauseIncomingBytesThreshold: 50, resumeIncomingBytesThreshold: 25, cancellationToken: CancellationToken.None);
             var streamId = GetStreamId(remoteJSDataStream, jsRuntime);
             var chunk = new byte[5];
 
@@ -157,7 +158,7 @@ public async void ReceiveData_ProvidedWithOutOfOrderChunk_SimulatesSignalRDiscon
         private static async Task<RemoteJSDataStream> CreateRemoteJSDataStreamAsync(TestRemoteJSRuntime jsRuntime = null)
         {
             var jsStreamReference = Mock.Of<IJSStreamReference>();
-            var remoteJSDataStream = await RemoteJSDataStream.CreateRemoteJSDataStreamAsync(jsRuntime ?? _jsRuntime, jsStreamReference, totalLength: 100, maxBufferSize: 50, maximumIncomingBytes: 10_000, jsInteropDefaultCallTimeout: TimeSpan.FromMinutes(1));
+            var remoteJSDataStream = await RemoteJSDataStream.CreateRemoteJSDataStreamAsync(jsRuntime ?? _jsRuntime, jsStreamReference, totalLength: 100, maximumIncomingBytes: 10_000, jsInteropDefaultCallTimeout: TimeSpan.FromMinutes(1), pauseIncomingBytesThreshold: 50, resumeIncomingBytesThreshold: 25, cancellationToken: CancellationToken.None);
             return remoteJSDataStream;
         }
 

src/Components/Web/src/Forms/InputFile/RemoteBrowserFileStream.cs

@@ -45,7 +45,11 @@ private async Task<Stream> OpenReadStreamAsync(RemoteBrowserFileStreamOptions op
                 _inputFileElement,
                 File.Id);
 
-            return await dataReference.OpenReadStreamAsync(_maxAllowedSize, options.MaxBufferSize, cancellationToken);
+            return await dataReference.OpenReadStreamAsync(
+                _maxAllowedSize,
+                pauseIncomingBytesThreshold: options.MaxBufferSize,
+                resumeIncomingBytesThreshold: options.MaxBufferSize / 2,
+                cancellationToken);
         }
 
         protected override async ValueTask<int> CopyFileDataIntoBuffer(long sourceOffset, Memory<byte> destination, CancellationToken cancellationToken)

src/JSInterop/Microsoft.JSInterop/src/IJSStreamReference.cs

@@ -16,15 +16,16 @@ public interface IJSStreamReference : IAsyncDisposable
         /// <summary>
         /// Length of the <see cref="Stream"/> provided by JavaScript.
         /// </summary>
-        public long Length { get; }
+        long Length { get; }
 
         /// <summary>
         /// Opens a <see cref="Stream"/> with the <see cref="JSRuntime"/> for the current data reference.
         /// </summary>
         /// <param name="maxAllowedSize">Maximum number of bytes permitted to be read from JavaScript.</param>
-        /// <param name="maxBufferSize">Maximum number of bytes that are allowed to be buffered.</param>
+        /// <param name="pauseIncomingBytesThreshold">The number of unconsumed bytes to accept from JS before blocking. A value of zero prevents JS from blocking, allowing .NET to receive an unlimited number of bytes.</param>
+        /// <param name="resumeIncomingBytesThreshold">The number of unflushed bytes at which point JS stops blocking.</param>
         /// <param name="cancellationToken"><see cref="CancellationToken" /> for cancelling read.</param>
         /// <returns><see cref="Stream"/> which can provide data associated with the current data reference.</returns>
-        ValueTask<Stream> OpenReadStreamAsync(long maxAllowedSize = 512000, long maxBufferSize = 100 * 1024, CancellationToken cancellationToken = default);
+        ValueTask<Stream> OpenReadStreamAsync(long maxAllowedSize = 512000, long pauseIncomingBytesThreshold = -1, long resumeIncomingBytesThreshold = -1, CancellationToken cancellationToken = default);
     }
 }

src/JSInterop/Microsoft.JSInterop/src/Implementation/JSStreamReference.cs

@@ -36,14 +36,14 @@ internal JSStreamReference(JSRuntime jsRuntime, long id, long totalLength) : bas
         }
 
         /// <inheritdoc />
-        async ValueTask<Stream> IJSStreamReference.OpenReadStreamAsync(long maxLength, long maxBufferSize, CancellationToken cancellationToken)
+        async ValueTask<Stream> IJSStreamReference.OpenReadStreamAsync(long maxLength, long pauseIncomingBytesThreshold, long resumeIncomingBytesThreshold, CancellationToken cancellationToken)
         {
             if (Length > maxLength)
             {
                 throw new ArgumentOutOfRangeException(nameof(maxLength), $"The incoming data stream of length {Length} exceeds the maximum length {maxLength}.");
             }
 
-            return await _jsRuntime.ReadJSDataAsStreamAsync(this, Length, maxBufferSize, cancellationToken);
+            return await _jsRuntime.ReadJSDataAsStreamAsync(this, Length, pauseIncomingBytesThreshold, resumeIncomingBytesThreshold, cancellationToken);
         }
     }
 }

src/JSInterop/Microsoft.JSInterop/src/JSRuntime.cs

@@ -215,10 +215,11 @@ protected internal virtual void ReceiveByteArray(int id, byte[] data)
         /// </summary>
         /// <param name="jsStreamReference"><see cref="IJSStreamReference"/> to produce a data stream for.</param>
         /// <param name="totalLength">Expected length of the incoming data stream.</param>
-        /// <param name="maxBufferSize">Amount of bytes to buffer before flushing.</param>
+        /// <param name="pauseIncomingBytesThreshold">The number of unconsumed bytes to accept from JS before blocking. A value of zero prevents JS from blocking, allowing .NET to receive an unlimited number of bytes.</param>
+        /// <param name="resumeIncomingBytesThreshold">The number of unflushed bytes at which point JS stops blocking.</param>
         /// <param name="cancellationToken"><see cref="CancellationToken" /> for cancelling read.</param>
         /// <returns><see cref="Stream"/> for the data reference represented by <paramref name="jsStreamReference"/>.</returns>
-        protected internal virtual Task<Stream> ReadJSDataAsStreamAsync(IJSStreamReference jsStreamReference, long totalLength, long maxBufferSize, CancellationToken cancellationToken)
+        protected internal virtual Task<Stream> ReadJSDataAsStreamAsync(IJSStreamReference jsStreamReference, long totalLength, long pauseIncomingBytesThreshold, long resumeIncomingBytesThreshold, CancellationToken cancellationToken)
         {
             // The reason it's virtual and not abstract is just for back-compat
 

src/JSInterop/Microsoft.JSInterop/src/PublicAPI.Unshipped.txt

@@ -1,7 +1,7 @@
 #nullable enable
 Microsoft.JSInterop.IJSStreamReference
 Microsoft.JSInterop.IJSStreamReference.Length.get -> long
-Microsoft.JSInterop.IJSStreamReference.OpenReadStreamAsync(long maxAllowedSize = 512000, long maxBufferSize = 102400, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask<System.IO.Stream!>
+Microsoft.JSInterop.IJSStreamReference.OpenReadStreamAsync(long maxAllowedSize = 512000, long pauseIncomingBytesThreshold = -1, long resumeIncomingBytesThreshold = -1, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask<System.IO.Stream!>
 Microsoft.JSInterop.Implementation.JSStreamReference
 Microsoft.JSInterop.Implementation.JSStreamReference.Length.get -> long
 Microsoft.JSInterop.Implementation.JSObjectReferenceJsonWorker
@@ -36,7 +36,7 @@ static Microsoft.JSInterop.JSRuntimeExtensions.InvokeVoidAsync(this Microsoft.JS
 static Microsoft.JSInterop.JSRuntimeExtensions.InvokeVoidAsync(this Microsoft.JSInterop.IJSRuntime! jsRuntime, string! identifier, System.TimeSpan timeout, params object?[]? args) -> System.Threading.Tasks.ValueTask
 *REMOVED*static Microsoft.JSInterop.JSRuntimeExtensions.InvokeVoidAsync(this Microsoft.JSInterop.IJSRuntime! jsRuntime, string! identifier, params object![]! args) -> System.Threading.Tasks.ValueTask
 static Microsoft.JSInterop.JSRuntimeExtensions.InvokeVoidAsync(this Microsoft.JSInterop.IJSRuntime! jsRuntime, string! identifier, params object?[]? args) -> System.Threading.Tasks.ValueTask
-virtual Microsoft.JSInterop.JSRuntime.ReadJSDataAsStreamAsync(Microsoft.JSInterop.IJSStreamReference! jsStreamReference, long totalLength, long maxBufferSize, System.Threading.CancellationToken cancellationToken) -> System.Threading.Tasks.Task<System.IO.Stream!>!
+virtual Microsoft.JSInterop.JSRuntime.ReadJSDataAsStreamAsync(Microsoft.JSInterop.IJSStreamReference! jsStreamReference, long totalLength, long pauseIncomingBytesThreshold, long resumeIncomingBytesThreshold, System.Threading.CancellationToken cancellationToken) -> System.Threading.Tasks.Task<System.IO.Stream!>!
 virtual Microsoft.JSInterop.JSRuntime.ReceiveByteArray(int id, byte[]! data) -> void
 virtual Microsoft.JSInterop.JSRuntime.SendByteArray(int id, byte[]! data) -> void
 Microsoft.JSInterop.JSDisconnectedException

src/JSInterop/Microsoft.JSInterop/test/JSRuntimeTest.cs

@@ -404,7 +404,7 @@ public async void ReadJSDataAsStreamAsync_ThrowsNotSupportedException()
             var dataReference = new JSStreamReference(runtime, 10, 10);
 
             // Act
-            var exception = await Assert.ThrowsAsync<NotSupportedException>(async () => await runtime.ReadJSDataAsStreamAsync(dataReference, 10, 10, CancellationToken.None));
+            var exception = await Assert.ThrowsAsync<NotSupportedException>(async () => await runtime.ReadJSDataAsStreamAsync(dataReference, 10, 10, 10, CancellationToken.None));
 
             // Assert
             Assert.Equal("The current JavaScript runtime does not support reading data streams.", exception.Message);