fix(rpc): close streaming calls on shutdown

lib/grpc/GrpcServer.ts

@@ -12,6 +12,7 @@ import { exists, readFile, writeFile } from '../utils/fsUtils';
 
 class GrpcServer {
   private server: Server;
+  private grpcService: GrpcService;
 
   constructor(private logger: Logger, service: Service) {
     this.server = new grpc.Server();
@@ -45,6 +46,8 @@ class GrpcServer {
     this.server.addService(HashResolverService, {
       resolveHash: grpcService.resolveHash,
     });
+
+    this.grpcService = grpcService;
   }
 
   /**
@@ -90,6 +93,7 @@ class GrpcServer {
    * Stop listening for requests
    */
   public close = (): Promise<void> => {
+    this.grpcService.closeStreams();
     return new Promise((resolve) => {
       this.server.tryShutdown(() => {
         this.logger.info('GRPC server completed shutdown');

lib/grpc/GrpcService.ts

@@ -94,6 +94,9 @@ const createPlaceOrderEvent = (e: PlaceOrderEvent) => {
 
 /** Class containing the available RPC methods for XUD */
 class GrpcService {
+  /** The set of active streaming calls. */
+  private streams: Set<grpc.ServerWriteableStream<any>> = new Set<grpc.ServerWriteableStream<any>>();
+
   /** Create an instance of available RPC methods and bind all exposed functions. */
   constructor(private logger: Logger, private service: Service) {}
 
@@ -152,6 +155,22 @@ class GrpcService {
     return grpcError;
   }
 
+  /** Closes and removes all active streaming calls. */
+  public closeStreams = () => {
+    this.streams.forEach((stream) => {
+      stream.end();
+    });
+    this.streams.clear();
+  }
+
+  /** Adds an active streaming call and adds a listener to remove it if it is cancelled. */
+  private addStream = (stream: grpc.ServerWriteableStream<any>) => {
+    this.streams.add(stream);
+    stream.once('cancelled', () => {
+      this.streams.delete(stream);
+    });
+  }
+
   /**
    * See [[Service.addCurrency]]
    */
@@ -508,6 +527,7 @@ class GrpcService {
     this.service.subscribeAddedOrders((order: Order) => {
       call.write(createOrder(order));
     });
+    this.addStream(call);
   }
 
   /*
@@ -523,6 +543,7 @@ class GrpcService {
       orderRemoval.setIsOwnOrder(order.localId !== undefined);
       call.write(orderRemoval);
     });
+    this.addStream(call);
   }
 
   /*
@@ -532,6 +553,7 @@ class GrpcService {
     this.service.subscribeSwaps((result: SwapResult) => {
       call.write(createSwapResult(result));
     });
+    this.addStream(call);
   }
 }