fix(stream): unhandled rejection on connection (#79)

* fix(stream): unhandled rejection on connection

* refactor: removed interface for compatibility

src/client/command.handler.ts

@@ -2,7 +2,7 @@ import { StreamClient } from '../stream/stream.client';
 import { EventSubscribed, EventSubscription } from './types';
 import { Queue } from './utils/queue';
 import { CensusCommand } from '../stream';
-import { StreamClosedException } from './exceptions/stream-closed.exception';
+import { StreamResponseException } from './exceptions/stream-response.exception';
 
 interface CommandCallback<T> {
   resolve(data: T): void;
@@ -45,7 +45,7 @@ export class CommandHandler {
           this.subscriptionQueue
             .dequeue()
             .reject(
-              new StreamClosedException(
+              new StreamResponseException(
                 'Stream closed before receiving response',
               ),
             );
@@ -54,7 +54,7 @@ export class CommandHandler {
           this.recentCharacterQueue
             .dequeue()
             .reject(
-              new StreamClosedException(
+              new StreamResponseException(
                 'Stream closed before receiving response',
               ),
             );
@@ -113,10 +113,10 @@ export class CommandHandler {
     queue: Queue<CommandCallback<T>>,
     command: CensusCommand,
   ): Promise<T> {
-    return new Promise<T>((resolve, reject) => {
-      if (!this.stream.isReady)
-        reject(new StreamClosedException('Stream is closed'));
+    if (!this.stream.isReady)
+      return Promise.reject(new StreamResponseException('Stream is closed'));
 
+    return new Promise<T>((resolve, reject) => {
       this.stream
         .send(command)
         .then(() => {

src/client/exceptions/index.ts

@@ -1,3 +1,3 @@
 export * from './max-retry.exception';
-export * from './stream-closed.exception';
+export * from './stream-response.exception';
 export * from './service-id-reject.exception';

src/client/exceptions/max-retry.exception.ts

@@ -12,6 +12,6 @@ export class MaxRetryException extends Error {
       )}`,
     );
 
-    this.name = 'MaxRetryException';
+    this.name = MaxRetryException.name;
   }
 }

src/client/exceptions/service-id-reject.exception.ts

@@ -2,6 +2,6 @@ export class ServiceIdRejectException extends Error {
   constructor(message: string) {
     super(message);
 
-    this.name = 'StreamClosedException';
+    this.name = ServiceIdRejectException.name;
   }
 }

src/client/exceptions/stream-response.exception.ts

@@ -0,0 +1,7 @@
+export class StreamResponseException extends Error {
+  constructor(message: string) {
+    super(message);
+
+    this.name = StreamResponseException.name;
+  }
+}

src/client/stream.manager.ts

@@ -10,6 +10,7 @@ import Timeout = NodeJS.Timeout;
 
 export interface StreamManagerOptions extends StreamClientOptions {
   subscription?: EventSubscription;
+  reconnectDelay?: number;
 }
 
 export class StreamManager {
@@ -41,7 +42,7 @@ export class StreamManager {
   /**
    * @type {number} delay before trying to reconnect
    */
-  private reconnectDelay = 2000;
+  private readonly reconnectDelay: number;
 
   /**
    * @type {Timeout?} The reconnect timeout
@@ -77,6 +78,8 @@ export class StreamManager {
       options.subscription,
     );
 
+    this.reconnectDelay = options.reconnectDelay ?? 2000;
+
     this.prepareEventStream();
   }
 
@@ -87,7 +90,7 @@ export class StreamManager {
     /**
      * Stream closed
      */
-    this.stream.on('close', ({ code, reason }) => {
+    this.stream.on('close', (code, reason) => {
       if (!this.isStarted) {
         this.client.emit('disconnected', code, reason);
         return;
@@ -174,29 +177,27 @@ export class StreamManager {
    * Connection has done something, now we need a new one
    */
   private async reconnect(): Promise<void> {
-    try {
-      await this.stream.connect();
-    } catch (e: any) {
-      if ([403].includes(e.httpState)) {
+    if (this.reconnectTimeout) clearTimeout(this.reconnectTimeout);
+
+    this.reconnectTimeout = setTimeout(async () => {
+      try {
+        await this.stream.connect();
+      } catch (e: any) {
+        if ([403].includes(e.httpState)) {
+          this.client.emit(
+            'error',
+            new Error(`Service ID rejected while trying to reconnect.`),
+          );
+          this.disconnect();
+
+          return;
+        }
+
         this.client.emit(
-          'error',
-          new Error(`Service ID rejected while trying to reconnect.`),
+          'debug',
+          `Reconnect failed, trying again in ${this.reconnectDelay}ms.`,
         );
-        this.disconnect();
-
-        return;
       }
-
-      this.client.emit(
-        'debug',
-        `Reconnect failed, trying again in ${this.reconnectDelay}ms.`,
-      );
-
-      if (this.reconnectTimeout) clearTimeout(this.reconnectTimeout);
-      this.reconnectTimeout = setTimeout(
-        () => this.reconnect(),
-        this.reconnectDelay,
-      );
-    }
+    }, this.reconnectDelay);
   }
 }

src/client/subscription.manager.ts

@@ -3,7 +3,7 @@ import { StreamClient } from '../stream/stream.client';
 import { PS2EventNames } from '../stream/types/ps2.events';
 import { EventSubscribed, EventSubscription } from './types';
 import { CommandHandler } from './command.handler';
-import { StreamClosedException } from './exceptions';
+import { StreamResponseException } from './exceptions/stream-response.exception';
 
 export class SubscriptionManager {
   /**
@@ -59,7 +59,7 @@ export class SubscriptionManager {
       try {
         await this.commandHandler.subscribe(this.subscription);
       } catch (err) {
-        if (!(err instanceof StreamClosedException)) throw err;
+        if (!(err instanceof StreamResponseException)) throw err;
       }
     });
   }

src/rest/exceptions/census-rest.exception.ts

@@ -2,7 +2,7 @@ export class CensusRestException extends Error {
   constructor(message: string) {
     super(message);
 
-    this.name = 'CensusRestException';
+    this.name = CensusRestException.name;
   }
 
   hasNoData(): boolean {

src/rest/exceptions/census-server.error.ts

@@ -4,7 +4,7 @@ export class CensusServerError extends Error {
   constructor(data: any) {
     super(data.errorMessage ?? 'Error message not supplied');
 
-    this.name = 'CensusServerError';
+    this.name = CensusServerError.name;
     this.code = data.errorCode;
   }
 }

src/stream/exceptions/index.ts

@@ -0,0 +1,2 @@
+export * from './stream-closed.exception';
+export * from './stream-destroyed.exception';

src/stream/exceptions/stream-closed.exception.ts

@@ -0,0 +1,7 @@
+export class StreamClosedException extends Error {
+  constructor(readonly code: number, readonly reason?: string) {
+    super(`Stream closed with code ${code}: ${reason}`);
+
+    this.name = StreamClosedException.name;
+  }
+}

src/stream/exceptions/stream-destroyed.exception.ts

@@ -0,0 +1,7 @@
+export class StreamDestroyedException extends Error {
+  constructor() {
+    super('Stream destroyed before connection could be made');
+
+    this.name = StreamDestroyedException.name;
+  }
+}

src/stream/index.ts

@@ -5,3 +5,4 @@ import {
 
 export { Client, ClientOptions };
 export * from './types';
+export * from './exceptions';

src/stream/stream.client.ts

@@ -3,6 +3,8 @@ import WebSocket, { ClientOptions, Data } from 'ws';
 import { PS2Environment } from '../types/ps2.options';
 import { CensusMessage } from './types/messages.types';
 import { CensusCommand } from './types/command.types';
+import { StreamDestroyedException } from './exceptions/stream-destroyed.exception';
+import { StreamClosedException } from './exceptions/stream-closed.exception';
 import Timeout = NodeJS.Timeout;
 
 enum State {
@@ -23,7 +25,7 @@ export interface StreamClientOptions {
 type StreamClientEvents = {
   ready: () => void;
   destroyed: () => void;
-  close: (info: { code: number; reason: string }) => void;
+  close: (code: number, reason?: string) => void;
   error: (err: Error) => void;
   warn: (err: Error) => void;
   debug: (info: string) => void;
@@ -130,33 +132,34 @@ export class StreamClient extends EventEmitter<StreamClientEvents> {
       return Promise.resolve();
 
     return new Promise((resolve, reject) => {
-      const accept = () => {
+      const ready = () => {
         cleanup();
         resolve();
       };
 
-      const decline = (...e: any[]) => {
+      const destroyed = () => {
         cleanup();
-        reject(e);
+        reject(new StreamDestroyedException());
+      };
+
+      const closed = (code: number, reason?: string) => {
+        cleanup();
+        reject(new StreamClosedException(code, reason));
       };
 
       const cleanup = () => {
-        this.removeListener('ready', accept);
-        this.removeListener('destroyed', decline);
-        this.removeListener('close', decline);
+        this.removeListener('ready', ready)
+          .removeListener('destroyed', destroyed)
+          .removeListener('close', closed);
       };
 
-      this.once('ready', accept);
-      this.once('destroyed', decline);
-      this.once('close', decline);
+      this.once('ready', ready)
+        .once('destroyed', destroyed)
+        .once('close', closed);
 
-      if (this.connection && this.connection.readyState === WebSocket.OPEN) {
+      if (this.connection && this.state == State.READY) {
         this.emit('debug', `Open connection found, continuing operations.`);
 
-        // Assume everything is fine
-        this.state = State.READY;
-        this.emit('ready');
-
         return;
       }
 
@@ -176,10 +179,10 @@ export class StreamClient extends EventEmitter<StreamClientEvents> {
         this.wsOptions,
       ));
 
-      ws.on('open', this.onOpen.bind(this));
-      ws.on('message', this.onMessage.bind(this));
-      ws.on('close', this.onClose.bind(this));
-      ws.on('error', this.onError.bind(this));
+      ws.on('open', this.onOpen.bind(this))
+        .on('message', this.onMessage.bind(this))
+        .on('close', this.onClose.bind(this))
+        .on('error', this.onError.bind(this));
     });
   }
 
@@ -263,7 +266,7 @@ export class StreamClient extends EventEmitter<StreamClientEvents> {
     this.cleanupConnection();
 
     this.state = State.DISCONNECTED;
-    this.emit('close', { code, reason: reason.toString() });
+    this.emit('close', code, reason.length ? reason.toString() : undefined);
   }
 
   /**