feat: implmented queueLength method on Redis message queue

Author: SerhiyGreench

src/ClusteredRedisQueue.ts

@@ -108,7 +108,7 @@ export class ClusteredRedisQueue implements IMessageQueue,
      *
      * @type {number}
      */
-    private queueLength: number = 0;
+    private imqLength: number = 0;
 
     /**
      * Template EventEmitter instance used to replicate queue EventEmitters when
@@ -223,7 +223,7 @@ export class ClusteredRedisQueue implements IMessageQueue,
         delay?: number,
         errorHandler?: (err: Error) => void,
     ): Promise<string> {
-        if (!this.queueLength) {
+        if (!this.imqLength) {
             return await new Promise(resolve => this.clusterEmitter.once(
                 'initialized',
                 async ({ imq }) => {
@@ -237,7 +237,7 @@ export class ClusteredRedisQueue implements IMessageQueue,
             ));
         }
 
-        if (this.currentQueue >= this.queueLength) {
+        if (this.currentQueue >= this.imqLength) {
             this.currentQueue = 0;
         }
 
@@ -285,6 +285,18 @@ export class ClusteredRedisQueue implements IMessageQueue,
             'Clearing clustered redis message queue...');
     }
 
+    public async queueLength(): Promise<number> {
+        const promises = [];
+
+        for (const imq of this.imqs) {
+            promises.push(imq.queueLength());
+        }
+
+        const lengths = await Promise.all(promises);
+
+        return lengths.reduce((total, length) => total + length, 0);
+    }
+
     /**
      * Batch imq action processing on all registered imqs at once
      *
@@ -511,7 +523,7 @@ export class ClusteredRedisQueue implements IMessageQueue,
             imqToRemove.destroy().catch();
         }
 
-        this.queueLength = this.imqs.length;
+        this.imqLength = this.imqs.length;
         this.servers = this.servers.filter(
             existing => !ClusteredRedisQueue.matchServers(
                 existing,
@@ -557,7 +569,7 @@ export class ClusteredRedisQueue implements IMessageQueue,
         this.imqs.push(imq);
         this.servers.push(newServer);
         this.clusterEmitter.emit('add', { server: newServer, imq });
-        this.queueLength = this.imqs.length;
+        this.imqLength = this.imqs.length;
 
         return newServer;
     }
@@ -605,5 +617,4 @@ export class ClusteredRedisQueue implements IMessageQueue,
 
         return sameId || sameAddress;
     }
-
 }

src/IMessageQueue.ts

@@ -421,4 +421,12 @@ export interface IMessageQueue extends EventEmitter<EventMap> {
      * @returns {Promise<IMessageQueue>}
      */
     clear(): Promise<IMessageQueue>;
+
+    /**
+     * Retrieves the current count of messages in the queue.
+     * Supposed to be an async function.
+     *
+     * @returns {Promise<number>}
+     */
+    queueLength(): Promise<number>;
 }

src/RedisQueue.ts

@@ -512,7 +512,7 @@ export class RedisQueue extends EventEmitter<EventMap>
     }
 
     /**
-     * Clears queue data in redis;
+     * Clears queue data in redis
      *
      * @returns {Promise<void>}
      */
@@ -541,6 +541,20 @@ export class RedisQueue extends EventEmitter<EventMap>
         return this;
     }
 
+    /**
+     * Retrieves the current count of messages in the queue
+     *
+     * @returns {Promise<number>}
+     */
+    @profile()
+    public async queueLength(): Promise<number> {
+        if (!this.writer) {
+            return 0;
+        }
+
+        return this.writer.llen(this.key);
+    }
+
     /**
      * Returns true if publisher mode is enabled on this queue, false otherwise.
      *
@@ -698,6 +712,8 @@ export class RedisQueue extends EventEmitter<EventMap>
                 retryStrategy: this.retryStrategy(context),
                 autoResubscribe: true,
                 enableReadyCheck: true,
+                enableOfflineQueue: true,
+                autoResendUnfulfilledCommands: true,
             });
 
             context[channel] = makeRedisSafe(redis);