fix: improve disconnection for queue events

taskforcesh · Sep 2, 2019 · 56b53a1 · 56b53a1
1 parent 2b0d8f3
commit 56b53a1
Show file tree

Hide file tree

Showing 4 changed files with 69 additions and 52 deletions.
diff --git a/src/classes/queue-base.ts b/src/classes/queue-base.ts
@@ -77,4 +77,8 @@ export class QueueBase extends EventEmitter {
     this.closing = this.connection.close();
     return this.closing;
   }
+
+  disconnect() {
+    return this.connection.disconnect();
+  }
 }
diff --git a/src/classes/queue-events.ts b/src/classes/queue-events.ts
@@ -1,7 +1,7 @@
 import { QueueEventsOptions } from '@src/interfaces';
-import { QueueBase } from './queue-base';
+import { delay } from 'bluebird';
 import { array2obj } from '../utils';
-import { Job } from './job';
+import { QueueBase } from './queue-base';
 
 export class QueueEvents extends QueueBase {
   constructor(name: string, opts?: QueueEventsOptions) {
@@ -27,38 +27,50 @@ export class QueueEvents extends QueueBase {
     let id = opts.lastEventId || '0-0';
 
     while (!this.closing) {
-      const data = await this.client.xread(
-        'BLOCK',
-        opts.blockingTimeout,
-        'STREAMS',
-        key,
-        id,
-      );
+      try {
+        const data = await this.client.xread(
+          'BLOCK',
+          opts.blockingTimeout,
+          'STREAMS',
+          key,
+          id,
+        );
 
-      if (data) {
-        const stream = data[0];
-        const events = stream[1];
+        if (data) {
+          const stream = data[0];
+          const events = stream[1];
 
-        for (let i = 0; i < events.length; i++) {
-          id = events[i][0];
-          const args = array2obj(events[i][1]);
+          for (let i = 0; i < events.length; i++) {
+            id = events[i][0];
+            const args = array2obj(events[i][1]);
 
-          //
-          // TODO: we may need to have a separate xtream for progress data
-          // to avoid this hack.
-          switch (args.event) {
-            case 'progress':
-              args.data = JSON.parse(args.data);
-              break;
-            case 'completed':
-              args.returnvalue = JSON.parse(args.returnvalue);
-              break;
-          }
+            //
+            // TODO: we may need to have a separate xtream for progress data
+            // to avoid this hack.
+            switch (args.event) {
+              case 'progress':
+                args.data = JSON.parse(args.data);
+                break;
+              case 'completed':
+                args.returnvalue = JSON.parse(args.returnvalue);
+                break;
+            }
 
-          this.emit(args.event, args, id);
-          this.emit(`${args.event}:${args.jobId}`, args, id);
+            this.emit(args.event, args, id);
+            this.emit(`${args.event}:${args.jobId}`, args, id);
+          }
+        }
+      } catch (err) {
+        if (err.message !== 'Connection is closed.') {
+          await delay(5000);
+          throw err;
         }
       }
     }
   }
+
+  async close() {
+    await super.close();
+    return this.disconnect();
+  }
 }
diff --git a/src/classes/redis-connection.ts b/src/classes/redis-connection.ts
@@ -72,6 +72,29 @@ export class RedisConnection {
     return this.client;
   }
 
+  async disconnect() {
+    const client = this.client;
+    if (client.status !== 'end') {
+      let _resolve, _reject;
+
+      const disconnecting = new Promise((resolve, reject) => {
+        client.once('end', resolve);
+        client.once('error', reject);
+        _resolve = resolve;
+        _reject = reject;
+      });
+
+      client.disconnect();
+
+      try {
+        await disconnecting;
+      } finally {
+        client.removeListener('end', _resolve);
+        client.removeListener('error', _reject);
+      }
+    }
+  }
+
   async close() {}
 
   private async getRedisVersion() {

diff --git a/src/classes/worker.ts b/src/classes/worker.ts
@@ -263,7 +263,7 @@ export class Worker extends QueueBase {
     //
     // Force reconnection of blocking connection to abort blocking redis call immediately.
     //
-    this.waiting && (await redisClientDisconnect(this.client));
+    this.waiting && (await this.client.disconnect());
 
     // If we are disconnected, how are we going to update the completed/failed sets?
     if (this.processing) {
@@ -282,33 +282,11 @@ export class Worker extends QueueBase {
       if (!force) {
         await this.whenCurrentJobsFinished(false);
       } else {
-        await redisClientDisconnect(this.client);
+        await this.disconnect();
       }
     } finally {
       this.childPool && this.childPool.clean();
     }
     this.emit('closed');
   }
 }
-
-async function redisClientDisconnect(client: IORedis.Redis) {
-  if (client.status !== 'end') {
-    let _resolve, _reject;
-
-    const disconnecting = new Promise((resolve, reject) => {
-      client.once('end', resolve);
-      client.once('error', reject);
-      _resolve = resolve;
-      _reject = reject;
-    });
-
-    client.disconnect();
-
-    try {
-      await disconnecting;
-    } finally {
-      client.removeListener('end', _resolve);
-      client.removeListener('error', _reject);
-    }
-  }
-}