--wip-- [skip ci]

Author: nkaradzhov

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

@@ -0,0 +1,50 @@
+import { RedisClientOptions } from ".";
+import RedisCommandsQueue from "./commands-queue";
+import RedisSocket from "./socket";
+
+export default class EnterpriseMaintenanceManager {
+  client: any;
+  commandsQueue: RedisCommandsQueue;
+  options: RedisClientOptions;
+  constructor(
+    client: any,
+    commandsQueue: RedisCommandsQueue,
+    options: RedisClientOptions,
+  ) {
+    this.client = client;
+    this.commandsQueue = commandsQueue;
+    this.options = options;
+
+    this.commandsQueue.events.on("moving", this.#onMoving);
+  }
+
+  #onMoving = async (_afterMs: number, host: string, port: number) => {
+
+    this.client.pause()
+
+    const socket = new RedisSocket({
+      ...this.options.socket,
+      host,
+      port
+    });
+    await socket.connect();
+
+    //wait until waitingForReply is empty
+    await new Promise<void>(resolve => {
+      if(!this.commandsQueue.isWaitingForReply()) {
+        resolve()
+      } else {
+        this.commandsQueue.events.once('waitingForReplyEmpty', resolve)
+      }
+    })
+
+    const oldSocket = this.client.socket
+    oldSocket.removeAllListeners();
+    oldSocket.destroy();
+
+    this.client.socket = socket;
+
+    this.client.resume()
+  };
+
+}

packages/client/lib/client/index.ts

@@ -460,6 +460,15 @@ export default class RedisClient<
     return this._self.#dirtyWatch !== undefined
   }
 
+  get socket() {
+    return this._self.#socket;
+  }
+
+  set socket(socket: RedisSocket) {
+    this._self.#socket = socket;
+    this.#initiateSocket();
+  }
+
   /**
    * Marks the client's WATCH command as invalidated due to a topology change.
    * This will cause any subsequent EXEC in a transaction to fail with a WatchError.
@@ -557,6 +566,7 @@ export default class RedisClient<
     }
 
   }
+
   #initiateOptions(options?: RedisClientOptions<M, F, S, RESP, TYPE_MAPPING>): RedisClientOptions<M, F, S, RESP, TYPE_MAPPING> | undefined {
 
     // Convert username/password to credentialsProvider if no credentialsProvider is already in place
@@ -766,6 +776,29 @@ export default class RedisClient<
 
   async #initiateSocket(options?: RedisClientOptions<M, F, S, RESP, TYPE_MAPPING>): Promise<void> {
     await this.#socket.waitForReady();
+
+    this.#socket
+      .on('data', chunk => {
+        try {
+          this.#queue.decoder.write(chunk);
+        } catch (err) {
+          this.#queue.resetDecoder();
+          this.emit('error', err);
+        }
+      })
+      .on('error', err => {
+        this.emit('error', err);
+        this.#clientSideCache?.onError();
+        if (this.#socket.isOpen && !this.#options?.disableOfflineQueue) {
+          this.#queue.flushWaitingForReply(err);
+        } else {
+          this.#queue.flushAll(err);
+        }
+      })
+      .on('reconnecting', () => this.emit('reconnecting'))
+      .on('drain', () => this.#maybeScheduleWrite())
+      .on('end', () => this.emit('end'));
+
     console.log('Initiator...');
     const promises = [];
     const chainId = Symbol('Socket Initiator');
@@ -800,31 +833,11 @@ export default class RedisClient<
 
   #createSocket(options?: RedisClientOptions<M, F, S, RESP, TYPE_MAPPING>): RedisSocket {
     return new RedisSocket(options?.socket)
-      .on('data', chunk => {
-        try {
-          this.#queue.decoder.write(chunk);
-        } catch (err) {
-          this.#queue.resetDecoder();
-          this.emit('error', err);
-        }
-      })
-      .on('error', err => {
-        this.emit('error', err);
-        this.#clientSideCache?.onError();
-        if (this.#socket.isOpen && !this.#options?.disableOfflineQueue) {
-          this.#queue.flushWaitingForReply(err);
-        } else {
-          this.#queue.flushAll(err);
-        }
-      })
       .on('connect', () => this.emit('connect'))
       .on('ready', () => {
         console.log('Socket ready');
         this.emit('ready');
-      })
-      .on('reconnecting', () => this.emit('reconnecting'))
-      .on('drain', () => this.#maybeScheduleWrite())
-      .on('end', () => this.emit('end'));
+      });
   }
 
   #pingTimer?: NodeJS.Timeout;