fix(microservices): fix client-side grpc streaming

Author: kamilmysliwiec

integration/microservices/e2e/orders-grpc.spec.ts

@@ -166,4 +166,52 @@ describe('Advanced GRPC transport', () => {
       setTimeout(() => resolve(), 1000);
     });
   });
+
+  it('GRPC Sending Stream and receiving a single message from RX handler', async () => {
+    const callHandler = client.streamReq((err, res) => {
+      if (err) {
+        throw err;
+      }
+      expect(res).to.eql({
+        id: 1,
+        itemTypes: [1],
+        shipmentType: {
+          from: 'test',
+          to: 'test1',
+          carrier: 'test-carrier',
+        },
+      });
+    });
+
+    return new Promise((resolve, reject) => {
+      callHandler.write({
+        id: 1,
+      });
+      setTimeout(() => resolve(), 1000);
+    });
+  });
+
+  it('GRPC Sending Stream and receiving a single message from Call handler', async () => {
+    const callHandler = client.streamReqCall((err, res) => {
+      if (err) {
+        throw err;
+      }
+      expect(res).to.eql({
+        id: 1,
+        itemTypes: [1],
+        shipmentType: {
+          from: 'test',
+          to: 'test1',
+          carrier: 'test-carrier',
+        },
+      })
+    });
+
+    return new Promise((resolve, reject) => {
+      callHandler.write({
+        id: 1,
+      });
+      setTimeout(() => resolve(), 1000);
+    });
+  });
 });

integration/microservices/src/grpc-advanced/advanced.grpc.controller.ts

@@ -97,4 +97,44 @@ export class AdvancedGrpcController {
       });
     });
   }
+
+  @GrpcStreamMethod('orders.OrderService')
+  async streamReq(messages: Observable<any>): Promise<any> {
+    const s = new Subject();
+    const o = s.asObservable();
+    messages.subscribe(
+      msg => {
+        s.next({
+          id: 1,
+          itemTypes: [1],
+          shipmentType: {
+            from: 'test',
+            to: 'test1',
+            carrier: 'test-carrier',
+          },
+        });
+      },
+      null,
+      () => s.complete(),
+    );
+    return o;
+  }
+
+  @GrpcStreamCall('orders.OrderService')
+  async streamReqCall(stream: any, callback: Function) {
+    stream.on('data', (msg: any) => {
+      // process msg
+    });
+    stream.on('end', () => {
+      callback(null, {
+        id: 1,
+        itemTypes: [1],
+        shipmentType: {
+          from: 'test',
+          to: 'test1',
+          carrier: 'test-carrier',
+        },
+      });
+    });
+  }
 }

integration/microservices/src/grpc-advanced/proto/orders/service.proto

@@ -6,4 +6,6 @@ service OrderService {
     rpc Find(Order) returns (Order);
     rpc Sync(stream Order) returns (stream Order);
     rpc SyncCall(stream Order) returns (stream Order);
+    rpc StreamReq(stream Order) returns (Order);
+    rpc StreamReqCall(stream Order) returns (Order);
 }

packages/microservices/server/server-grpc.ts

@@ -182,11 +182,19 @@ export class ServerGrpc extends Server implements CustomTransportStrategy {
     // streaming from the side of requester
     if (protoNativeHandler.requestStream) {
       // If any handlers were defined with GrpcStreamMethod annotation use RX
-      if (streamType === GrpcMethodStreamingType.RX_STREAMING)
-        return this.createStreamDuplexMethod(methodHandler);
+      if (streamType === GrpcMethodStreamingType.RX_STREAMING) {
+        return this.createRequestStreamMethod(
+          methodHandler,
+          protoNativeHandler.responseStream,
+        );
+      }
       // If any handlers were defined with GrpcStreamCall annotation
-      else if (streamType === GrpcMethodStreamingType.PT_STREAMING)
-        return this.createStreamCallMethod(methodHandler);
+      else if (streamType === GrpcMethodStreamingType.PT_STREAMING) {
+        return this.createStreamCallMethod(
+          methodHandler,
+          protoNativeHandler.responseStream,
+        );
+      }
     }
     return protoNativeHandler.responseStream
       ? this.createStreamServiceMethod(methodHandler)
@@ -220,17 +228,23 @@ export class ServerGrpc extends Server implements CustomTransportStrategy {
     };
   }
 
-  public createStreamDuplexMethod(methodHandler: Function) {
-    return async (call: GrpcCall) => {
+  public createRequestStreamMethod(
+    methodHandler: Function,
+    isResponseStream: boolean,
+  ) {
+    return async (
+      call: GrpcCall,
+      callback: (err: unknown, value: unknown) => void,
+    ) => {
       const req = new Subject<any>();
       call.on('data', (m: any) => req.next(m));
       call.on('error', (e: any) => {
         // Check if error means that stream ended on other end
-        if (
-          String(e)
-            .toLowerCase()
-            .indexOf('cancelled') > -1
-        ) {
+        const isCancelledError = String(e)
+          .toLowerCase()
+          .indexOf('cancelled');
+
+        if (isCancelledError) {
           call.end();
           return;
         }
@@ -241,23 +255,49 @@ export class ServerGrpc extends Server implements CustomTransportStrategy {
 
       const handler = methodHandler(req.asObservable());
       const res = this.transformToObservable(await handler);
-      await res
-        .pipe(
-          takeUntil(fromEvent(call as any, CANCEL_EVENT)),
-          catchError(err => {
-            call.emit('error', err);
-            return EMPTY;
-          }),
-        )
-        .forEach(m => call.write(m));
-
-      call.end();
+      if (isResponseStream) {
+        await res
+          .pipe(
+            takeUntil(fromEvent(call as any, CANCEL_EVENT)),
+            catchError(err => {
+              call.emit('error', err);
+              return EMPTY;
+            }),
+          )
+          .forEach(m => call.write(m));
+
+        call.end();
+      } else {
+        const response = await res
+          .pipe(
+            takeUntil(fromEvent(call as any, CANCEL_EVENT)),
+            catchError(err => {
+              callback(err, null);
+              return EMPTY;
+            }),
+          )
+          .toPromise();
+
+        if (typeof response !== 'undefined') {
+          callback(null, response);
+        }
+      }
     };
   }
 
-  public createStreamCallMethod(methodHandler: Function) {
-    return async (call: GrpcCall) => {
-      methodHandler(call);
+  public createStreamCallMethod(
+    methodHandler: Function,
+    isResponseStream: boolean,
+  ) {
+    return async (
+      call: GrpcCall,
+      callback: (err: unknown, value: unknown) => void,
+    ) => {
+      if (isResponseStream) {
+        methodHandler(call);
+      } else {
+        methodHandler(call, callback);
+      }
     };
   }
 

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

@@ -6,7 +6,6 @@ import { of } from 'rxjs';
 import * as sinon from 'sinon';
 import { InvalidGrpcPackageException } from '../../errors/invalid-grpc-package.exception';
 import { ServerGrpc } from '../../server/server-grpc';
-import { ClientGrpcProxy } from '../../client';
 
 class NoopLogger extends Logger {
   log(message: any, context?: string): void {}
@@ -336,9 +335,9 @@ describe('ServerGrpc', () => {
     });
     describe('when request is a stream', () => {
       describe('when stream type is RX_STREAMING', () => {
-        it('should call "createStreamDuplexMethod"', () => {
+        it('should call "createRequestStreamMethod"', () => {
           const cln = sinon.spy();
-          const spy = sinon.spy(server, 'createStreamDuplexMethod');
+          const spy = sinon.spy(server, 'createRequestStreamMethod');
           server.createServiceMethod(
             cln,
             { requestStream: true } as any,
@@ -428,17 +427,17 @@ describe('ServerGrpc', () => {
     });
   });
 
-  describe('createStreamDuplexMethod', () => {
+  describe('createRequestStreamMethod', () => {
     it('should wrap call into Subject', () => {
       const handler = sinon.spy();
-      const fn = server.createStreamDuplexMethod(handler);
+      const fn = server.createRequestStreamMethod(handler, false);
       const call = {
         on: (event, callback) => callback(),
         off: sinon.spy(),
         end: sinon.spy(),
         write: sinon.spy(),
       };
-      fn(call as any);
+      fn(call as any, sinon.spy());
 
       expect(handler.called).to.be.true;
     });
@@ -447,9 +446,9 @@ describe('ServerGrpc', () => {
   describe('createStreamCallMethod', () => {
     it('should pass through to "methodHandler"', () => {
       const handler = sinon.spy();
-      const fn = server.createStreamCallMethod(handler);
+      const fn = server.createStreamCallMethod(handler, false);
       const args = [1, 2, 3];
-      fn(args as any);
+      fn(args as any, sinon.spy());
 
       expect(handler.calledWith(args)).to.be.true;
     });