Merge pull request #124 from harness/FFM-12477-streaming-fixes

FFM-12477 Streaming fixes

Author: EmilianoSanchez

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "@harnessio/ff-nodejs-server-sdk",
-  "version": "1.8.5",
+  "version": "1.8.6",
   "description": "Feature flags SDK for NodeJS environments",
   "main": "dist/cjs/index.js",
   "module": "dist/esm/index.mjs",

package.json

@@ -45,6 +45,7 @@ export class StreamProcessor {
   private readyState: number;
   private log: ConsoleLog;
   private retryAttempt = 0;
+  private retryTimeout: NodeJS.Timeout | undefined;
 
   constructor(
     api: ClientApi,
@@ -111,16 +112,19 @@ export class StreamProcessor {
     };
 
     const onFailed = (msg: string) => {
-      if (this.readyState !== StreamProcessor.CLOSED) {
+      if (this.readyState !== StreamProcessor.CLOSED && !this.retryTimeout) {
         this.retryAttempt += 1;
 
         const delayMs = this.getRandomRetryDelayMs();
         warnStreamDisconnectedWithRetry(msg, delayMs, this.log);
         this.readyState = StreamProcessor.RETRYING;
         this.eventBus.emit(StreamEvent.RETRYING);
 
-        setTimeout(() => {
-          this.connect(url, options, onConnected, onFailed);
+        this.retryTimeout = setTimeout(() => {
+          this.retryTimeout = undefined;
+          if (this.readyState !== StreamProcessor.CLOSED) {
+            this.connect(url, options, onConnected, onFailed);
+          }
         }, delayMs);
       }
     };
@@ -135,6 +139,14 @@ export class StreamProcessor {
     return Math.min(delayMs, 60000);
   }
 
+  private cleanupConnection(): void {
+    if (this.request) {
+      this.request.removeAllListeners();
+      this.request.destroy();
+      this.request = undefined;
+    }
+  }
+
   private connect(
     url: string,
     options: RequestOptions,
@@ -146,6 +158,8 @@ export class StreamProcessor {
       return;
     }
 
+    this.cleanupConnection();
+
     const isSecure = url.startsWith('https:');
     this.log.debug('SSE HTTP start request', url);
 
@@ -174,8 +188,8 @@ export class StreamProcessor {
       .on('timeout', () => {
         onFailed(
           'SSE request timed out after ' +
-            StreamProcessor.SSE_TIMEOUT_MS +
-            'ms',
+          StreamProcessor.SSE_TIMEOUT_MS +
+          'ms',
         );
       })
       .setTimeout(StreamProcessor.SSE_TIMEOUT_MS);
@@ -256,6 +270,8 @@ export class StreamProcessor {
       return;
     }
 
+    clearTimeout(this.retryTimeout);
+
     this.readyState = StreamProcessor.CLOSED;
     this.log.info('Closing StreamProcessor');
 

src/streaming.ts

@@ -1 +1 @@
-export const VERSION = '1.8.5';
+export const VERSION = '1.8.6';