api: add websocket events to the lnd & loop apis

jamaljsr · jamaljsr · commit cac4d9319f8c · 2020-06-03T22:24:20.000-04:00
diff --git a/app/src/__mocks__/@improbable-eng/grpc-web.ts b/app/src/__mocks__/@improbable-eng/grpc-web.ts
@@ -31,4 +31,6 @@ export const grpc = {
         });
       },
     ),
+  // mock client function to simulate server-side streaming
+  client: jest.fn(),
 };
diff --git a/app/src/__stories__/StoryWrapper.tsx b/app/src/__stories__/StoryWrapper.tsx
@@ -20,6 +20,7 @@ const grpc = {
     const response: any = { toObject: () => data };
     return Promise.resolve(response);
   },
+  subscribe: () => undefined,
 };
 
 // fake the AppStorage dependency so that settings aren't shared across stories
diff --git a/app/src/api/auth.ts b/app/src/api/auth.ts
diff --git a/app/src/api/base.ts b/app/src/api/base.ts
@@ -0,0 +1,59 @@
+import { Emitter, EventKey, EventMap, EventReceiver } from 'types/emitter';
+import { EventEmitter } from 'events';
+
+/**
+ * A shared base class containing logic for storing the API credentials
+ */
+class BaseApi<T extends EventMap> implements Emitter<T> {
+  private _credentials = '';
+  /** an internal event emitter used to track event subscriptions  */
+  private _emitter = new EventEmitter();
+
+  /**
+   * Returns a metadata object containing authorization info that was
+   * previous set if any
+   */
+  protected get _meta() {
+    return this._credentials
+      ? { authorization: `Basic ${this._credentials}` }
+      : undefined;
+  }
+
+  /**
+   * Sets the credentials to use for all API requests
+   * @param credentials the base64 encoded password
+   */
+  setCredentials(credentials: string) {
+    this._credentials = credentials;
+  }
+
+  /**
+   * Subscribe to have a handler function called when an event is emitted
+   * @param eventName the name of the event
+   * @param handler the function to call when the event is emitted
+   */
+  on<K extends EventKey<T>>(eventName: K, handler: EventReceiver<T[K]>) {
+    this._emitter.on(eventName, handler);
+  }
+
+  /**
+   * Unsubscribes the handler for the provided event
+   * @param eventName the name of the event
+   * @param handler the function that was used to subscribe to the event
+   */
+  off<K extends EventKey<T>>(eventName: K, handler: EventReceiver<T[K]>) {
+    this._emitter.off(eventName, handler);
+  }
+
+  /**
+   * Call all of the subscribed handlers for an event with the supplied argument
+   * @param eventName the name of the event
+   * @param params the argument to pass to the handlers that are subscribed
+   * to this event
+   */
+  emit<K extends EventKey<T>>(eventName: K, params: T[K]) {
+    this._emitter.emit(eventName, params);
+  }
+}
+
+export default BaseApi;
diff --git a/app/src/api/grpc.ts b/app/src/api/grpc.ts
@@ -1,7 +1,10 @@
 import { grpc } from '@improbable-eng/grpc-web';
 import { ProtobufMessage } from '@improbable-eng/grpc-web/dist/typings/message';
 import { Metadata } from '@improbable-eng/grpc-web/dist/typings/metadata';
-import { UnaryMethodDefinition } from '@improbable-eng/grpc-web/dist/typings/service';
+import {
+  MethodDefinition,
+  UnaryMethodDefinition,
+} from '@improbable-eng/grpc-web/dist/typings/service';
 import { DEV_HOST } from 'config';
 import { AuthenticationError } from 'util/errors';
 import { grpcLog as log } from 'util/log';
@@ -19,30 +22,62 @@ class GrpcClient {
     metadata?: Metadata.ConstructorArg,
   ): Promise<TRes> {
     return new Promise((resolve, reject) => {
-      log.debug(
-        `Request: ${methodDescriptor.service.serviceName}.${methodDescriptor.methodName}`,
-      );
-      log.debug(` - req: `, request.toObject());
+      const method = `${methodDescriptor.methodName}`;
+      log.debug(`${method} request`, request.toObject());
       grpc.unary(methodDescriptor, {
         host: DEV_HOST,
         request,
         metadata,
         onEnd: ({ status, statusMessage, headers, message, trailers }) => {
-          log.debug(' - status', status, statusMessage);
-          log.debug(' - headers', headers);
+          log.debug(`${method} status`, status, statusMessage);
+          log.debug(`${method} headers`, headers);
           if (status === grpc.Code.OK && message) {
-            log.debug(' - message', message.toObject());
+            log.debug(`${method} message`, message.toObject());
             resolve(message as TRes);
           } else if (status === grpc.Code.Unauthenticated) {
             reject(new AuthenticationError(`${statusMessage}`));
           } else {
             reject(new Error(`${status}: ${statusMessage}`));
           }
-          log.debug(' - trailers', trailers);
+          log.debug(`${method} trailers`, trailers);
         },
       });
     });
   }
+
+  /**
+   * Subscribes to a GRPC server-streaming endpoint and executes the `onMessage` handler
+   * when a new message is received from the server
+   * @param methodDescriptor the GRPC method to call on the service
+   * @param request the GRPC request message to send
+   * @param onMessage the callback function to execute when a new message is received
+   * @param metadata headers to include with the request
+   */
+  subscribe<TReq extends ProtobufMessage, TRes extends ProtobufMessage>(
+    methodDescriptor: MethodDefinition<TReq, TRes>,
+    request: TReq,
+    onMessage: (res: TRes) => void,
+    metadata?: Metadata.ConstructorArg,
+  ) {
+    const method = `${methodDescriptor.methodName}`;
+    const client = grpc.client(methodDescriptor, {
+      host: DEV_HOST,
+      transport: grpc.WebsocketTransport(),
+    });
+    client.onHeaders(headers => {
+      log.debug(`${method} - headers`, headers);
+    });
+    client.onMessage(message => {
+      log.debug(`${method} - message`, message.toObject());
+      onMessage(message as TRes);
+    });
+    client.onEnd((status, statusMessage, trailers) => {
+      log.debug(`${method} - status`, status, statusMessage);
+      log.debug(`${method} - trailers`, trailers);
+    });
+    client.start(metadata);
+    client.send(request);
+  }
 }
 
 export default GrpcClient;
diff --git a/app/src/api/lnd.ts b/app/src/api/lnd.ts
@@ -1,12 +1,19 @@
 import * as LND from 'types/generated/lnd_pb';
 import { Lightning } from 'types/generated/lnd_pb_service';
-import AuthenticatedApi from './auth';
+import BaseApi from './base';
 import GrpcClient from './grpc';
 
+/** the names and argument types for the subscription events */
+interface LndEvents {
+  transaction: LND.Transaction.AsObject;
+  channel: LND.ChannelEventUpdate.AsObject;
+  invoice: LND.Invoice.AsObject;
+}
+
 /**
  * An API wrapper to communicate with the LND node via GRPC
  */
-class LndApi extends AuthenticatedApi {
+class LndApi extends BaseApi<LndEvents> {
   private _grpc: GrpcClient;
 
   constructor(grpc: GrpcClient) {
diff --git a/app/src/api/loop.ts b/app/src/api/loop.ts
@@ -2,13 +2,18 @@ import * as LOOP from 'types/generated/loop_pb';
 import { SwapClient } from 'types/generated/loop_pb_service';
 import { Quote } from 'types/state';
 import Big from 'big.js';
-import AuthenticatedApi from './auth';
+import BaseApi from './base';
 import GrpcClient from './grpc';
 
+/** the names and argument types for the subscription events */
+interface LoopEvents {
+  monitor: LOOP.SwapStatus.AsObject;
+}
+
 /**
  * An API wrapper to communicate with the Loop daemon via GRPC
  */
-class LoopApi extends AuthenticatedApi {
+class LoopApi extends BaseApi<LoopEvents> {
   private _grpc: GrpcClient;
 
   constructor(grpc: GrpcClient) {
diff --git a/app/src/types/emitter.ts b/app/src/types/emitter.ts
@@ -0,0 +1,10 @@
+export type EventMap = Record<string, any>;
+export type EventKey<T extends EventMap> = string & keyof T;
+export type EventReceiver<T> = (params: T) => void;
+
+/** Generic interface to represent a type-safe pubsub Emitter object */
+export interface Emitter<T extends EventMap> {
+  on<K extends EventKey<T>>(eventName: K, handler: EventReceiver<T[K]>): void;
+  off<K extends EventKey<T>>(eventName: K, handler: EventReceiver<T[K]>): void;
+  emit<K extends EventKey<T>>(eventName: K, params: T[K]): void;
+}
