Merge branch 'micmro-feature/grpc-cancellation'

Author: kamilmysliwiec

packages/microservices/client/client-grpc.ts

@@ -8,6 +8,7 @@ import { ClientOptions } from '../interfaces/client-metadata.interface';
 import { GRPC_DEFAULT_URL } from './../constants';
 import { ClientGrpc, GrpcOptions } from './../interfaces';
 import { ClientProxy } from './client-proxy';
+import { GRPC_CANCELLED } from './constants';
 
 let grpcPackage: any = {};
 
@@ -25,7 +26,7 @@ export class ClientGrpcProxy extends ClientProxy implements ClientGrpc {
     this.grpcClient = this.createClient();
   }
 
-  public getService<T = any>(name: string): T {
+  public getService<T extends {}>(name: string): T {
     const { options } = this.options as GrpcOptions;
     if (!this.grpcClient[name]) {
       throw new InvalidGrpcServiceException();
@@ -59,10 +60,30 @@ export class ClientGrpcProxy extends ClientProxy implements ClientGrpc {
   ): (...args) => Observable<any> {
     return (...args) => {
       return new Observable(observer => {
+        let isClientCanceled = false;
         const call = client[methodName](...args);
+
         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 === GRPC_CANCELLED) {
+            call.destroy();
+            if (isClientCanceled) {
+              return;
+            }
+          }
+          observer.error(error);
+        });
+        call.on('end', () => {
+          call.removeAllListeners();
+          observer.complete();
+        });
+        return () => {
+          if (call.finished) {
+            return undefined;
+          }
+          isClientCanceled = true;
+          call.cancel();
+        };
       });
     };
   }

packages/microservices/client/constants.ts

@@ -1,2 +1,3 @@
 export const ECONNREFUSED = 'ECONNREFUSED';
-export const CONN_ERR = 'CONN_ERR';
+export const CONN_ERR = 'CONN_ERR';
+export const GRPC_CANCELLED = 'Cancelled';

packages/microservices/constants.ts

@@ -10,6 +10,7 @@ export const MESSAGE_EVENT = 'message';
 export const ERROR_EVENT = 'error';
 export const CLOSE_EVENT = 'close';
 export const SUBSCRIBE = 'subscribe';
+export const CANCEL_EVENT = 'cancelled';
 
 export const PATTERN_METADATA = 'pattern';
 export const CLIENT_CONFIGURATION_METADATA = 'client';

packages/microservices/interfaces/client-grpc.interface.ts

@@ -1,3 +1,3 @@
 export interface ClientGrpc {
-  getService<T = any>(name: string): T;
+  getService<T extends {}>(name: string): T;
 }

packages/microservices/server/server-grpc.ts

@@ -1,7 +1,9 @@
+import { fromEvent } from 'rxjs';
+import { takeUntil } from 'rxjs/operators';
 import { InvalidGrpcPackageException } from '../exceptions/invalid-grpc-package.exception';
 import { InvalidProtoDefinitionException } from '../exceptions/invalid-proto-definition.exception';
 import { GrpcOptions, MicroserviceOptions } from '../interfaces/microservice-configuration.interface';
-import { GRPC_DEFAULT_URL } from './../constants';
+import { CANCEL_EVENT, GRPC_DEFAULT_URL } from './../constants';
 import { CustomTransportStrategy } from './../interfaces';
 import { Server } from './server';
 
@@ -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, CANCEL_EVENT)),
+      ).forEach(data => call.write(data));
       call.end();
     };
   }

packages/microservices/test/client/client-grpc.spec.ts

@@ -55,7 +55,7 @@ describe('ClientGrpcProxy', () => {
         const cln = { [methodName]: { responseStream: false } };
         const spy = sinon.spy(client, 'createUnaryServiceMethod');
         client.createServiceMethod(cln, methodName);
-        
+
         expect(spy.called).to.be.true;
       });
     });
@@ -71,7 +71,7 @@ describe('ClientGrpcProxy', () => {
       const obj = { [methodName]: () => ({ on: (type, fn) => fn() }) };
 
       let stream$: Observable<any>;
-    
+
       beforeEach(() => {
         stream$ = client.createStreamServiceMethod(obj, methodName)();
       });
@@ -83,6 +83,75 @@ describe('ClientGrpcProxy', () => {
         expect(spy.called).to.be.true;
       });
     });
+
+    describe('flow-control', () => {
+      const methodName = 'm';
+      type EvtCallback = (...args: any[]) => void;
+      let callMock: {
+        on: (type: string, fn: EvtCallback) => void,
+        cancel: sinon.SinonSpy,
+        finished: boolean,
+        destroy: sinon.SinonSpy,
+        removeAllListeners: sinon.SinonSpy,
+      };
+      let eventCallbacks: { [type: string]: EvtCallback };
+      let obj;
+      const dataSpy = sinon.spy();
+      const errorSpy = sinon.spy();
+      const completeSpy = sinon.spy();
+
+      let stream$: Observable<any>;
+
+      beforeEach(() => {
+        dataSpy.reset();
+        errorSpy.reset();
+        completeSpy.reset();
+        eventCallbacks = {};
+        callMock = {
+          on: (type, fn) => eventCallbacks[type] = fn,
+          cancel: sinon.spy(),
+          finished: false,
+          destroy: sinon.spy(),
+          removeAllListeners: sinon.spy(),
+        };
+        obj = { [methodName]: () => callMock };
+        stream$ = client.createStreamServiceMethod(obj, methodName)();
+      });
+
+      it('propagates server errors', () => {
+        const err = new Error('something happened');
+        stream$.subscribe(dataSpy, errorSpy, completeSpy);
+        eventCallbacks.data('a');
+        eventCallbacks.data('b');
+        callMock.finished = true;
+        eventCallbacks.error(err);
+        eventCallbacks.data('c');
+
+        expect(Object.keys(eventCallbacks).length).to.eq(3);
+        expect(dataSpy.args).to.eql([['a'], ['b']]);
+        expect(errorSpy.args[0][0]).to.eql(err);
+        expect(completeSpy.called).to.be.false;
+        expect(callMock.cancel.called).to.be.false;
+      });
+
+      it('handles client side cancel', () => {
+        const grpcServerCancelErrMock = {
+          details: 'Cancelled',
+        };
+        const subscription = stream$.subscribe(dataSpy, errorSpy);
+        eventCallbacks.data('a');
+        eventCallbacks.data('b');
+        subscription.unsubscribe();
+        eventCallbacks.error(grpcServerCancelErrMock);
+        eventCallbacks.end();
+        eventCallbacks.data('c');
+
+        expect(callMock.cancel.called, 'should call call.cancel()').to.be.true;
+        expect(callMock.destroy.called, 'should call call.destroy()').to.be.true;
+        expect(dataSpy.args).to.eql([['a'], ['b']]);
+        expect(errorSpy.called, 'should not error if client canceled').to.be.false;
+      });
+    });
   });
 
   describe('createUnaryServiceMethod', () => {
@@ -95,7 +164,7 @@ describe('ClientGrpcProxy', () => {
       const obj = { [methodName]: (callback) => callback(null, {}) };
 
       let stream$: Observable<any>;
-    
+
       beforeEach(() => {
         stream$ = client.createUnaryServiceMethod(obj, methodName)();
       });

packages/microservices/test/server/server-grpc.spec.ts

@@ -1,7 +1,7 @@
 import { ServerGrpc } from '../../server/server-grpc';
 import { expect } from 'chai';
 import * as sinon from 'sinon';
-import { Observable } from 'rxjs';
+import { of } from 'rxjs';
 import { join } from 'path';
 import { InvalidGrpcPackageException } from '../../exceptions/invalid-grpc-package.exception';
 
@@ -149,12 +149,36 @@ describe('ServerGrpc', () => {
     });
     describe('on call', () => {
       it('should call native method', async () => {
-        const call = { write: sinon.spy(), end: sinon.spy() };
+        const call = {
+          write: sinon.spy(),
+          end: sinon.spy(),
+          addListener: sinon.spy(),
+          removeListener: sinon.spy(),
+        };
         const callback = sinon.spy();
         const native = sinon.spy();
 
         await server.createStreamServiceMethod(native)(call, callback);
         expect(native.called).to.be.true;
+        expect(call.addListener.calledWith('cancelled')).to.be.true;
+        expect(call.removeListener.calledWith('cancelled')).to.be.true;
+      });
+
+      it(`should close the result observable when receiving an 'cancelled' event from the client`, async () => {
+        let cancelCb: () => void;
+        const call = {
+          write: sinon.stub().onSecondCall().callsFake(() => cancelCb()),
+          end: sinon.spy(),
+          addListener: (name, cb) => cancelCb = cb,
+          removeListener: sinon.spy(),
+        };
+        const result$ = of(1, 2, 3);
+        const callback = sinon.spy();
+        const native = sinon.stub().returns(new Promise((resolve, reject) => resolve(result$)));
+
+        await server.createStreamServiceMethod(native)(call, callback);
+        expect(call.write.calledTwice).to.be.true;
+        expect(call.end.called).to.be.true;
       });
     });
   });