fix client.quit, add error events on cluster, fix some "deepsource.io" warnings

Author: leibale

lib/client.ts

@@ -298,9 +298,10 @@ export default class RedisClient<M extends RedisModules = RedisModules, S extend
     }
 
     QUIT(): Promise<void> {
-        return this.#socket.quit(async () => {
-            this.#queue.addEncodedCommand(encodeCommand(['QUIT']));
+        return this.#socket.quit(() => {
+            const promise = this.#queue.addEncodedCommand(encodeCommand(['QUIT']));
             this.#tick();
+            return promise;
         });
     }
 

lib/cluster-slots.ts

@@ -17,54 +17,40 @@ interface SlotNodes<M extends RedisModules, S extends RedisLuaScripts> {
     clientIterator: IterableIterator<RedisClientType<M, S>> | undefined;
 }
 
+type OnError = (err: unknown) => void;
+
 export default class RedisClusterSlots<M extends RedisModules, S extends RedisLuaScripts> {
     readonly #options: RedisClusterOptions;
+    readonly #onError: OnError;
     readonly #nodeByUrl = new Map<string, ClusterNode<M, S>>();
     readonly #slots: Array<SlotNodes<M, S>> = [];
 
-    constructor(options: RedisClusterOptions) {
+    constructor(options: RedisClusterOptions, onError: OnError) {
         this.#options = options;
+        this.#onError = onError;
     }
 
     async connect(): Promise<void> {
         for (const rootNode of this.#options.rootNodes) {
-            try {
-                await this.#discoverNodes(rootNode);
-                return;
-            } catch (err) {
-                console.error(err);
-                // this.emit('error', err);
-            }
+            if (await this.#discoverNodes(rootNode)) return;
         }
 
         throw new Error('None of the root nodes is available');
     }
 
     async discover(startWith: RedisClientType<M, S>): Promise<void> {
-        try {
-            await this.#discoverNodes(startWith.options?.socket);
-            return;
-        } catch (err) {
-            console.error(err);
-            // this.emit('error', err);
-        }
+        if (await this.#discoverNodes(startWith.options?.socket)) return;
 
         for (const { client } of this.#nodeByUrl.values()) {
             if (client === startWith) continue;
-            
-            try {
-                await this.#discoverNodes(client.options?.socket);
-                return;
-            } catch (err) {
-                console.error(err);
-                // this.emit('error', err);
-            }
+
+            if (await this.#discoverNodes(client.options?.socket)) return;
         }
 
         throw new Error('None of the cluster nodes is available');
     }
 
-    async #discoverNodes(socketOptions?: RedisSocketOptions): Promise<void> {
+    async #discoverNodes(socketOptions?: RedisSocketOptions): Promise<boolean> {
         const client = RedisClient.create({
             socket: socketOptions
         });
@@ -73,8 +59,14 @@ export default class RedisClusterSlots<M extends RedisModules, S extends RedisLu
 
         try {
             await this.#reset(await client.clusterNodes());
+            return true;
+        } catch (err) {
+            this.#onError(err);
+            return false;
         } finally {
-            await client.disconnect(); // TODO: catch error from disconnect?
+            if (client.isOpen) {
+                await client.disconnect();
+            }
         }
     }
 
@@ -102,7 +94,6 @@ export default class RedisClusterSlots<M extends RedisModules, S extends RedisLu
         for (const [url, { client }] of this.#nodeByUrl.entries()) {
             if (clientsInUse.has(url)) continue;
 
-            // TODO: ignore error from `.disconnect`?
             promises.push(client.disconnect());
             this.#nodeByUrl.delete(url);
         }

lib/cluster.ts

@@ -5,6 +5,7 @@ import RedisClusterSlots, { ClusterNode } from './cluster-slots';
 import { RedisLuaScript, RedisLuaScripts } from './lua-script';
 import { extendWithModulesAndScripts, extendWithDefaultCommands, transformCommandArguments } from './commander';
 import RedisMultiCommand, { MultiQueuedCommand, RedisMultiCommandType } from './multi-command';
+import { EventEmitter } from 'events';
 
 export interface RedisClusterOptions<M = RedisModules, S = RedisLuaScripts> {
     rootNodes: Array<RedisSocketOptions>;
@@ -17,7 +18,7 @@ export interface RedisClusterOptions<M = RedisModules, S = RedisLuaScripts> {
 export type RedisClusterType<M extends RedisModules, S extends RedisLuaScripts> =
     WithPlugins<M, S> & RedisCluster;
 
-export default class RedisCluster<M extends RedisModules = RedisModules, S extends RedisLuaScripts = RedisLuaScripts> {
+export default class RedisCluster<M extends RedisModules = RedisModules, S extends RedisLuaScripts = RedisLuaScripts> extends EventEmitter {
     static #extractFirstKey(command: RedisCommand, originalArgs: Array<unknown>, redisArgs: Array<string>): string | undefined {
         if (command.FIRST_KEY_INDEX === undefined) {
             return undefined;
@@ -83,8 +84,10 @@ export default class RedisCluster<M extends RedisModules = RedisModules, S exten
     readonly #Multi: new (...args: ConstructorParameters<typeof RedisMultiCommand>) => RedisMultiCommandType<M, S>;
 
     constructor(options: RedisClusterOptions<M, S>) {
+        super();
+
         this.#options = options;
-        this.#slots = new RedisClusterSlots(options);
+        this.#slots = new RedisClusterSlots(options, err => this.emit('error', err));
         this.#Multi = RedisMultiCommand.extend(options);
     }
 

lib/commander.ts

@@ -97,12 +97,12 @@ export function transformCommandArguments<T = unknown>(
 export function encodeCommand(args: Array<string>): string {
     const encoded = [
         `*${args.length}`,
-        `$${Buffer.byteLength(args[0])}`,
+        `$${Buffer.byteLength(args[0]).toString()}`,
         args[0]
     ];
 
     for (let i = 1; i < args.length; i++) {
-        encoded.push(`$${Buffer.byteLength(args[i])}`, args[i]);
+        encoded.push(`$${Buffer.byteLength(args[i]).toString()}`, args[i]);
     }
 
     return encoded.join('\r\n') + '\r\n';

lib/commands-queue.ts

@@ -12,6 +12,7 @@ export interface QueueCommandOptions {
 
 interface CommandWaitingToBeSent extends CommandWaitingForReply {
     encodedCommand: string;
+    byteLength: number;
     chainId?: symbol;
     abort?: {
         signal: any; // TODO: `AbortSignal` type is incorrect
@@ -130,6 +131,7 @@ export default class RedisCommandsQueue {
         return new Promise((resolve, reject) => {
             const node = new LinkedList.Node<CommandWaitingToBeSent>({
                 encodedCommand,
+                byteLength: Buffer.byteLength(encodedCommand),
                 chainId: options?.chainId,
                 resolve,
                 reject
@@ -156,7 +158,7 @@ export default class RedisCommandsQueue {
                 this.#waitingToBeSent.pushNode(node);
             }
 
-            this.#waitingToBeSentCommandsLength += encodedCommand.length;
+            this.#waitingToBeSentCommandsLength += node.value.byteLength;
         });
     }
 
@@ -230,8 +232,12 @@ export default class RedisCommandsQueue {
             }
 
             this.#pubSubState[inProgressKey] += channelsCounter;
+
+            const encodedCommand = encodeCommand(commandArgs),
+                byteLength = Buffer.byteLength(encodedCommand);
             this.#waitingToBeSent.push({
-                encodedCommand: encodeCommand(commandArgs),
+                encodedCommand,
+                byteLength,
                 channelsCounter,
                 resolve: () => {
                     this.#pubSubState[inProgressKey] -= channelsCounter;
@@ -243,6 +249,7 @@ export default class RedisCommandsQueue {
                     reject();
                 }
             });
+            this.#waitingToBeSentCommandsLength += byteLength;
         });
     }
 
@@ -268,7 +275,7 @@ export default class RedisCommandsQueue {
             lastCommandChainId: symbol | undefined;
         for (const command of this.#waitingToBeSent) {
             encoded.push(command.encodedCommand);
-            size += command.encodedCommand.length;
+            size += command.byteLength;
             if (size > recommendedSize) {
                 lastCommandChainId = command.chainId;
                 break;

lib/commands/MIGRATE.ts

@@ -19,7 +19,7 @@ export function transformArguments(
         isKeyString = typeof key === 'string';
 
     if (isKeyString) {
-        args.push(key as string);
+        args.push(key);
     } else {
         args.push('""');
     }

lib/commands/generic-transformers.spec.ts

@@ -1,5 +1,4 @@
 import { strict as assert } from 'assert';
-import { isObject } from 'util';
 import {
     transformReplyBoolean,
     transformReplyBooleanArray,