feat: added possibility to turn off server alive timeout check in udp cluster manager & added extra logging

Author: SerhiyGreench

src/ClusteredRedisQueue.ts

@@ -169,6 +169,8 @@ export class ClusteredRedisQueue implements IMessageQueue,
         }
 
         if (this.options.clusterManagers?.length) {
+            this.verbose('Initializing cluster managers...');
+
             for (const manager of this.options.clusterManagers) {
                 this.initializedClusters.push(manager.init({
                     add: this.addServer.bind(this),
@@ -297,6 +299,14 @@ export class ClusteredRedisQueue implements IMessageQueue,
         return lengths.reduce((total, length) => total + length, 0);
     }
 
+    private verbose(message: string): void {
+        if (this.options.verbose) {
+            this.logger.info(`[IMQ-CORE][ClusteredRedisQueue][${ 
+                this.name
+            }]: ${ message }`);
+        }
+    }
+
     /**
      * Batch imq action processing on all registered imqs at once
      *
@@ -306,7 +316,7 @@ export class ClusteredRedisQueue implements IMessageQueue,
      * @return {Promise<this>}
      */
     private async batch(action: string, message: string): Promise<this> {
-        this.logger.log(message);
+        this.logger.info(message);
 
         const promises = [];
 
@@ -498,6 +508,8 @@ export class ClusteredRedisQueue implements IMessageQueue,
      * @returns {void}
      */
     protected addServer(server: IServerInput): ClusterServer {
+        this.verbose(`Adding new server: ${ JSON.stringify(server) }`);
+
         return this.addServerWithQueueInitializing(server, true);
     }
 
@@ -508,6 +520,8 @@ export class ClusteredRedisQueue implements IMessageQueue,
      * @returns {void}
      */
     protected removeServer(server: IServerInput): void {
+        this.verbose(`Removing the server: ${ JSON.stringify(server) }`);
+
         const remove = this.findServer(server);
 
         if (!remove) {
@@ -580,6 +594,9 @@ export class ClusteredRedisQueue implements IMessageQueue,
 
     private async initializeQueue(imq: RedisQueue): Promise<void> {
         copyEventEmitter(this.templateEmitter, imq);
+        this.verbose(`Initializing queue with state: ${ 
+            JSON.stringify(this.state) 
+        }`);
 
         if (this.state.started) {
            await imq.start();

src/RedisQueue.ts

@@ -899,14 +899,11 @@ export class RedisQueue extends EventEmitter<EventMap>
             return;
         }
 
-        this.reconnecting[channel] = true;
-
         const attempts = (this.reconnectAttempts[channel] || 0) + 1;
-        this.reconnectAttempts[channel] = attempts;
+        const delay = Math.min(30000, 1000 * Math.pow(2, attempts - 1));
 
-        const base = Math.min(30000, 1000 * Math.pow(2, attempts - 1));
-        const jitter = Math.floor(base * 0.2 * Math.random());
-        const delay = base + jitter;
+        this.reconnecting[channel] = true;
+        this.reconnectAttempts[channel] = attempts;
 
         this.verbose(`Scheduling ${ channel } reconnect in ${
             delay } ms (attempt ${ attempts })`);

src/UDPClusterManager.ts

@@ -59,12 +59,28 @@ export interface UDPClusterManagerOptions {
      * @type {number}
      */
     aliveTimeoutCorrection: number;
+
+    /**
+     * Message queue alive-server check flag. If set to false, the server will
+     * not be checked for liveness on each broadcast message with a timeout.
+     * Can be specified by the environment variable if the given option is not
+     * bypassed: IMQ_UDP_CLUSTER_MANAGER_ALIVE_CHECK
+     *
+     * @default true
+     * @type {boolean}
+     */
+    useAliveCheck: boolean;
 }
 
+const IMQ_UDP_CLUSTER_MANAGER_ALIVE_CHECK = !!+(
+    process.env.IMQ_UDP_CLUSTER_MANAGER_ALIVE_CHECK || 1
+);
+
 export const DEFAULT_UDP_CLUSTER_MANAGER_OPTIONS: UDPClusterManagerOptions = {
     port: 63000,
     address: '255.255.255.255',
     aliveTimeoutCorrection: 5000,
+    useAliveCheck: IMQ_UDP_CLUSTER_MANAGER_ALIVE_CHECK,
 };
 
 export class UDPClusterManager extends ClusterManager {

src/UDPWorker.ts

@@ -49,7 +49,7 @@ interface Message {
     timeout: number;
 }
 
-class UDPClusterWorker {
+class UDPWorker {
     private readonly socket: Socket;
     private readonly servers = new Map<string, string>();
 
@@ -91,16 +91,19 @@ class UDPClusterWorker {
     private addServer(message: Message): void {
         this.messagePort.postMessage({
             type: 'cluster:add',
-            server: UDPClusterWorker.mapMessage(message),
+            server: UDPWorker.mapMessage(message),
         });
-        this.serverAliveWait(message);
+
+        if (this.options.useAliveCheck) {
+            this.serverAliveWait(message);
+        }
     }
 
     private removeServer(message: Message): void {
-        this.servers.delete(UDPClusterWorker.getServerKey(message));
+        this.servers.delete(UDPWorker.getServerKey(message));
         this.messagePort.postMessage({
             type: 'cluster:remove',
-            server: UDPClusterWorker.mapMessage(message),
+            server: UDPWorker.mapMessage(message),
         });
     }
 
@@ -119,7 +122,7 @@ class UDPClusterWorker {
         const stamp = uuid();
         const correction = this.options.aliveTimeoutCorrection ?? 0;
         const effectiveTimeout = message.timeout + correction + 1;
-        const key = UDPClusterWorker.getServerKey(message);
+        const key = UDPWorker.getServerKey(message);
 
         this.servers.set(key, stamp);
 
@@ -223,5 +226,5 @@ class UDPClusterWorker {
 }
 
 if (!isMainThread && parentPort) {
-    new UDPClusterWorker(workerData, parentPort);
+    new UDPWorker(workerData, parentPort);
 }