LongPoll

.eslintrc.json

@@ -22,7 +22,6 @@
     "babel/no-invalid-this": 2,
     "array-callback-return": 2,
     "valid-typeof": 2,
-    "arrow-body-style": 2,
     "react/prop-types": 0,
     "no-var": 2,
     "linebreak-style": [2, "unix"],

src/client.ts

@@ -11,6 +11,7 @@ import { StableWSConnection } from './connection';
 import { isValidEventType } from './events';
 import { JWTUserToken, DevToken, CheckSignature } from './signing';
 import { TokenManager } from './token_manager';
+import { WSConnectionFallback } from './connection_fallback';
 import {
   isFunction,
   isOwnUserBaseProperty,
@@ -20,6 +21,7 @@ import {
   randomId,
   sleep,
   retryInterval,
+  isWSFailure,
 } from './utils';
 
 import {
@@ -146,7 +148,6 @@ export class StreamChat<
   cleaningIntervalRef?: NodeJS.Timeout;
   clientID?: string;
   configs: Configs<CommandType>;
-  connectionID?: string;
   key: string;
   listeners: {
     [key: string]: Array<
@@ -185,9 +186,12 @@ export class StreamChat<
     MessageType,
     ReactionType
   > | null;
+  wsFallback?: WSConnectionFallback;
   wsPromise: ConnectAPIResponse<ChannelType, CommandType, UserType> | null;
   consecutiveFailures: number;
   insightMetrics: InsightMetrics;
+  defaultWSTimeoutWithFallback: number;
+  defaultWSTimeout: number;
 
   /**
    * Initialize a client
@@ -273,6 +277,9 @@ export class StreamChat<
     this.consecutiveFailures = 0;
     this.insightMetrics = new InsightMetrics();
 
+    this.defaultWSTimeoutWithFallback = 6000;
+    this.defaultWSTimeout = 15000;
+
     /**
      * logger function should accept 3 parameters:
      * @param logLevel string
@@ -433,7 +440,9 @@ export class StreamChat<
     this.wsBaseURL = this.baseURL.replace('http', 'ws').replace(':3030', ':8800');
   }
 
-  _hasConnectionID = () => Boolean(this.wsConnection?.connectionID);
+  _getConnectionID = () => this.wsConnection?.connectionID || this.wsFallback?.connectionID;
+
+  _hasConnectionID = () => Boolean(this._getConnectionID());
 
   /**
    * connectUser - Set the current user and open a WebSocket connection
@@ -534,17 +543,14 @@ export class StreamChat<
    * @param timeout Max number of ms, to wait for close event of websocket, before forcefully assuming succesful disconnection.
    *                https://developer.mozilla.org/en-US/docs/Web/API/CloseEvent
    */
-  closeConnection = (timeout?: number) => {
+  closeConnection = async (timeout?: number) => {
     if (this.cleaningIntervalRef != null) {
       clearInterval(this.cleaningIntervalRef);
       this.cleaningIntervalRef = undefined;
     }
 
-    if (!this.wsConnection) {
-      return Promise.resolve();
-    }
-
-    return this.wsConnection.disconnect(timeout);
+    await Promise.all([this.wsConnection?.disconnect(timeout), this.wsFallback?.disconnect(timeout)]);
+    return Promise.resolve();
   };
 
   /**
@@ -555,7 +561,7 @@ export class StreamChat<
       throw Error('User is not set on client, use client.connectUser or client.connectAnonymousUser instead');
     }
 
-    if (this.wsConnection?.isHealthy && this._hasConnectionID()) {
+    if ((this.wsConnection?.isHealthy || this.wsFallback?.isHealthy()) && this._hasConnectionID()) {
       this.logger('info', 'client:openConnection() - openConnection called twice, healthy connection already exists', {
         tags: ['connection', 'client'],
       });
@@ -736,7 +742,7 @@ export class StreamChat<
     // reset client state
     this.state = new ClientState();
     // reset token manager
-    this.tokenManager.reset();
+    setTimeout(this.tokenManager.reset); // delay reseting to use token for disconnect calls
 
     // close the WS connection
     return closePromise;
@@ -1022,6 +1028,7 @@ export class StreamChat<
       this._logApiError(type, url, e);
       this.consecutiveFailures += 1;
       if (e.response) {
+        /** connection_fallback depends on this token expiration logic */
         if (e.response.data.code === chatCodes.TOKEN_EXPIRED && !this.tokenManager.isStatic()) {
           if (this.consecutiveFailures > 1) {
             await sleep(retryInterval(this.consecutiveFailures));
@@ -1099,6 +1106,8 @@ export class StreamChat<
   dispatchEvent = (
     event: Event<AttachmentType, ChannelType, CommandType, EventType, MessageType, ReactionType, UserType>,
   ) => {
+    if (!event.received_at) event.received_at = new Date();
+
     // client event handlers
     const postListenerCallbacks = this._handleClientEvent(event);
 
@@ -1130,7 +1139,6 @@ export class StreamChat<
       ReactionType,
       UserType
     >;
-    event.received_at = new Date();
     this.dispatchEvent(event);
   };
 
@@ -1369,13 +1377,9 @@ export class StreamChat<
   };
 
   recoverState = async () => {
-    this.logger(
-      'info',
-      `client:recoverState() - Start of recoverState with connectionID ${this.wsConnection?.connectionID}`,
-      {
-        tags: ['connection'],
-      },
-    );
+    this.logger('info', `client:recoverState() - Start of recoverState with connectionID ${this._getConnectionID()}`, {
+      tags: ['connection'],
+    });
 
     const cids = Object.keys(this.activeChannels);
     if (cids.length && this.recoverStateOnReconnect) {
@@ -1435,7 +1439,22 @@ export class StreamChat<
       ReactionType
     >({ client: this });
 
-    return await this.wsConnection.connect();
+    try {
+      // if WSFallback is enabled, ws connect should timeout faster so fallback can try
+      return await this.wsConnection.connect(
+        this.options.enableWSFallback ? this.defaultWSTimeoutWithFallback : this.defaultWSTimeout,
+      );
+    } catch (err) {
+      // run fallback only if it's WS/Network error and not a normal API error
+      if (this.options.enableWSFallback && isWSFailure(err)) {
+        this.wsConnection._destroyCurrentWSConnection();
+        this.wsConnection.disconnect().then(); // close WS so no retry
+        this.wsFallback = new WSConnectionFallback({ client: (this as unknown) as StreamChat });
+        return await this.wsFallback.connect();
+      }
+
+      throw err;
+    }
   }
 
   /**
@@ -1671,7 +1690,7 @@ export class StreamChat<
    *
    */
   setLocalDevice(device: BaseDeviceFields) {
-    if (this.wsConnection) {
+    if (this.wsConnection || this.wsFallback) {
       throw new Error('you can only set device before opening a websocket connection');
     }
 
@@ -2538,7 +2557,7 @@ export class StreamChat<
         user_id: this.userID,
         ...options.params,
         api_key: this.key,
-        connection_id: this.wsConnection?.connectionID,
+        connection_id: this._getConnectionID(),
       },
       headers: {
         Authorization: token,
@@ -2569,6 +2588,22 @@ export class StreamChat<
     }, 500);
   }
 
+  /**
+   * encode ws url payload
+   * @private
+   * @returns json string
+   */
+  _buildWSPayload = (client_request_id?: string) => {
+    return JSON.stringify({
+      user_id: this.userID,
+      user_details: this._user,
+      user_token: this.tokenManager.getToken(),
+      server_determines_connection_id: true,
+      device: this.options.device,
+      client_request_id,
+    });
+  };
+
   verifyWebhook(requestBody: string, xSignature: string) {
     return !!this.secret && CheckSignature(requestBody, this.secret, xSignature);
   }

src/connection.ts

@@ -48,6 +48,7 @@ export class StableWSConnection<
   pingInterval: number;
   healthCheckTimeoutRef?: NodeJS.Timeout;
   isConnecting: boolean;
+  isDisconnected: boolean;
   isHealthy: boolean;
   isResolved?: boolean;
   lastEvent: Date | null;
@@ -75,6 +76,8 @@ export class StableWSConnection<
     this.totalFailures = 0;
     /** We only make 1 attempt to reconnect at the same time.. */
     this.isConnecting = false;
+    /** To avoid reconnect if client is disconnected */
+    this.isDisconnected = false;
     /** Boolean that indicates if the connection promise is resolved */
     this.isResolved = false;
     /** Boolean that indicates if we have a working connection to the server */
@@ -95,14 +98,16 @@ export class StableWSConnection<
 
   /**
    * connect - Connect to the WS URL
-   *
+   * the default 15s timeout allows between 2~3 tries
    * @return {ConnectAPIResponse<ChannelType, CommandType, UserType>} Promise that completes once the first health check message is received
    */
-  async connect() {
+  async connect(timeout = 15000) {
     if (this.isConnecting) {
       throw Error(`You've called connect twice, can only attempt 1 connection at the time`);
     }
 
+    this.isDisconnected = false;
+
     try {
       const healthCheck = await this._connect();
       this.consecutiveFailures = 0;
@@ -128,7 +133,7 @@ export class StableWSConnection<
       }
     }
 
-    return await this._waitForHealthy();
+    return await this._waitForHealthy(timeout);
   }
 
   /**
@@ -160,6 +165,7 @@ export class StableWSConnection<
       })(),
       (async () => {
         await sleep(timeout);
+        this.isConnecting = false;
         throw new Error(
           JSON.stringify({
             code: '',
@@ -174,20 +180,13 @@ export class StableWSConnection<
 
   /**
    * Builds and returns the url for websocket.
-   * @param reqID Unique identifier generated on client side, to help tracking apis on backend.
+   * @private
    * @returns url string
    */
-  _buildUrl = (reqID?: string) => {
-    const params = {
-      user_id: this.client.userID,
-      user_details: this.client._user,
-      user_token: this.client.tokenManager.getToken(),
-      server_determines_connection_id: true,
-      device: this.client.options.device,
-      client_request_id: reqID,
-    };
-    const qs = encodeURIComponent(JSON.stringify(params));
+  _buildUrl = () => {
+    const qs = encodeURIComponent(this.client._buildWSPayload(this.requestID));
     const token = this.client.tokenManager.getToken();
+
     return `${this.client.wsBaseURL}/connect?json=${qs}&api_key=${
       this.client.key
     }&authorization=${token}&stream-auth-type=${this.client.getAuthType()}&X-Stream-Client=${this.client.getUserAgent()}`;
@@ -201,6 +200,8 @@ export class StableWSConnection<
     this._log(`disconnect() - Closing the websocket connection for wsID ${this.wsID}`);
 
     this.wsID += 1;
+    this.isConnecting = false;
+    this.isDisconnected = true;
 
     // start by removing all the listeners
     if (this.healthCheckTimeoutRef) {
@@ -256,14 +257,14 @@ export class StableWSConnection<
    * @return {ConnectAPIResponse<ChannelType, CommandType, UserType>} Promise that completes once the first health check message is received
    */
   async _connect() {
-    if (this.isConnecting) return; // simply ignore _connect if it's currently trying to connect
+    if (this.isConnecting || this.isDisconnected) return; // simply ignore _connect if it's currently trying to connect
     this.isConnecting = true;
     this.requestID = randomId();
     this.client.insightMetrics.connectionStartTimestamp = new Date().getTime();
     try {
       await this.client.tokenManager.tokenReady();
       this._setupConnectionPromise();
-      const wsURL = this._buildUrl(this.requestID);
+      const wsURL = this._buildUrl();
       this.ws = new WebSocket(wsURL);
       this.ws.onopen = this.onopen.bind(this, this.wsID);
       this.ws.onclose = this.onclose.bind(this, this.wsID);
@@ -307,6 +308,12 @@ export class StableWSConnection<
    */
   async _reconnect(options: { interval?: number; refreshToken?: boolean } = {}): Promise<void> {
     this._log('_reconnect() - Initiating the reconnect');
+
+    if (this.isDisconnected) {
+      this._log('_reconnect() - Abort (0) since disconnect() is called');
+      return;
+    }
+
     // only allow 1 connection at the time
     if (this.isConnecting || this.isHealthy) {
       this._log('_reconnect() - Abort (1) since already connecting or healthy');