fix #2345 - allow PING in PubSub mode (remove client side validation)

Author: leibale

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

@@ -10,7 +10,6 @@ export interface QueueCommandOptions {
     chainId?: symbol;
     signal?: AbortSignal;
     returnBuffers?: boolean;
-    ignorePubSubMode?: boolean;
 }
 
 export interface CommandWaitingToBeSent extends CommandWaitingForReply {
@@ -31,6 +30,8 @@ interface CommandWaitingForReply {
 
 type PubSubCommand = ReturnType<typeof PubSub.prototype.subscribe | typeof PubSub.prototype.unsubscribe>;
 
+const PONG = Buffer.from('pong');
+
 export default class RedisCommandsQueue {
     static #flushQueue<T extends CommandWaitingForReply>(queue: LinkedList<T>, err: Error): void {
         while (queue.length) {
@@ -53,24 +54,25 @@ export default class RedisCommandsQueue {
         },
         onReply: reply => {
             if (this.#pubSub.isActive && Array.isArray(reply)) {
-                if (
-                    !this.#pubSub.handleMessageReply(reply as Array<Buffer>) &&
-                    this.#pubSub.isStatusReply(reply as Array<Buffer>)
-                ) {
+                if (this.#pubSub.handleMessageReply(reply as Array<Buffer>)) return;
+                if (this.#pubSub.isStatusReply(reply as Array<Buffer>)) {
                     const head = this.#waitingForReply.head!.value;
                     if (
                         (Number.isNaN(head.channelsCounter!) && reply[2] === 0) ||
                         --head.channelsCounter! === 0
                     ) {
                         this.#waitingForReply.shift()!.resolve();
                     }
+                    return;
+                }
+                if (PONG.equals(reply[0] as Buffer)) {
+                    const { resolve, returnBuffers } = this.#waitingForReply.shift()!,
+                        buffer = ((reply[1] as Buffer).length === 0 ? reply[0] : reply[1]) as Buffer;
+                    resolve(returnBuffers ? buffer : buffer.toString());
+                    return;
                 }
-                
-                return;
-            } else if (!this.#waitingForReply.length) {
-                throw new Error('Got an unexpected reply from Redis');
             }
-
+            
             const { resolve, reject } = this.#waitingForReply.shift()!;
             if (reply instanceof ErrorReply) {
                 reject(reply);
@@ -85,9 +87,7 @@ export default class RedisCommandsQueue {
     }
 
     addCommand<T = RedisCommandRawReply>(args: RedisCommandArguments, options?: QueueCommandOptions): Promise<T> {
-        if (this.#pubSub.isActive && !options?.ignorePubSubMode) {
-            return Promise.reject(new Error('Cannot send commands in PubSub mode'));
-        } else if (this.#maxLength && this.#waitingToBeSent.length + this.#waitingForReply.length >= this.#maxLength) {
+        if (this.#maxLength && this.#waitingToBeSent.length + this.#waitingForReply.length >= this.#maxLength) {
             return Promise.reject(new Error('The queue is full'));
         } else if (options?.signal?.aborted) {
             return Promise.reject(new AbortError());

packages/client/lib/client/index.spec.ts

@@ -675,7 +675,7 @@ describe('Client', () => {
         );
     }, GLOBAL.SERVERS.OPEN);
 
-    describe('PubSub', () => {
+    describe.only('PubSub', () => {
         testUtils.testWithClient('should be able to publish and subscribe to messages', async publisher => {
             function assertStringListener(message: string, channel: string) {
                 assert.equal(typeof message, 'string');
@@ -793,7 +793,34 @@ describe('Client', () => {
             }
         }, GLOBAL.SERVERS.OPEN);
 
-        testUtils.testWithClient('should be able to quit in PubSub mode', async client => {
+        testUtils.testWithClient('should be able to PING in PubSub mode', async client => {
+            await client.connect();
+
+            try {
+                await client.subscribe('channel', () => {
+                    // noop
+                });
+                
+                const [string, buffer, customString, customBuffer] = await Promise.all([
+                    client.ping(),
+                    client.ping(client.commandOptions({ returnBuffers: true })),
+                    client.ping('custom'),
+                    client.ping(client.commandOptions({ returnBuffers: true }), 'custom')
+                ]);
+    
+                assert.equal(string, 'pong');
+                assert.deepEqual(buffer, Buffer.from('pong'));
+                assert.equal(customString, 'custom');
+                assert.deepEqual(customBuffer, Buffer.from('custom'));
+            } finally {
+                await client.disconnect();
+            }
+        }, {
+            ...GLOBAL.SERVERS.OPEN,
+            disableClientSetup: true
+        });
+
+        testUtils.testWithClient('should be able to QUIT in PubSub mode', async client => {
             await client.subscribe('channel', () => {
                 // noop
             });
@@ -803,6 +830,24 @@ describe('Client', () => {
             assert.equal(client.isOpen, false);
         }, GLOBAL.SERVERS.OPEN);
 
+
+        testUtils.testWithClient('should reject GET in PubSub mode', async client => {
+            await client.connect();
+
+            try {
+                await client.subscribe('channel', () => {
+                    // noop
+                });
+
+                await assert.rejects(client.get('key'));
+            } finally {
+                await client.disconnect();
+            }
+        }, {
+            ...GLOBAL.SERVERS.OPEN,
+            disableClientSetup: true
+        });
+
         testUtils.testWithClient('shareded PubSub', async publisher => {
             const subscriber = publisher.duplicate();
 
@@ -862,7 +907,7 @@ describe('Client', () => {
             }
         }, {
             // this test change ACL rules, running in isolated server
-            ...GLOBAL.SERVERS.OPEN
+            serverArguments: []
         });
     });
 

packages/client/lib/client/index.ts

@@ -603,9 +603,7 @@ export default class RedisClient<
 
     QUIT(): Promise<void> {
         return this.#socket.quit(() => {
-            const quitPromise = this.#queue.addCommand(['QUIT'], {
-                ignorePubSubMode: true
-            });
+            const quitPromise = this.#queue.addCommand(['QUIT']);
             this.#tick();
             return Promise.all([
                 quitPromise,

packages/client/lib/commands/PING.spec.ts

@@ -1,8 +1,24 @@
 import { strict as assert } from 'assert';
 import testUtils, { GLOBAL } from '../test-utils';
-import RedisClient from '../client';
+import { transformArguments } from './PING';
 
 describe('PING', () => {
+    describe('transformArguments', () => {
+        it('default', () => {
+            assert.deepEqual(
+                transformArguments(),
+                ['PING']
+            );
+        });
+
+        it('with message', () => {
+            assert.deepEqual(
+                transformArguments('message'),
+                ['PING', 'message']
+            );
+        });
+    });
+
     describe('client.ping', () => {
         testUtils.testWithClient('string', async client => {
             assert.equal(
@@ -13,7 +29,7 @@ describe('PING', () => {
 
         testUtils.testWithClient('buffer', async client => {
             assert.deepEqual(
-                await client.ping(RedisClient.commandOptions({ returnBuffers: true })),
+                await client.ping(client.commandOptions({ returnBuffers: true })),
                 Buffer.from('PONG')
             );
         }, GLOBAL.SERVERS.OPEN);

packages/client/lib/commands/PING.ts

@@ -1,7 +1,12 @@
-import { RedisCommandArgument } from '.';
+import { RedisCommandArgument, RedisCommandArguments } from '.';
 
-export function transformArguments(): Array<string> {
-    return ['PING'];
+export function transformArguments(message?: RedisCommandArgument): RedisCommandArguments {
+    const args: RedisCommandArguments = ['PING'];
+    if (message) {
+        args.push(message);
+    }
+
+    return args;
 }
 
 export declare function transformReply(): RedisCommandArgument;