feat: implemented destroy method on UDPClusterManager

Author: SerhiyGreench

src/ClusterManager.ts

@@ -22,6 +22,7 @@
  * <[email protected]> to get commercial licensing options.
  */
 import { IMessageQueueConnection, IServerInput } from './IMessageQueue';
+import { uuid } from './uuid';
 
 export interface ICluster {
     add: (server: IServerInput) => void;
@@ -31,12 +32,39 @@ export interface ICluster {
     ) => T | undefined;
 }
 
+export interface InitializedCluster extends ICluster {
+    id: string;
+}
+
 export abstract class ClusterManager {
-    protected clusters: ICluster[] = [];
+    protected clusters: InitializedCluster[] = [];
 
     protected constructor() {}
 
-    public init(cluster: ICluster): void {
-        this.clusters.push(cluster);
+    public init(cluster: ICluster): InitializedCluster {
+        const initializedCluster = Object.assign(cluster, { id: uuid() });
+
+        this.clusters.push(initializedCluster);
+
+        return initializedCluster;
     }
+
+    public async remove(
+        cluster: string | InitializedCluster,
+        destroy: boolean = true,
+    ): Promise<void> {
+        const id = typeof cluster === 'string' ? cluster : cluster.id;
+
+        this.clusters = this.clusters.filter(cluster => cluster.id !== id);
+
+        if (
+            this.clusters.length === 0
+            && destroy
+            && typeof this.destroy === 'function'
+        ) {
+            await this.destroy();
+        }
+    }
+
+    public abstract destroy(): Promise<void>;
 }

src/ClusteredRedisQueue.ts

@@ -36,6 +36,7 @@ import {
     IServerInput,
     copyEventEmitter,
 } from '.';
+import { InitializedCluster } from './ClusterManager';
 
 interface ClusterServer extends IMessageQueueConnection {
     imq?: RedisQueue;
@@ -130,6 +131,8 @@ export class ClusteredRedisQueue implements IMessageQueue,
         subscription: null,
     };
 
+    private initializedClusters: InitializedCluster[] = [];
+
     /**
      * Class constructor
      *
@@ -167,11 +170,11 @@ export class ClusteredRedisQueue implements IMessageQueue,
 
         if (this.options.clusterManagers?.length) {
             for (const manager of this.options.clusterManagers) {
-                manager.init({
+                this.initializedClusters.push(manager.init({
                     add: this.addServer.bind(this),
                     remove: this.removeServer.bind(this),
                     find: this.findServer.bind(this),
-                });
+                }));
             }
         }
     }
@@ -247,7 +250,7 @@ export class ClusteredRedisQueue implements IMessageQueue,
     }
 
     /**
-     * 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.
      *
@@ -258,6 +261,16 @@ export class ClusteredRedisQueue implements IMessageQueue,
 
         await this.batch('destroy',
             'Destroying clustered redis message queue...');
+
+        if (!this.options.clusterManagers?.length) {
+            return;
+        }
+
+        for await (const manager of this.options.clusterManagers) {
+            for await (const cluster of this.initializedClusters) {
+                await manager.remove(cluster);
+            }
+        }
     }
 
     // noinspection JSUnusedGlobalSymbols

src/RedisQueue.ts

@@ -931,7 +931,7 @@ export class RedisQueue extends EventEmitter<EventMap>
                 await this.processKeys(keys, now);
 
                 if (cursor === '0') {
-                    return ;
+                    return;
                 }
             } catch (err) {
                 this.emitError('OnSafeDelivery',
@@ -954,7 +954,7 @@ export class RedisQueue extends EventEmitter<EventMap>
      */
     private async processKeys(keys: string[], now: number): Promise<void> {
         if (!keys.length) {
-            return ;
+            return;
         }
 
         for (const key of keys) {
@@ -1393,5 +1393,4 @@ export class RedisQueue extends EventEmitter<EventMap>
             }
         });
     }
-
 }