From 7ab559063e043d32fd1335fe715314041b3e4374 Mon Sep 17 00:00:00 2001
From: ThisIsEsh <13537179+ThisIsEsh@users.noreply.github.com>
Date: Thu, 4 Apr 2024 20:18:57 +0300
Subject: [PATCH 1/3] Change sending commands logic, rely on transport state
 instead of client connected state

---
 src/centrifuge.ts   | 43 ++++++++++++++++++-------------------------
 src/subscription.ts | 31 ++++++++++++++++++-------------
 2 files changed, 36 insertions(+), 38 deletions(-)

diff --git a/src/centrifuge.ts b/src/centrifuge.ts
index 5fb29e62..1501e394 100644
--- a/src/centrifuge.ts
+++ b/src/centrifuge.ts
@@ -65,6 +65,7 @@ export class UnauthorizedError extends Error {
 /** Centrifuge is a Centrifuge/Centrifugo bidirectional client. */
 export class Centrifuge extends (EventEmitter as new () => TypedEventEmitter<ClientEvents>) {
   state: State;
+  private _transportIsOpen: boolean;
   private _endpoint: string | Array<TransportEndpoint>;
   private _emulation: boolean;
   private _transports: any[];
@@ -111,6 +112,7 @@ export class Centrifuge extends (EventEmitter as new () => TypedEventEmitter<Cli
   constructor(endpoint: string | Array<TransportEndpoint>, options?: Partial<Options>) {
     super();
     this.state = State.Disconnected;
+    this._transportIsOpen = false;
     this._endpoint = endpoint;
     this._emulation = false;
     this._transports = [];
@@ -409,6 +411,10 @@ export class Centrifuge extends (EventEmitter as new () => TypedEventEmitter<Cli
     })
   }
 
+  isTransportOpen() {
+    return this._transportIsOpen;
+  }
+
   private _debug(...args: any[]) {
     if (!this._debugEnabled) {
       return;
@@ -762,28 +768,11 @@ export class Centrifuge extends (EventEmitter as new () => TypedEventEmitter<Cli
     const transportId = this._nextTransportId();
     self._debug("id of transport", transportId);
     let wasOpen = false;
-
-    let optimistic = true;
-    if (this._transport.name() === 'sse') {
-      // Avoid using optimistic subscriptions with SSE/EventSource as we are sending
-      // initial data in URL params. URL is recommended to be 2048 chars max – so adding
-      // subscription data may be risky.
-      optimistic = false;
-    }
-
     const initialCommands: any[] = [];
 
     if (this._transport.emulation()) {
       const connectCommand = self._sendConnect(true);
       initialCommands.push(connectCommand);
-      if (optimistic) {
-        const subscribeCommands: any[] = self._sendSubscribeCommands(true, true);
-        for (const i in subscribeCommands) {
-          if (subscribeCommands.hasOwnProperty(i)) {
-            initialCommands.push(subscribeCommands[i]);
-          }
-        }
-      }
     }
 
     this._setNetworkEvents();
@@ -810,15 +799,14 @@ export class Centrifuge extends (EventEmitter as new () => TypedEventEmitter<Cli
         }
         wasOpen = true;
         self._debug(transport.subName(), 'transport open');
-        self._transportWasOpen = true;
         if (transport.emulation()) {
           return;
         }
+        self._transportIsOpen = true;
+        self._transportWasOpen = true;
         self.startBatching();
         self._sendConnect(false);
-        if (optimistic) {
-          self._sendSubscribeCommands(true, false);
-        }
+        self._sendSubscribeCommands();
         self.stopBatching();
       },
       onError: function (e: any) {
@@ -839,6 +827,7 @@ export class Centrifuge extends (EventEmitter as new () => TypedEventEmitter<Cli
         }
         self._debug(transport.subName(), 'transport closed');
         self._transportClosed = true;
+        self._transportIsOpen = false;
 
         let reason = 'connection closed';
         let needReconnect = true;
@@ -1261,6 +1250,9 @@ export class Centrifuge extends (EventEmitter as new () => TypedEventEmitter<Cli
       const transport = this._transport;
       this._transport = null;
       transport.close(); // Close only after setting this._transport to null to avoid recursion when calling transport close().
+      // Need to mark as closed here, because connect call may be sync called after disconnect,
+      // transport onClose callback will not be called yet
+      this._transportIsOpen = false;
       this._transportClosed = true;
       this._nextTransportId();
     } else {
@@ -1379,7 +1371,7 @@ export class Centrifuge extends (EventEmitter as new () => TypedEventEmitter<Cli
   }
 
   protected _unsubscribe(sub: Subscription) {
-    if (!this._isConnected()) {
+    if (!this.isTransportOpen()) {
       return;
     }
     const req = {
@@ -1415,7 +1407,7 @@ export class Centrifuge extends (EventEmitter as new () => TypedEventEmitter<Cli
     return this._serverSubs[channel] !== undefined;
   }
 
-  private _sendSubscribeCommands(optimistic: boolean, skipSending: boolean): any[] {
+  private _sendSubscribeCommands(): any[] {
     const commands: any[] = [];
     for (const channel in this._subs) {
       if (!this._subs.hasOwnProperty(channel)) {
@@ -1428,7 +1420,7 @@ export class Centrifuge extends (EventEmitter as new () => TypedEventEmitter<Cli
       }
       if (sub.state === SubscriptionState.Subscribing) {
         // @ts-ignore – we are hiding some symbols from public API autocompletion.
-        const cmd = sub._subscribe(optimistic, skipSending);
+        const cmd = sub._subscribe();
         if (cmd) {
           commands.push(cmd);
         }
@@ -1438,6 +1430,7 @@ export class Centrifuge extends (EventEmitter as new () => TypedEventEmitter<Cli
   }
 
   private _connectResponse(result: any) {
+    this._transportIsOpen = true;
     this._transportWasOpen = true;
     this._reconnectAttempts = 0;
     this._refreshRequired = false;
@@ -1460,7 +1453,7 @@ export class Centrifuge extends (EventEmitter as new () => TypedEventEmitter<Cli
     this._node = result.node;
 
     this.startBatching();
-    this._sendSubscribeCommands(false, false);
+    this._sendSubscribeCommands();
     this.stopBatching();
 
     const ctx: any = {
diff --git a/src/subscription.ts b/src/subscription.ts
index fe852ca8..728f521a 100644
--- a/src/subscription.ts
+++ b/src/subscription.ts
@@ -254,14 +254,16 @@ export class Subscription extends (EventEmitter as new () => TypedEventEmitter<S
     if (this._setState(SubscriptionState.Subscribing)) {
       this.emit('subscribing', { channel: this.channel, code: code, reason: reason });
     }
-    this._subscribe(false, false);
+    this._subscribe();
   }
 
-  private _subscribe(optimistic: boolean, skipSending: boolean): any {
+  private _subscribe(): any {
     // @ts-ignore – we are hiding some symbols from public API autocompletion.
     this._centrifuge._debug('subscribing on', this.channel);
 
-    if (this._centrifuge.state !== State.Connected && !optimistic) {
+    // need to check transport readiness here, because there's no point for calling getData or getToken
+    // if transport is not ready yet
+    if (!this._centrifuge.isTransportOpen()) {
       // @ts-ignore – we are hiding some symbols from public API autocompletion.
       this._centrifuge._debug('delay subscribe on', this.channel, 'till connected');
       // subscribe will be called later automatically.
@@ -280,16 +282,14 @@ export class Subscription extends (EventEmitter as new () => TypedEventEmitter<S
             return;
           }
           self._data = data;
-          self._sendSubscribe(self._token, false);
+          self._sendSubscribe(self._token);
         })
         return null;
       } else {
-        return self._sendSubscribe(self._token, skipSending);
+        return self._sendSubscribe(self._token);
       }
     }
-    if (optimistic) {
-      return null;
-    }
+
     this._getSubscriptionToken().then(function (token) {
       if (!self._isSubscribing()) {
         return;
@@ -305,10 +305,10 @@ export class Subscription extends (EventEmitter as new () => TypedEventEmitter<S
             return;
           }
           self._data = data;
-          self._sendSubscribe(token, false);
+          self._sendSubscribe(token);
         })
       } else {
-        self._sendSubscribe(token, false);
+        self._sendSubscribe(token);
       }
     }).catch(function (e) {
       if (!self._isSubscribing()) {
@@ -331,7 +331,12 @@ export class Subscription extends (EventEmitter as new () => TypedEventEmitter<S
     return null;
   }
 
-  private _sendSubscribe(token: string, skipSending: boolean): any {
+  private _sendSubscribe(token: string): any {
+    // we also need to check for transport state before sending subscription
+    // because it may change for subscription with side effects (getData, getToken options)
+    if (!this._centrifuge.isTransportOpen()) {
+      return null;
+    }
     const channel = this.channel;
 
     const req: any = {
@@ -375,7 +380,7 @@ export class Subscription extends (EventEmitter as new () => TypedEventEmitter<S
     this._inflight = true;
 
     // @ts-ignore – we are hiding some symbols from public API autocompletion.
-    this._centrifuge._call(cmd, skipSending).then(resolveCtx => {
+    this._centrifuge._call(cmd).then(resolveCtx => {
       this._inflight = false;
       // @ts-ignore - improve later.
       const result = resolveCtx.reply.subscribe;
@@ -492,7 +497,7 @@ export class Subscription extends (EventEmitter as new () => TypedEventEmitter<S
     const delay = this._getResubscribeDelay();
     this._resubscribeTimeout = setTimeout(function () {
       if (self._isSubscribing()) {
-        self._subscribe(false, false);
+        self._subscribe();
       }
     }, delay);
   }

From 9b325e00cb887e0823bddae70e68ef2f97d6cefc Mon Sep 17 00:00:00 2001
From: ThisIsEsh <13537179+ThisIsEsh@users.noreply.github.com>
Date: Fri, 5 Apr 2024 16:11:07 +0300
Subject: [PATCH 2/3] Hide isTransportOpen from public API, do not send
 subscribe frames during disconnect

---
 src/centrifuge.ts   |  6 +-----
 src/subscription.ts | 10 +++++++---
 2 files changed, 8 insertions(+), 8 deletions(-)

diff --git a/src/centrifuge.ts b/src/centrifuge.ts
index 1501e394..2230dbd8 100644
--- a/src/centrifuge.ts
+++ b/src/centrifuge.ts
@@ -411,10 +411,6 @@ export class Centrifuge extends (EventEmitter as new () => TypedEventEmitter<Cli
     })
   }
 
-  isTransportOpen() {
-    return this._transportIsOpen;
-  }
-
   private _debug(...args: any[]) {
     if (!this._debugEnabled) {
       return;
@@ -1371,7 +1367,7 @@ export class Centrifuge extends (EventEmitter as new () => TypedEventEmitter<Cli
   }
 
   protected _unsubscribe(sub: Subscription) {
-    if (!this.isTransportOpen()) {
+    if (!this._transportIsOpen) {
       return;
     }
     const req = {
diff --git a/src/subscription.ts b/src/subscription.ts
index 728f521a..9787b3ce 100644
--- a/src/subscription.ts
+++ b/src/subscription.ts
@@ -254,7 +254,9 @@ export class Subscription extends (EventEmitter as new () => TypedEventEmitter<S
     if (this._setState(SubscriptionState.Subscribing)) {
       this.emit('subscribing', { channel: this.channel, code: code, reason: reason });
     }
-    this._subscribe();
+    if (code === subscribingCodes.subscribeCalled) {
+      this._subscribe();
+    }
   }
 
   private _subscribe(): any {
@@ -263,7 +265,8 @@ export class Subscription extends (EventEmitter as new () => TypedEventEmitter<S
 
     // need to check transport readiness here, because there's no point for calling getData or getToken
     // if transport is not ready yet
-    if (!this._centrifuge.isTransportOpen()) {
+    // @ts-ignore – we are hiding some symbols from public API autocompletion.
+    if (!this._centrifuge._transportIsOpen) {
       // @ts-ignore – we are hiding some symbols from public API autocompletion.
       this._centrifuge._debug('delay subscribe on', this.channel, 'till connected');
       // subscribe will be called later automatically.
@@ -334,7 +337,8 @@ export class Subscription extends (EventEmitter as new () => TypedEventEmitter<S
   private _sendSubscribe(token: string): any {
     // we also need to check for transport state before sending subscription
     // because it may change for subscription with side effects (getData, getToken options)
-    if (!this._centrifuge.isTransportOpen()) {
+    // @ts-ignore – we are hiding some symbols from public API autocompletion.
+    if (!this._centrifuge._transportIsOpen) {
       return null;
     }
     const channel = this.channel;

From 1e35d237c570876d715a6a2f73e78ef06b5d8b89 Mon Sep 17 00:00:00 2001
From: ThisIsEsh <13537179+ThisIsEsh@users.noreply.github.com>
Date: Fri, 5 Apr 2024 16:42:13 +0300
Subject: [PATCH 3/3] Do not send subscribe frames during disconnect based only
 on _transportIsOpen flag

---
 src/centrifuge.ts   | 5 +++--
 src/subscription.ts | 4 +---
 2 files changed, 4 insertions(+), 5 deletions(-)

diff --git a/src/centrifuge.ts b/src/centrifuge.ts
index 2230dbd8..f25e927b 100644
--- a/src/centrifuge.ts
+++ b/src/centrifuge.ts
@@ -1207,7 +1207,9 @@ export class Centrifuge extends (EventEmitter as new () => TypedEventEmitter<Cli
     if (this._isDisconnected()) {
       return;
     }
-
+    // we mark transport is closed right away, because _clearConnectedState will move subscriptions to subscribing state
+    // if transport will still be open at this time, subscribe frames will be sent to closing transport
+    this._transportIsOpen = false;
     const previousState = this.state;
 
     const ctx = {
@@ -1248,7 +1250,6 @@ export class Centrifuge extends (EventEmitter as new () => TypedEventEmitter<Cli
       transport.close(); // Close only after setting this._transport to null to avoid recursion when calling transport close().
       // Need to mark as closed here, because connect call may be sync called after disconnect,
       // transport onClose callback will not be called yet
-      this._transportIsOpen = false;
       this._transportClosed = true;
       this._nextTransportId();
     } else {
diff --git a/src/subscription.ts b/src/subscription.ts
index 9787b3ce..8c48406b 100644
--- a/src/subscription.ts
+++ b/src/subscription.ts
@@ -254,9 +254,7 @@ export class Subscription extends (EventEmitter as new () => TypedEventEmitter<S
     if (this._setState(SubscriptionState.Subscribing)) {
       this.emit('subscribing', { channel: this.channel, code: code, reason: reason });
     }
-    if (code === subscribingCodes.subscribeCalled) {
-      this._subscribe();
-    }
+    this._subscribe();
   }
 
   private _subscribe(): any {