sample(grpc): update grpc sample with client streaming

Author: kamilmysliwiec

sample/04-grpc/src/hero/hero.controller.ts

@@ -1,15 +1,25 @@
-import { Controller, Get, Inject, OnModuleInit } from '@nestjs/common';
-import { ClientGrpc, GrpcMethod } from '@nestjs/microservices';
-import { Observable } from 'rxjs';
+import { Controller, Get, Inject, OnModuleInit, Param } from '@nestjs/common';
+import {
+  ClientGrpc,
+  GrpcMethod,
+  GrpcStreamMethod,
+} from '@nestjs/microservices';
+import { Observable, ReplaySubject, Subject } from 'rxjs';
+import { toArray } from 'rxjs/operators';
 import { HeroById } from './interfaces/hero-by-id.interface';
 import { Hero } from './interfaces/hero.interface';
 
 interface HeroService {
-  findOne(data: HeroById): Observable<any>;
+  findOne(data: HeroById): Observable<Hero>;
+  findMany(upstream: Observable<HeroById>): Observable<Hero>;
 }
 
-@Controller()
+@Controller('hero')
 export class HeroController implements OnModuleInit {
+  private readonly items: Hero[] = [
+    { id: 1, name: 'John' },
+    { id: 2, name: 'Doe' },
+  ];
   private heroService: HeroService;
 
   constructor(@Inject('HERO_PACKAGE') private readonly client: ClientGrpc) {}
@@ -19,16 +29,37 @@ export class HeroController implements OnModuleInit {
   }
 
   @Get()
-  execute(): Observable<any> {
-    return this.heroService.findOne({ id: 1 });
+  getMany(): Observable<Hero[]> {
+    const ids$ = new ReplaySubject<HeroById>();
+    ids$.next({ id: 1 });
+    ids$.next({ id: 2 });
+    ids$.complete();
+
+    const stream = this.heroService.findMany(ids$.asObservable());
+    return stream.pipe(toArray());
+  }
+
+  @Get(':id')
+  getById(@Param('id') id: string): Observable<Hero> {
+    return this.heroService.findOne({ id: +id });
   }
 
   @GrpcMethod('HeroService')
   findOne(data: HeroById): Hero {
-    const items: Hero[] = [
-      { id: 1, name: 'John' },
-      { id: 2, name: 'Doe' },
-    ];
-    return items.find(({ id }) => id === data.id);
+    return this.items.find(({ id }) => id === data.id);
+  }
+
+  @GrpcStreamMethod('HeroService')
+  findMany(data$: Observable<HeroById>): Observable<Hero> {
+    const hero$ = new Subject<Hero>();
+
+    const onNext = (heroById: HeroById) => {
+      const item = this.items.find(({ id }) => id === heroById.id);
+      hero$.next(item);
+    };
+    const onComplete = () => hero$.complete();
+    data$.subscribe(onNext, null, onComplete);
+
+    return hero$.asObservable();
   }
 }

sample/04-grpc/src/hero/hero.proto

@@ -4,6 +4,7 @@ package hero;
 
 service HeroService {
   rpc FindOne (HeroById) returns (Hero) {}
+  rpc FindMany (stream HeroById) returns (stream Hero) {}
 }
 
 message HeroById {