refactor - extract utility method to wait for in-flight commands to complete

Author: nkaradzhov

packages/client/lib/client/commands-queue.ts

@@ -77,7 +77,6 @@ export default class RedisCommandsQueue {
     this.#maxLength = maxLength;
     this.#onShardedChannelMoved = onShardedChannelMoved;
     this.decoder = this.#initiateDecoder();
-    this.#waitingForReply.events.on('empty', this.events.emit.bind(this.events, 'waitingForReplyEmpty'))
   }
 
   #onReply(reply: ReplyUnion) {
@@ -154,8 +153,15 @@ export default class RedisCommandsQueue {
     this.#invalidateCallback = callback;
   }
 
-  isWaitingForReply(): boolean {
-    return this.#waitingForReply.length > 0;
+  async waitForInflightCommandsToComplete(): Promise<void> {
+    // In-flight commands already completed
+    if(this.#waitingForReply.length === 0) {
+      return
+    };
+    // Otherwise wait for in-flight commands to fire `empty` event
+    return new Promise(resolve => {
+      this.#waitingForReply.events.on('empty', resolve)
+    });
   }
 
   addCommand<T>(

packages/client/lib/client/enterprise-maintenance-manager.ts

@@ -19,14 +19,14 @@ export default class EnterpriseMaintenanceManager extends EventEmitter {
 
   //  Queue:
   //     toWrite [ C D E ]
-  //     waitingForReply [ A B ]
+  //     waitingForReply [ A B ]   - aka In-flight commands
   //
   //  time: ---1-2---3-4-5-6---------------------------
   //
   //  1. [EVENT] MOVING PN received
   //  2. [ACTION] Pause writing ( we need to wait for new socket to connect and for all in-flight commands to complete )
   //  3. [EVENT] New socket connected
-  //  4. [EVENT] WaitingForReply commands completed
+  //  4. [EVENT] In-flight commands completed
   //  5. [ACTION] Destroy old socket
   //  6. [ACTION] Resume writing -> we are going to write to the new socket from now on
   #onMoving = async (
@@ -46,17 +46,8 @@ export default class EnterpriseMaintenanceManager extends EventEmitter {
     await newSocket.connect();
     // 3 [EVENT] New socket connected
 
-    // Wait until waitingForReply is empty
-    await new Promise<void>((resolve) => {
-      if (!this.commandsQueue.isWaitingForReply()) {
-        resolve();
-      } else {
-        this.commandsQueue.events.once("waitingForReplyEmpty", () => {
-          resolve();
-        });
-      }
-    });
-    // 4 [EVENT] WaitingForReply commands completed
+    await this.commandsQueue.waitForInflightCommandsToComplete();
+    // 4 [EVENT] In-flight commands completed
 
     // 5 + 6
     this.emit('resume', newSocket);