feat: rework UDP cluster manager message handling - create a thread to process messages without blocking event loop

Author: SerhiyGreench

package.json

@@ -13,8 +13,8 @@
   "scripts": {
     "benchmark": "node benchmark -c $(( $(nproc) - 2 )) -m 100000",
     "prepare": "./node_modules/.bin/tsc",
-    "test": "./node_modules/.bin/tsc && ./node_modules/.bin/nyc mocha && ./node_modules/.bin/nyc report --reporter=text-lcov",
-    "test-fast": "./node_modules/.bin/tsc && ./node_modules/.bin/nyc mocha && /usr/bin/env node -e \"import('open').then(open => open.default('file://`pwd`/coverage/index.html', { wait: false }))\"",
+    "test": "./node_modules/.bin/tsc && ./node_modules/.bin/nyc mocha --exit --timeout 10000 && ./node_modules/.bin/nyc report --reporter=text-lcov",
+    "test-fast": "./node_modules/.bin/tsc && ./node_modules/.bin/nyc mocha --exit --timeout 10000 && /usr/bin/env node -e \"import('open').then(open => open.default('file://`pwd`/coverage/index.html', { wait: false }))\"",
     "test-local": "export COVERALLS_REPO_TOKEN=$IMQ_COVERALLS_TOKEN && npm test && /usr/bin/env node -e \"import('open').then(open => open.default('https://coveralls.io/github/imqueue/imq', { wait: false }))\"",
     "test-dev": "npm run test && npm run clean-js && npm run clean-typedefs && npm run clean-maps",
     "test-coverage": "cat ./coverage/lcov.info | CODECLIMATE_API_HOST=https://codebeat.co/webhooks/code_coverage CODECLIMATE_REPO_TOKEN=85bb2a18-4ebb-4e48-a2ce-92b7bf438b1a ./node_modules/.bin/codeclimate-test-reporter",
@@ -80,7 +80,9 @@
     ],
     "recursive": true,
     "bail": true,
-    "full-trace": true
+    "full-trace": true,
+    "exit": true,
+    "timeout": 10000
   },
   "nyc": {
     "check-coverage": false,

src/ClusteredRedisQueue.ts

@@ -502,15 +502,14 @@ export class ClusteredRedisQueue implements IMessageQueue,
             return;
         }
 
-        if (remove.imq) {
-            this.imqs = this.imqs.filter(imq => remove.imq !== imq);
-            remove.imq.destroy().catch();
-        }
+        const imqToRemove = remove.imq;
 
-        this.clusterEmitter.emit('remove', {
-            server: remove,
-            imq: remove.imq,
-        });
+        if (imqToRemove) {
+            this.imqs = this.imqs.filter(
+                imq => imqToRemove.redisKey !== imq.redisKey
+            );
+            imqToRemove.destroy().catch();
+        }
 
         this.queueLength = this.imqs.length;
         this.servers = this.servers.filter(
@@ -519,12 +518,22 @@ export class ClusteredRedisQueue implements IMessageQueue,
                 server,
             ),
         );
+        this.clusterEmitter.emit('remove', {
+            server: remove,
+            imq: imqToRemove,
+        });
     }
 
     private addServerWithQueueInitializing(
         server: ClusterServer,
         initializeQueue: boolean = true,
     ): ClusterServer {
+        const existingServer = this.findServer(server);
+
+        if (existingServer) {
+            return existingServer;
+        }
+
         const newServer: ClusterServer = {
             id: server.id,
             host: server.host,

src/RedisQueue.ts

@@ -201,7 +201,7 @@ export class RedisQueue extends EventEmitter<EventMap>
     /**
      * This queue instance unique key (identifier), for internal use
      */
-    private readonly redisKey: string;
+    public readonly redisKey: string;
 
     /**
      * LUA scripts for redis
@@ -697,6 +697,7 @@ export class RedisQueue extends EventEmitter<EventMap>
                 ),
                 retryStrategy: this.retryStrategy(context),
                 autoResubscribe: true,
+                enableReadyCheck: true,
             });
 
             context[channel] = makeRedisSafe(redis);