fix: Try to fix reconnection error by always force-closing requests, …

…and locking reconnection process
PLhery · Oct 22, 2021 · d73426e · d73426e
1 parent 568fea8
commit d73426e
Show file tree

Hide file tree

Showing 2 changed files with 34 additions and 19 deletions.
diff --git a/src/client-mixins/request-maker.mixin.ts b/src/client-mixins/request-maker.mixin.ts
@@ -38,6 +38,7 @@ export interface IGetHttpRequestArgs {
   forceBodyMode?: TBodyMode;
   enableAuth?: boolean;
   enableRateLimitSave?: boolean;
+  timeout?: number;
 }
 
 export interface IGetStreamRequestArgs {
@@ -77,7 +78,7 @@ export abstract class ClientRequestMaker {
   /** Send a new request and returns a wrapped `Promise<TwitterResponse<T>`. */
   send<T = any>(requestParams: IGetHttpRequestArgs) : Promise<TwitterResponse<T>> {
     const args = this.getHttpRequestArgs(requestParams);
-    const options = { method: args.method, headers: args.headers };
+    const options = { method: args.method, headers: args.headers, timeout: requestParams.timeout };
     const enableRateLimitSave = requestParams.enableRateLimitSave !== false;
 
     if (args.body) {

diff --git a/src/stream/TweetStream.ts b/src/stream/TweetStream.ts
@@ -24,7 +24,13 @@ export interface IConnectTweetStreamParams {
 /** Returns a number of milliseconds to wait for {tryOccurence} (starting from 1) */
 export type TStreamConnectRetryFn = (tryOccurence: number) => number;
 
-const basicReconnectRetry: TStreamConnectRetryFn = tryOccurence => Math.min((tryOccurence ** 2) * 1000, 25000);
+// In seconds
+const basicRetriesAttempt = [5, 15, 30, 60, 90, 120];
+// Default retry function
+const basicReconnectRetry: TStreamConnectRetryFn =
+  tryOccurence => tryOccurence > basicRetriesAttempt.length
+    ? 120000
+    : basicRetriesAttempt[tryOccurence - 1] * 1000;
 
 export class TweetStream<T = any> extends EventEmitter {
   public autoReconnect = false;
@@ -36,6 +42,7 @@ export class TweetStream<T = any> extends EventEmitter {
   protected retryTimeout?: NodeJS.Timeout;
   protected keepAliveTimeout?: NodeJS.Timeout;
   protected parser = new TweetStreamParser();
+  protected connectionProcessRunning = false;
 
   protected req?: ClientRequest;
   protected res?: IncomingMessage;
@@ -178,8 +185,11 @@ export class TweetStream<T = any> extends EventEmitter {
     }
     if (this.req) {
       this.req.removeAllListeners();
-      // Close connection silentely
-      this.req.destroy();
+
+      if (!this.req.destroyed) {
+        // Close connection silentely
+        this.req.destroy();
+      }
     }
   }
 
@@ -263,33 +273,37 @@ export class TweetStream<T = any> extends EventEmitter {
 
   /** Make a new request to (re)connect to Twitter. */
   async reconnect() {
-    let initialConnection = true;
+    if (this.connectionProcessRunning) {
+      throw new Error('Connection process is already running.');
+    }
 
-    if (this.req) {
-      initialConnection = false;
+    this.connectionProcessRunning = true;
 
-      if (!this.req.destroyed) {
+    try {
+      let initialConnection = true;
+
+      if (this.req) {
+        initialConnection = false;
         this.closeWithoutEmit();
       }
-    }
 
-    const { req, res } = await new RequestHandlerHelper(this.requestData).makeRequestAndResolveWhenReady();
+      const { req, res } = await new RequestHandlerHelper(this.requestData).makeRequestAndResolveWhenReady();
 
-    this.req = req;
-    this.res = res;
+      this.req = req;
+      this.res = res;
 
-    this.emit(initialConnection ? ETwitterStreamEvent.Connected : ETwitterStreamEvent.Reconnected);
-    this.parser.reset();
-    this.initEventsFromRequest();
+      this.emit(initialConnection ? ETwitterStreamEvent.Connected : ETwitterStreamEvent.Reconnected);
+      this.parser.reset();
+      this.initEventsFromRequest();
+    } finally {
+      this.connectionProcessRunning = false;
+    }
   }
 
   protected async onConnectionError(retryOccurence = 0) {
     this.unbindTimeouts();
-
     // Close the request if necessary
-    if (this.req && !this.req.destroyed) {
-      this.closeWithoutEmit();
-    }
+    this.closeWithoutEmit();
 
     // Terminate stream by events if necessary (no auto-reconnect or retries exceeded)
     if (!this.autoReconnect) {