feat: refactored RedisQueue & fixed inspected issues

README.md

@@ -39,7 +39,8 @@ With current implementation on RedisQueue:
 # Requirements
 
 Currently this module have only one available adapter which is Redis server
-related. So redis-server > 3.8+ is required.
+related. Redis server 6.2+ is required (the queue relies on `LMOVE`/`BLMOVE`
+commands for safe message delivery).
 
 If config command is disabled on redis it will be required to turn on manually
 keyspace notification events (actual on use with ElasticCache on AWS), like:

benchmark/index.ts

@@ -27,7 +27,8 @@ import * as fs from 'fs';
 import { parseArgs } from 'node:util';
 import { run } from './redis-test';
 import { resolve } from 'path';
-import { uuid, AnyJson } from '..';
+import { randomUUID as uuid } from 'crypto';
+import { AnyJson } from '..';
 import { setAffinity } from './affinity';
 
 const cluster: any = require('cluster');

benchmark/redis-test.ts

@@ -21,14 +21,8 @@
  * purchase a proprietary commercial license. Please contact us at
  * <[email protected]> to get commercial licensing options.
  */
-import IMQ, {
-    IMQOptions,
-    IJson,
-    uuid,
-    pack,
-    JsonObject,
-    AnyJson,
-} from '../index';
+import { randomUUID as uuid } from 'crypto';
+import IMQ, { IMQOptions, IJson, pack, JsonObject, AnyJson } from '../index';
 
 /**
  * Sample message used within tests

src/ClusterManager.ts

@@ -22,7 +22,7 @@
  * <[email protected]> to get commercial licensing options.
  */
 import { IMessageQueueConnection, IServerInput } from './IMessageQueue';
-import { uuid } from './uuid';
+import { randomUUID } from 'crypto';
 
 export interface ICluster {
     add: (server: IServerInput) => IMessageQueueConnection;
@@ -41,7 +41,7 @@ export abstract class ClusterManager {
 
     public init(cluster: ICluster): InitializedCluster {
         const initializedCluster = Object.assign(cluster, {
-            id: uuid(),
+            id: randomUUID(),
         }) as InitializedCluster;
 
         this.clusters.push(initializedCluster);

src/ClusteredRedisQueue.ts

@@ -92,7 +92,6 @@ export class ClusteredRedisQueue
      *
      * @type {IMessageQueueConnection[]}
      */
-    // tslint:disable-next-line:completed-docs
     private servers: ClusterServer[] = [];
 
     /**
@@ -102,7 +101,6 @@ export class ClusteredRedisQueue
      */
     private currentQueue: number = 0;
 
-    // noinspection TypeScriptFieldCanBeMadeReadonly
     /**
      * Total length of RedisQueue instances
      *
@@ -150,7 +148,6 @@ export class ClusteredRedisQueue
         this.clusterEmitter = new EventEmitter();
         this.options = buildOptions<IMQOptions>(DEFAULT_IMQ_OPTIONS, options);
 
-        // istanbul ignore next
         this.logger = this.options.logger || console;
 
         if (!this.options.cluster && !this.options.clusterManagers?.length) {
@@ -280,7 +277,6 @@ export class ClusteredRedisQueue
         }
     }
 
-    // noinspection JSUnusedGlobalSymbols
     /**
      * Clears queue data in queue host application.
      * Supposed to be an async function.
@@ -322,13 +318,18 @@ export class ClusteredRedisQueue
      * @param {string} message
      * @return {Promise<this>}
      */
-    private async batch(action: string, message: string): Promise<this> {
+    private async batch(
+        action: 'start' | 'stop' | 'destroy' | 'clear',
+        message: string,
+    ): Promise<this> {
         this.logger.info(message);
 
-        const promises = [];
+        const promises: Promise<unknown>[] = [];
 
         for (const imq of this.imqs) {
-            promises.push(imq[action]());
+            const run = imq[action] as () => Promise<unknown>;
+
+            promises.push(run.call(imq));
         }
 
         await Promise.all(promises);
@@ -337,7 +338,6 @@ export class ClusteredRedisQueue
     }
 
     // EventEmitter interface
-    // istanbul ignore next
     /**
      * Applies the named EventEmitter method to every underlying emitter,
      * forwarding the call across the whole cluster. Dispatch is reflective
@@ -363,110 +363,93 @@ export class ClusteredRedisQueue
         return results;
     }
 
-    // istanbul ignore next
     public on(...args: any[]): this {
         this.applyToEmitters('on', args);
 
         return this;
     }
 
-    // istanbul ignore next
-    // noinspection JSUnusedGlobalSymbols
     public off(...args: any[]): this {
         this.applyToEmitters('off', args);
 
         return this;
     }
 
-    // istanbul ignore next
     public once(...args: any[]): this {
         this.applyToEmitters('once', args);
 
         return this;
     }
 
-    // istanbul ignore next
     public addListener(...args: any[]): this {
         this.applyToEmitters('addListener', args);
 
         return this;
     }
 
-    // istanbul ignore next
     public removeListener(...args: any[]): this {
         this.applyToEmitters('removeListener', args);
 
         return this;
     }
 
-    // istanbul ignore next
     public removeAllListeners(...args: any[]): this {
         this.applyToEmitters('removeAllListeners', args);
 
         return this;
     }
 
-    // istanbul ignore next
     public prependListener(...args: any[]): this {
         this.applyToEmitters('prependListener', args);
 
         return this;
     }
 
-    // istanbul ignore next
     public prependOnceListener(...args: any[]): this {
         this.applyToEmitters('prependOnceListener', args);
 
         return this;
     }
 
-    // istanbul ignore next
     public setMaxListeners(...args: any[]): this {
         this.applyToEmitters('setMaxListeners', args);
 
         return this;
     }
 
-    // istanbul ignore next
     // Aggregates listeners across every underlying emitter, so the result is
     // an untyped union rather than Node's per-emitter conditional listener type.
     public listeners(...args: any[]): any[] {
         return this.applyToEmitters('listeners', args).flat();
     }
 
-    // istanbul ignore next
     public rawListeners(...args: any[]): any[] {
         return this.applyToEmitters('rawListeners', args).flat();
     }
 
-    // istanbul ignore next
     public getMaxListeners(): number {
         return this.templateEmitter.getMaxListeners();
     }
 
-    // istanbul ignore next
     public emit(...args: any[]): boolean {
         this.applyToEmitters('emit', args);
 
         return true;
     }
 
-    // istanbul ignore next
     public eventNames(): (keyof EventMap)[] {
         const source = this.imqs[0] || this.templateEmitter;
 
         return source.eventNames() as (keyof EventMap)[];
     }
 
-    // istanbul ignore next
     public listenerCount(...args: any[]): number {
         const source = this.imqs[0] || this.templateEmitter;
         const fn = source.listenerCount as (...a: any[]) => number;
 
         return fn.apply(source, args);
     }
 
-    // istanbul ignore next
     public async publish(data: JsonObject, toName?: string): Promise<void> {
         const promises: Array<Promise<void>> = [];
 
@@ -477,7 +460,6 @@ export class ClusteredRedisQueue
         await Promise.all(promises);
     }
 
-    // istanbul ignore next
     public async subscribe(
         channel: string,
         handler: (data: JsonObject) => any,
@@ -493,7 +475,6 @@ export class ClusteredRedisQueue
         await Promise.all(promises);
     }
 
-    // istanbul ignore next
     public async unsubscribe(): Promise<void> {
         this.state.subscription = null;
 

src/IMessageQueue.ts

@@ -282,6 +282,28 @@ export interface IMQOptions extends Partial<IMessageQueueConnection> {
      */
     clusterManagers?: ClusterManager[];
 
+    /**
+     * Enables/disables process signal handling (SIGTERM, SIGINT, SIGABRT)
+     * by the queue. When enabled, the queue frees its watcher lock and
+     * exits the process on those signals. Disable when the host
+     * application manages its own shutdown sequence.
+     *
+     * @default true
+     * @type {boolean}
+     */
+    handleSignals?: boolean;
+
+    /**
+     * When enabled, send() resolves only after the message write is
+     * confirmed by redis (and rejects on write failures). By default
+     * writes are fire-and-forget for maximum throughput and failures are
+     * reported through the optional errorHandler argument only.
+     *
+     * @default false
+     * @type {boolean}
+     */
+    awaitWrites?: boolean;
+
     /**
      * Enables/disables verbose logging
      *
@@ -317,7 +339,8 @@ export type IMessageQueueConstructor = new (
  *
  * @example
  * ~~~typescript
- * import { IMessageQueue, EventEmitter, uuid } from '@imqueue/core';
+ * import { IMessageQueue, EventEmitter } from '@imqueue/core';
+ * import { randomUUID } from 'crypto';
  *
  * class SomeMQAdapter implements IMessageQueue extends EventEmitter {
  *      public async start(): Promise<SomeMQAdapter> {
@@ -333,7 +356,7 @@ export type IMessageQueueConstructor = new (
  *          message: JsonObject,
  *          delay?: number
  *      ): Promise<string> {
- *          const messageId = uuid();
+ *          const messageId = randomUUID();
  *          // ... implementation goes here
  *          return messageId;
  *      }
@@ -374,21 +397,21 @@ export interface IMessageQueue extends EventEmitter<EventMap> {
     start(): Promise<IMessageQueue>;
 
     /**
-     * Stops the queue (should stop handle queue messages).
+     * Stops the queue (should stop to handle queue messages).
      * Supposed to be an async function.
      *
      * @returns {Promise<IMessageQueue>}
      */
     stop(): Promise<IMessageQueue>;
 
     /**
-     * Sends a message to given queue name with the given data.
+     * Sends a message to the given queue name with the given data.
      * Supposed to be an async function.
      *
-     * @param {string} toQueue - queue name to which message should be sent to
+     * @param {string} toQueue - queue name to which a message should be sent to
      * @param {JsonObject} message - message data
-     * @param {number} [delay] - if specified, message will be handled in the
-     *        target queue after specified period of time in milliseconds.
+     * @param {number} [delay] - if specified, a message will be handled in the
+     *        target queue after a specified period of time in milliseconds.
      * @param {(err: Error) => void} [errorHandler] - callback called only when
      *        internal error occurs during message send execution.
      * @returns {Promise<string>} - message identifier
@@ -401,7 +424,7 @@ export interface IMessageQueue extends EventEmitter<EventMap> {
     ): Promise<string>;
 
     /**
-     * Creates or uses subscription channel with the given name and sets
+     * Creates or uses a subscription channel with the given name and sets
      * message handler on data receive
      *
      * @param {string} channel - channel name
@@ -420,11 +443,11 @@ export interface IMessageQueue extends EventEmitter<EventMap> {
     unsubscribe(): Promise<void>;
 
     /**
-     * Publishes data to current queue channel
+     * Publishes data to the current queue channel
      *
-     * If toName specified will publish to pubsub with different name. This
+     * If toName specified will publish to pubsub with a different name. This
      * can be used to implement broadcasting some messages to other subscribers
-     * on other pubsub channels. Different name should be in the same namespace
+     * on other pubsub channels. Different names should be in the same namespace
      * (same imq prefix)
      *
      * @param {JsonObject} data - data to publish as channel message
@@ -434,7 +457,7 @@ export interface IMessageQueue extends EventEmitter<EventMap> {
     publish(data: JsonObject, toName?: string): Promise<void>;
 
     /**
-     * Safely destroys current queue, unregistered all set event
+     * Safely destroys the current queue, unregistered all set event
      * listeners and connections.
      * Supposed to be an async function.
      *