Cleanup & Fix Test

Author: TanayParikh

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

@@ -45,9 +45,10 @@ public static async ValueTask<RemoteJSDataStream> CreateRemoteJSDataStreamAsync(
             TimeSpan jsInteropDefaultCallTimeout,
             CancellationToken cancellationToken = default)
         {
-            // Enforce minimum 1 kb SignalR message size as we budget 512 bytes
-            // overhead for the transfer, thus leaving at least 512 bytes for data
-            // transfer per chunk.
+            // 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 :
                 throw new ArgumentException($"SignalR MaximumIncomingBytes must be at least 1 kb.");
@@ -72,7 +73,7 @@ private RemoteJSDataStream(
             _jsInteropDefaultCallTimeout = jsInteropDefaultCallTimeout;
             _streamCancellationToken = cancellationToken;
 
-            _lastDataReceivedTime = DateTime.UtcNow;
+            _lastDataReceivedTime = DateTimeOffset.UtcNow;
             _ = ThrowOnTimeout();
 
             _runtime.RemoteJSDataStreamInstances.Add(_streamId, this);
@@ -85,7 +86,7 @@ private async Task<bool> ReceiveData(long chunkId, byte[] chunk, string error)
         {
             try
             {
-                _lastDataReceivedTime = DateTime.UtcNow;
+                _lastDataReceivedTime = DateTimeOffset.UtcNow;
                 _ = ThrowOnTimeout();
 
                 if (!string.IsNullOrEmpty(error))
@@ -195,7 +196,7 @@ private async Task ThrowOnTimeout()
         {
             await Task.Delay(_jsInteropDefaultCallTimeout);
 
-            if (!_disposed && (DateTime.UtcNow >= _lastDataReceivedTime.Add(_jsInteropDefaultCallTimeout)))
+            if (!_disposed && (DateTimeOffset.UtcNow >= _lastDataReceivedTime.Add(_jsInteropDefaultCallTimeout)))
             {
                 // Dispose of the stream if a chunk isn't received within the jsInteropDefaultCallTimeout.
                 var timeoutException = new TimeoutException("Did not receive any data in the alloted time.");

src/Components/Server/src/ComponentHub.cs

@@ -232,6 +232,11 @@ public async ValueTask<bool> ReceiveJSDataChunk(long streamId, long chunkId, byt
                 return false;
             }
 
+            // Note: this await will block the circuit. This is intentional.
+            // The call into the circuitHost.ReceiveJSDataChunk will block regardless as we call into Renderer.Dispatcher.InvokeAsync
+            // which ensures we're running on the main circuit thread so that the server/client remain in the same
+            // synchronization context. Additionally, we're utilizing the return value as a heartbeat for the transfer
+            // process, and without it would likely need to setup a separate endpoint to handle that functionality.
             return await circuitHost.ReceiveJSDataChunk(streamId, chunkId, chunk, error);
         }
 

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

@@ -74,38 +74,78 @@ public async void ReceiveData_SuccessReadsBackStream()
         }
 
         [Fact]
-        public async void ReceiveData_ReceiveDataTimeout_StreamDisposed()
+        public async void ReceiveData_NoDataProvidedBeforeTimeout_StreamDisposed()
         {
             // Arrange
             var jsRuntime = new TestRemoteJSRuntime(Options.Create(new CircuitOptions()), Options.Create(new HubOptions()), Mock.Of<ILogger<RemoteJSRuntime>>());
+            var timeoutExceptionRaised = false;
+            jsRuntime.UnhandledException += (_, ex) =>
+            {
+                Assert.Equal("Did not receive any data in the alloted time.", ex.Message);
+                timeoutExceptionRaised = ex is TimeoutException;
+            };
+
             var jsStreamReference = Mock.Of<IJSStreamReference>();
             var remoteJSDataStream = await RemoteJSDataStream.CreateRemoteJSDataStreamAsync(
-                jsRuntime ?? _jsRuntime,
+                jsRuntime,
                 jsStreamReference,
                 totalLength: 9,
                 maxBufferSize: 50,
                 maximumIncomingBytes: 10_000,
-                jsInteropDefaultCallTimeout: TimeSpan.FromSeconds(40)); // Note we're using a 40 second timeout for this test
+                jsInteropDefaultCallTimeout: TimeSpan.FromSeconds(10)); // Note we're using a 10 second timeout for this test
+            var streamId = GetStreamId(remoteJSDataStream, jsRuntime);
+            var chunk = new byte[] { 3, 5, 7 };
+
+            // Act & Assert 1
+            // Wait past the initial timeout + 15 sec buffer room and then
+            // confirm unhandled exception raised to crush circuit
+            await Task.Delay(TimeSpan.FromSeconds(25));
+            Assert.True(timeoutExceptionRaised);
+
+            // Act & Assert 2
+            // Ensures stream is disposed after the timeout and any additional chunks aren't accepted
+            var success = await RemoteJSDataStream.ReceiveData(jsRuntime, streamId, chunkId: 0, chunk, error: null);
+            Assert.False(success);
+        }
+
+        [Fact]
+        public async void ReceiveData_ReceivesDataThenTimesout_StreamDisposed()
+        {
+            // Arrange
+            var jsRuntime = new TestRemoteJSRuntime(Options.Create(new CircuitOptions()), Options.Create(new HubOptions()), Mock.Of<ILogger<RemoteJSRuntime>>());
+            var timeoutExceptionRaised = false;
+            jsRuntime.UnhandledException += (_, ex) =>
+            {
+                Assert.Equal("Did not receive any data in the alloted time.", ex.Message);
+                timeoutExceptionRaised = ex is TimeoutException;
+            };
+
+            var jsStreamReference = Mock.Of<IJSStreamReference>();
+            var remoteJSDataStream = await RemoteJSDataStream.CreateRemoteJSDataStreamAsync(
+                jsRuntime,
+                jsStreamReference,
+                totalLength: 9,
+                maxBufferSize: 50,
+                maximumIncomingBytes: 10_000,
+                jsInteropDefaultCallTimeout: TimeSpan.FromSeconds(30)); // Note we're using a 30 second timeout for this test
             var streamId = GetStreamId(remoteJSDataStream, jsRuntime);
             var chunk = new byte[] { 3, 5, 7 };
 
             // Act & Assert 1
             var success = await RemoteJSDataStream.ReceiveData(jsRuntime, streamId, chunkId: 0, chunk, error: null);
             Assert.True(success);
 
-            await Task.Delay(TimeSpan.FromSeconds(20));
+            await Task.Delay(TimeSpan.FromSeconds(15));
 
             // Act & Assert 2
             success = await RemoteJSDataStream.ReceiveData(jsRuntime, streamId, chunkId: 1, chunk, error: null);
             Assert.True(success);
 
-            // Wait 80 seconds (20 sec between first two calls + 40 sec timeout + 20 sec buffer room)
-            await Task.Delay(TimeSpan.FromSeconds(80));
-
             // Act & Assert 3
-            using var mem = new MemoryStream();
-            var ex = await Assert.ThrowsAsync<TimeoutException>(async() => await remoteJSDataStream.CopyToAsync(mem));
-            Assert.Equal("Did not receive any data in the alloted time.", ex.Message);
+            // Wait past the 30 sec timeout + 15 sec buffer room
+            // confirm unhandled exception raised to crush circuit
+            await Task.Delay(TimeSpan.FromSeconds(45));
+            Assert.True(timeoutExceptionRaised);
 
             // Act & Assert 4
             // Ensures stream is disposed after the timeout and any additional chunks aren't accepted
@@ -210,7 +250,7 @@ public TestRemoteJSRuntime(IOptions<CircuitOptions> circuitOptions, IOptions<Hub
             {
             }
 
-            new public ValueTask<TValue> InvokeAsync<TValue>(string identifier, object[] args)
+            public new ValueTask<TValue> InvokeAsync<TValue>(string identifier, object[] args)
             {
                 Assert.Equal("Blazor._internal.sendJSDataStream", identifier);
                 return ValueTask.FromResult<TValue>(default);

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

@@ -12,8 +12,6 @@ export function sendJSDataStream(connection: HubConnection, data: ArrayBufferVie
             let position = 0;
             let chunkId = 0;
 
-            // Note: The server-side `StreamBufferCapacity` option (defaults to 10) can be configured to limit how many
-            // stream items from the client (per stream) will be stored before reading any more stream items (thus applying backpressure).
             while (position < data.byteLength) {
                 const nextChunkSize = Math.min(chunkSize, data.byteLength - position);
                 const nextChunkData = new Uint8Array(data.buffer, data.byteOffset + position, nextChunkSize);