feature(@nestjs/microservices) allow grpc clients to disconnect

- when client cancels the server stream gets completed too
- client introduced disconnection does not throw errors on the client side
- release client's event listener on 'end'

Author: micmro

packages/microservices/client/client-grpc.ts

@@ -60,9 +60,27 @@ export class ClientGrpcProxy extends ClientProxy implements ClientGrpc {
     return (...args) => {
       return new Observable(observer => {
         const call = client[methodName](...args);
+        let isClientCanceled = false;
         call.on('data', (data: any) => observer.next(data));
-        call.on('error', (error: any) => observer.error(error));
-        call.on('end', () => observer.complete());
+        call.on('error', (error: any) => {
+          if (error.details === 'Cancelled') {
+            call.destroy();
+            if ( isClientCanceled ) {
+              return; // do not error if cancel was inititiated by Client
+            }
+          }
+          observer.error(error);
+        });
+        call.on('end', () => {
+          call.removeAllListeners();
+          observer.complete();
+        });
+        return () => {
+          if (!call.finished) {
+            isClientCanceled = true;
+            call.cancel();
+          }
+        };
       });
     };
   }

packages/microservices/server/server-grpc.ts

@@ -4,6 +4,8 @@ import { GrpcOptions, MicroserviceOptions } from '../interfaces/microservice-con
 import { GRPC_DEFAULT_URL } from './../constants';
 import { CustomTransportStrategy } from './../interfaces';
 import { Server } from './server';
+import { fromEvent } from 'rxjs';
+import { takeUntil } from 'rxjs/operators';
 
 let grpcPackage: any = {};
 
@@ -103,7 +105,9 @@ export class ServerGrpc extends Server implements CustomTransportStrategy {
     return async (call, callback) => {
       const handler = methodHandler(call.request, call.metadata);
       const result$ = this.transformToObservable(await handler);
-      await result$.forEach(data => call.write(data));
+      await result$.pipe(
+        takeUntil(fromEvent(call, 'cancelled')),
+      ).forEach(data => call.write(data));
       call.end();
     };
   }