Merge pull request #75 from hyperledger/ws-connect

Allow custom app logic to trigger when a websocket connects/reconnects

Author: peterbroadhurst

lib/firefly.ts

@@ -79,6 +79,7 @@ import {
   FireFlyDeleteOptions,
   FireFlyTokenApprovalFilter,
   FireFlyTokenApprovalResponse,
+  FireFlyWebSocketConnectCallback,
 } from './interfaces';
 import { FireFlyWebSocket, FireFlyWebSocketCallback } from './websocket';
 import HttpBase, { mapConfig } from './http';
@@ -598,6 +599,7 @@ export default class FireFly extends HttpBase {
     subscriptions: string | string[] | FireFlySubscriptionBase,
     callback: FireFlyWebSocketCallback,
     socketOptions?: WebSocket.ClientOptions | http.ClientRequestArgs,
+    afterConnect?: FireFlyWebSocketConnectCallback,
   ): FireFlyWebSocket {
     const options: FireFlyWebSocketOptions = {
       host: this.options.websocket.host,
@@ -609,6 +611,7 @@ export default class FireFly extends HttpBase {
       reconnectDelay: this.options.websocket.reconnectDelay,
       heartbeatInterval: this.options.websocket.heartbeatInterval,
       socketOptions: socketOptions,
+      afterConnect: afterConnect,
     };
 
     const handler: FireFlyWebSocketCallback = (socket, event) => {

lib/interfaces.ts

@@ -61,6 +61,14 @@ export interface FireFlyOptions extends FireFlyOptionsInput {
   };
 }
 
+export interface FireFlyWebSocketSender {
+  send: (json: JSON) => void;
+}
+
+export interface FireFlyWebSocketConnectCallback {
+  (sender: FireFlyWebSocketSender): void | Promise<void>;
+}
+
 export interface FireFlyWebSocketOptions {
   host: string;
   namespace: string;
@@ -72,6 +80,7 @@ export interface FireFlyWebSocketOptions {
   reconnectDelay: number;
   heartbeatInterval: number;
   socketOptions?: WebSocket.ClientOptions | http.ClientRequestArgs;
+  afterConnect?: FireFlyWebSocketConnectCallback;
 }
 
 // Namespace

lib/websocket.ts

@@ -26,7 +26,7 @@ export class FireFlyWebSocket {
   private readonly logger = new Logger(FireFlyWebSocket.name);
 
   private socket?: WebSocket;
-  private closed = false;
+  private closed? = () => {};
   private pingTimer?: NodeJS.Timeout;
   private disconnectTimer?: NodeJS.Timeout;
   private reconnectTimer?: NodeJS.Timeout;
@@ -61,7 +61,7 @@ export class FireFlyWebSocket {
       auth,
       handshakeTimeout: this.options.heartbeatInterval,
     }));
-    this.closed = false;
+    this.closed = undefined;
 
     socket
       .on('open', () => {
@@ -83,13 +83,17 @@ export class FireFlyWebSocket {
           );
           this.logger.log(`Started listening on subscription ${this.options.namespace}:${name}`);
         }
+        if (this.options?.afterConnect !== undefined) {
+          this.options.afterConnect(this);
+        }
       })
       .on('error', (err) => {
         this.logger.error('Error', err.stack);
       })
       .on('close', () => {
         if (this.closed) {
           this.logger.log('Closed');
+          this.closed(); // do this after all logging
         } else {
           this.disconnectDetected = true;
           this.reconnect('Closed by peer');
@@ -156,6 +160,12 @@ export class FireFlyWebSocket {
     }
   }
 
+  send(json: JSON) {
+    if (this.socket !== undefined) {
+      this.socket.send(JSON.stringify(json));
+    }
+  }
+
   ack(event: FireFlyEventDelivery) {
     if (this.socket !== undefined && event.id !== undefined) {
       this.socket.send(
@@ -168,15 +178,18 @@ export class FireFlyWebSocket {
     }
   }
 
-  close() {
-    this.closed = true;
+  async close(wait?: boolean): Promise<void> {
+    const closedPromise = new Promise<void>(resolve => {
+      this.closed = resolve;
+    });
     this.clearPingTimers();
     if (this.socket) {
       try {
         this.socket.close();
       } catch (e: any) {
         this.logger.warn(`Failed to clean up websocket: ${e.message}`);
       }
+      if (wait) await closedPromise;
       this.socket = undefined;
     }
   }