Prevent an exception to be thrown when a web connection stream is closed (#414)

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

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 {