GRPC Streaming tests updated with support both Streaming decorators

- RX Streaming test
- Call streaming test

Author: anton-alation

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

@@ -52,7 +52,7 @@ describe('GRPC transport', () => {
       .expect(200, { result: 15 });
   });
 
-  it('GRPC Sending and receiving Stream', async () => {
+  it('GRPC Sending and receiving Stream from RX handler', async () => {
 
     const callHandler = client.SumStream();
 
@@ -77,6 +77,31 @@ describe('GRPC transport', () => {
 
   });
 
+  it('GRPC Sending and receiving Stream from Call Passthrough handler', async () => {
+
+    const callHandler = client.SumStreamPass();
+
+    callHandler.on('data', (msg: number) => {
+      // Do deep comparison (to.eql)
+      expect(msg).to.eql({result: 15});
+      callHandler.cancel();
+    });
+
+    callHandler.on('error', (err: any) => {
+      // We want to fail only on real errors while Cancellation error
+      // is expected
+      if (String(err).toLowerCase().indexOf('cancelled') === -1) {
+        fail('gRPC Stream error happened, error: ' + err);
+      }
+    });
+
+    return new Promise((resolve, reject) => {
+      callHandler.write({data: [1, 2, 3, 4, 5]});
+      setTimeout(() => resolve(), 1000);
+    });
+
+  });
+
   after(async () => {
     await app.close();
     client.close;

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

@@ -1,5 +1,8 @@
 import { Body, Controller, HttpCode, Post } from '@nestjs/common';
-import { Client, ClientGrpc, GrpcMethod, GrpcStream, Transport } from '@nestjs/microservices';
+import {
+  Client, ClientGrpc, GrpcMethod,
+  GrpcStreamMethod, GrpcStreamCall, Transport,
+} from '@nestjs/microservices';
 import { join } from 'path';
 import { Observable, of } from 'rxjs';
 
@@ -28,8 +31,24 @@ export class GrpcController {
     });
   }
 
-  @GrpcStream('Math')
-  async sumStream(stream: any) {
+  @GrpcStreamMethod('Math')
+  async sumStream(messages: Observable<any>): Promise<any> {
+    // Form a resulting promise
+    return new Promise<any>((resolve, reject) => {
+      // Subscribe for a message to for a test answer
+      messages.subscribe(msg => {
+        // Resolve with reduce function
+        resolve({
+          result: msg.data.reduce((a, b) => a + b),
+        });
+      }, err => {
+        reject(err);
+      });
+    });
+  }
+
+  @GrpcStreamCall('Math')
+  async sumStreamPass(stream: any) {
     stream.on('data', (msg: any) => {
       stream.write({result: msg.data.reduce((a, b) => a + b)});
     });

integration/microservices/src/grpc/math.proto

@@ -5,6 +5,7 @@ package math;
 service Math {
   rpc Sum (RequestSum) returns (SumResult);
   rpc SumStream(stream RequestSum) returns(stream SumResult);
+  rpc SumStreamPass(stream RequestSum) returns(stream SumResult);
 }
 
 message SumResult {