From 93c21feb56c5d1eb73f0a54bd21487c4e1136e54 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Luis=20Alejandro=20Herrera=20Le=C3=B3n?=
 <alex.herrera@joylabs.com>
Date: Sun, 20 Dec 2020 14:05:59 -0800
Subject: [PATCH] Prevent an exception to be thrown when a web connection
 stream is closed (#414)

---
 lib/src/client/transport/xhr_transport.dart |  6 +++---
 test/grpc_web_test.dart                     | 22 +++++++++++++++++++++
 2 files changed, 25 insertions(+), 3 deletions(-)

diff --git a/lib/src/client/transport/xhr_transport.dart b/lib/src/client/transport/xhr_transport.dart
index d496f80..9244491 100644
--- a/lib/src/client/transport/xhr_transport.dart
+++ b/lib/src/client/transport/xhr_transport.dart
@@ -59,7 +59,7 @@ class XhrTransportStream implements GrpcTransportStream {
         .listen((data) => _request.send(data), cancelOnError: true);
 
     _request.onReadyStateChange.listen((data) {
-      if (_incomingMessages.isClosed) {
+      if (_incomingProcessor.isClosed) {
         return;
       }
       switch (_request.readyState) {
@@ -80,7 +80,7 @@ class XhrTransportStream implements GrpcTransportStream {
     });
 
     _request.onError.listen((ProgressEvent event) {
-      if (_incomingMessages.isClosed) {
+      if (_incomingProcessor.isClosed) {
         return;
       }
       _onError(GrpcError.unavailable('XhrConnection connection-error'),
@@ -89,7 +89,7 @@ class XhrTransportStream implements GrpcTransportStream {
     });
 
     _request.onProgress.listen((_) {
-      if (_incomingMessages.isClosed) {
+      if (_incomingProcessor.isClosed) {
         return;
       }
       // Use response over responseText as most browsers don't support
diff --git a/test/grpc_web_test.dart b/test/grpc_web_test.dart
index 7b347bc..25a70b3 100644
--- a/test/grpc_web_test.dart
+++ b/test/grpc_web_test.dart
@@ -94,6 +94,28 @@ void main() {
     terminated = true;
     await channel.terminate();
   });
+
+  // Verify that stream cancellation does not cause an exception
+  test("stream cancellation works", () async {
+    final channel = GrpcWebClientChannel.xhr(server.uri);
+    final service = EchoServiceClient(channel);
+
+    const testMessage = 'hello from gRPC-web';
+
+    final stream = service
+        .serverStreamingEcho(ServerStreamingEchoRequest()
+          ..message = testMessage
+          ..messageCount = 20
+          ..messageInterval = 100)
+        .listen((response) {
+      expect(response.message, equals(testMessage));
+    });
+
+    await Future.delayed(Duration(milliseconds: 500));
+    await stream.cancel();
+
+    await channel.terminate();
+  });
 }
 
 class GrpcWebServer {