Return false instead of throwing InvalidOperationException on read of a gRPC server stream after the connection was closed

Author: erikmav

src/Grpc.AspNetCore.Server/Internal/HttpContextStreamReader.cs

@@ -58,7 +58,8 @@ async Task<bool> MoveNextAsync(ValueTask<TRequest?> readStreamTask)
 
             if (_completed || _serverCallContext.CancellationToken.IsCancellationRequested)
             {
-                return Task.FromException<bool>(new InvalidOperationException("Can't read messages after the request is complete."));
+                // gRPC specification indicates that MoveNext() should not throw. Simply return false.
+                return CommonGrpcProtocolHelpers.FalseTask;
             }
 
             var request = _serverCallContext.HttpContext.Request.BodyReader.ReadStreamMessageAsync(_serverCallContext, _deserializer, cancellationToken);

test/FunctionalTests/Client/StreamingTests.cs

@@ -42,7 +42,7 @@ namespace Grpc.AspNetCore.FunctionalTests.Client
     public class StreamingTests : FunctionalTestBase
     {
         [Test]
-        public async Task DuplexStream_SendLargeFileBatchedAndRecieveLargeFileBatched_Success()
+        public async Task DuplexStream_SendLargeFileBatchedAndReceiveLargeFileBatched_Success()
         {
             // Arrange
             var data = CreateTestData(1024 * 1024 * 1); // 1 MB
@@ -306,7 +306,7 @@ await call.RequestStream.WriteAsync(new DataMessage
         [TestCase(1)]
         [TestCase(5)]
         [TestCase(20)]
-        public async Task DuplexStreaming_SimultaniousSendAndReceiveInParallel_Success(int tasks)
+        public async Task DuplexStreaming_SimultaneousSendAndReceiveInParallel_Success(int tasks)
         {
             // Arrange
             const int total = 1024 * 1024 * 1;
@@ -316,7 +316,7 @@ public async Task DuplexStreaming_SimultaniousSendAndReceiveInParallel_Success(i
 
             var client = new StreamService.StreamServiceClient(Channel);
 
-            await TestHelpers.RunParallel(tasks, async taskIndex =>
+            await TestHelpers.RunParallel(tasks, async _ =>
             {
                 var (sent, received) = await EchoData(total, data, client).DefaultTimeout();
 
@@ -421,7 +421,7 @@ await TestHelpers.AssertIsTrueRetryAsync(
         [Test]
         public async Task DuplexStreaming_ParallelCallsFromOneChannel_Success()
         {
-            async Task UnaryDeadlineExceeded(IAsyncStreamReader<DataMessage> requestStream, IServerStreamWriter<DataMessage> responseStream, ServerCallContext context)
+            static async Task UnaryDeadlineExceeded(IAsyncStreamReader<DataMessage> requestStream, IServerStreamWriter<DataMessage> responseStream, ServerCallContext context)
             {
                 await foreach (var message in requestStream.ReadAllAsync())
                 {
@@ -460,7 +460,7 @@ async Task UnaryDeadlineExceeded(IAsyncStreamReader<DataMessage> requestStream,
         [Test]
         public async Task ServerStreaming_GetTrailersAndStatus_Success()
         {
-            async Task ServerStreamingWithTrailers(DataMessage request, IServerStreamWriter<DataMessage> responseStream, ServerCallContext context)
+            static async Task ServerStreamingWithTrailers(DataMessage request, IServerStreamWriter<DataMessage> responseStream, ServerCallContext context)
             {
                 await responseStream.WriteAsync(new DataMessage());
                 context.ResponseTrailers.Add("my-trailer", "value");
@@ -625,7 +625,7 @@ async Task ServerStreamingWithTrailers(DataMessage request, IServerStreamWriter<
 
         [TestCase(true)]
         [TestCase(false)]
-        public async Task ClientStreaming_ReadAfterMethodComplete_Error(bool readBeforeExit)
+        public async Task ClientStreaming_ReadAfterMethodComplete_False(bool readBeforeExit)
         {
             SetExpectedErrorsFilter(writeContext =>
             {
@@ -641,7 +641,7 @@ public async Task ClientStreaming_ReadAfterMethodComplete_Error(bool readBeforeE
             var tcs = new TaskCompletionSource<object?>(TaskCreationOptions.RunContinuationsAsynchronously);
             var readTcs = new TaskCompletionSource<Task>(TaskCreationOptions.RunContinuationsAsynchronously);
             var syncPoint = new SyncPoint(runContinuationsAsynchronously: true);
-            async Task<DataMessage> ClientStreamingWithTrailers(IAsyncStreamReader<DataMessage> requestStream, ServerCallContext context)
+            async Task<DataMessage> ClientStreamingWithTrailersAsync(IAsyncStreamReader<DataMessage> requestStream, ServerCallContext context)
             {
                 var readTask = Task.Run(async () =>
                 {
@@ -661,7 +661,7 @@ async Task<DataMessage> ClientStreamingWithTrailers(IAsyncStreamReader<DataMessa
             }
 
             // Arrange
-            var method = Fixture.DynamicGrpc.AddClientStreamingMethod<DataMessage, DataMessage>(ClientStreamingWithTrailers);
+            var method = Fixture.DynamicGrpc.AddClientStreamingMethod<DataMessage, DataMessage>(ClientStreamingWithTrailersAsync);
 
             var channel = CreateChannel();
 
@@ -680,13 +680,13 @@ async Task<DataMessage> ClientStreamingWithTrailers(IAsyncStreamReader<DataMessa
 
             tcs.SetResult(null);
 
-            var response = await call;
+            DataMessage response = await call;
+            Assert.IsNotNull(response);
 
             syncPoint.Continue();
 
             var readTask = await readTcs.Task.DefaultTimeout();
-            var ex = await ExceptionAssert.ThrowsAsync<InvalidOperationException>(() => readTask).DefaultTimeout();
-            Assert.AreEqual("Can't read messages after the request is complete.", ex.Message);
+            await readTask.DefaultTimeout();
 
             var clientException = await ExceptionAssert.ThrowsAsync<RpcException>(() => call.RequestStream.WriteAsync(new DataMessage())).DefaultTimeout();
             Assert.AreEqual(StatusCode.OK, clientException.StatusCode);