chore(@nestjs/microservices) add tests to grpc clients disconnection

Author: micmro

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)();
       });