refactor(client): Simplify socket initialization and improve readiness handling

- Changed `#initiateSocket` to a streamlined asynchronous method to handle socket readiness and initialization logic.
- Added `waitForReady` method to `RedisSocket` to await socket readiness.
- Introduced `#createSocket` method in `RedisClient` to decouple socket creation from initialization logic.
- Refactored `connect` method to ensure proper socket readiness and initialization before proceeding.

Author: nkaradzhov

packages/client/lib/client/index.ts

@@ -455,7 +455,7 @@ export default class RedisClient<
     this.#validateOptions(options)
     this.#options = this.#initiateOptions(options);
     this.#queue = this.#initiateQueue();
-    this.#socket = this.#initiateSocket();
+    this.#socket = this.#createSocket(this.#options);
 
     if (options?.clientSideCache) {
       if (options.clientSideCache instanceof ClientSideCacheProvider) {
@@ -688,38 +688,10 @@ export default class RedisClient<
     return commands;
   }
 
-  #initiateSocket(): RedisSocket {
-    const socketInitiator = async () => {
-      const promises = [],
-        chainId = Symbol('Socket Initiator');
+  async #initiateSocket(): Promise<void> {
+    await this.#socket.waitForReady();
 
-      const resubscribePromise = this.#queue.resubscribe(chainId);
-      if (resubscribePromise) {
-        promises.push(resubscribePromise);
-      }
-
-      if (this.#monitorCallback) {
-        promises.push(
-          this.#queue.monitor(
-            this.#monitorCallback,
-            {
-              typeMapping: this._commandOptions?.typeMapping,
-              chainId,
-              asap: true
-            }
-          )
-        );
-      }
-
-      promises.push(...(await this.#handshake(chainId, true)));
-
-      if (promises.length) {
-        this.#write();
-        return Promise.all(promises);
-      }
-    };
-
-    return new RedisSocket(socketInitiator, this.#options?.socket)
+    this.#socket
       .on('data', chunk => {
         try {
           this.#queue.decoder.write(chunk);
@@ -737,15 +709,47 @@ export default class RedisClient<
           this.#queue.flushAll(err);
         }
       })
-      .on('connect', () => this.emit('connect'))
-      .on('ready', () => {
-        this.emit('ready');
-        this.#setPingTimer();
-        this.#maybeScheduleWrite();
-      })
       .on('reconnecting', () => this.emit('reconnecting'))
       .on('drain', () => this.#maybeScheduleWrite())
       .on('end', () => this.emit('end'));
+
+    const promises = [];
+    const chainId = Symbol('Socket Initiator');
+
+    const resubscribePromise = this.#queue.resubscribe(chainId);
+    if (resubscribePromise) {
+      promises.push(resubscribePromise);
+    }
+
+    if (this.#monitorCallback) {
+      promises.push(
+        this.#queue.monitor(
+          this.#monitorCallback,
+          {
+            typeMapping: this._commandOptions?.typeMapping,
+            chainId,
+            asap: true
+          }
+        )
+      );
+    }
+
+    promises.push(...(await this.#handshake(chainId, true)));
+
+    this.#setPingTimer();
+
+    if (promises.length) {
+      this.#write();
+      await Promise.all(promises);
+    }
+  }
+
+  #createSocket(options?: RedisClientOptions<M, F, S, RESP, TYPE_MAPPING>): RedisSocket {
+    return new RedisSocket(options?.socket)
+      .on('connect', () => this.emit('connect'))
+      .on('ready', () => {
+        this.emit('ready');
+      });
   }
 
   #pingTimer?: NodeJS.Timeout;
@@ -854,6 +858,7 @@ export default class RedisClient<
 
   async connect() {
     await this._self.#socket.connect();
+    await this._self.#initiateSocket();
     return this as unknown as RedisClientType<M, F, S, RESP, TYPE_MAPPING>;
   }
 

packages/client/lib/client/socket.ts

@@ -51,10 +51,7 @@ export type RedisTcpSocketOptions = RedisTcpOptions | RedisTlsOptions;
 
 export type RedisSocketOptions = RedisTcpSocketOptions | RedisIpcOptions;
 
-export type RedisSocketInitiator = () => void | Promise<unknown>;
-
 export default class RedisSocket extends EventEmitter {
-  readonly #initiator;
   readonly #connectTimeout;
   readonly #reconnectStrategy;
   readonly #socketFactory;
@@ -85,13 +82,20 @@ export default class RedisSocket extends EventEmitter {
   constructor(initiator: RedisSocketInitiator, options?: RedisSocketOptions) {
     super();
 
-    this.#initiator = initiator;
     this.#connectTimeout = options?.connectTimeout ?? 5000;
     this.#reconnectStrategy = this.#createReconnectStrategy(options);
     this.#socketFactory = this.#createSocketFactory(options);
     this.#socketTimeout = options?.socketTimeout;
   }
 
+  async waitForReady(): Promise<void> {
+    if (this.#isReady) return
+    return new Promise((resolve, reject) => {
+      this.once('ready', resolve);
+      this.once('error', reject);
+    });
+  }
+
   #createReconnectStrategy(options?: RedisSocketOptions): ReconnectStrategyFunction {
     const strategy = options?.reconnectStrategy;
     if (strategy === false || typeof strategy === 'number') {
@@ -214,14 +218,6 @@ export default class RedisSocket extends EventEmitter {
       try {
         this.#socket = await this.#createSocket();
         this.emit('connect');
-
-        try {
-          await this.#initiator();
-        } catch (err) {
-          this.#socket.destroy();
-          this.#socket = undefined;
-          throw err;
-        }
         this.#isReady = true;
         this.#socketEpoch++;
         this.emit('ready');