feat(sentinel): add sSubscribe/sUnsubscribe methods to Sentinel client (redis#3178)

Add sharded pub/sub methods (sSubscribe/sUnsubscribe) to the Sentinel client
for API consistency with the standalone Redis client. Since Sentinel manages
a single master with no sharding, these methods provide no sharding benefit -
they simply pass through to the underlying Redis commands. This enables
compatibility with libraries like @socket.io/redis-adapter that expect
the sSubscribe method to be available.

fixes redis#3177

Author: nkaradzhov

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

@@ -241,6 +241,31 @@ describe('RedisSentinel', () => {
 
       assert.equal(tester, false);
   }, testOptions)
+
+    testUtils.testWithClientSentinel('plain pubsub - sharded', async sentinel => {
+      let pubSubResolve;
+      const pubSubPromise = new Promise((res) => {
+        pubSubResolve = res;
+      });
+
+      let tester = false;
+      await sentinel.sSubscribe('test', () => {
+        tester = true;
+        pubSubResolve && pubSubResolve(1);
+      })
+
+      await sentinel.sPublish('test', 'hello world');
+      await pubSubPromise;
+      assert.equal(tester, true);
+
+      // now unsubscribe
+      tester = false;
+      await sentinel.sUnsubscribe('test')
+      await sentinel.sPublish('test', 'hello world');
+      await setTimeout(1000);
+
+      assert.equal(tester, false);
+    }, testOptions);
   });
 });
 

packages/client/lib/sentinel/index.ts

@@ -552,6 +552,26 @@ export default class RedisSentinel<
 
   pUnsubscribe = this.PUNSUBSCRIBE;
 
+  async SSUBSCRIBE<T extends boolean = false>(
+    channels: string | Array<string>,
+    listener: PubSubListener<T>,
+    bufferMode?: T
+  ) {
+    return this._self.#internal.sSubscribe(channels, listener, bufferMode);
+  }
+
+  sSubscribe = this.SSUBSCRIBE;
+
+  async SUNSUBSCRIBE<T extends boolean = false>(
+    channels?: string | Array<string>,
+    listener?: PubSubListener<T>,
+    bufferMode?: T
+  ) {
+    return this._self.#internal.sUnsubscribe(channels, listener, bufferMode);
+  }
+
+  sUnsubscribe = this.SUNSUBSCRIBE;
+
   /**
    * Acquires a master client lease for exclusive operations
    *
@@ -1066,6 +1086,22 @@ class RedisSentinelInternal<
     return this.#pubSubProxy.pUnsubscribe(patterns, listener, bufferMode);
   }
 
+  async sSubscribe<T extends boolean = false>(
+    channels: string | Array<string>,
+    listener: PubSubListener<T>,
+    bufferMode?: T
+  ) {
+    return this.#pubSubProxy.sSubscribe(channels, listener, bufferMode);
+  }
+
+  async sUnsubscribe<T extends boolean = false>(
+    channels?: string | Array<string>,
+    listener?: PubSubListener<T>,
+    bufferMode?: T
+  ) {
+    return this.#pubSubProxy.sUnsubscribe(channels, listener, bufferMode);
+  }
+
   // observe/analyze/transform remediation functions
   async observe() {
     for (const node of this.#sentinelRootNodes) {

packages/client/lib/sentinel/pub-sub-proxy.ts

@@ -14,7 +14,7 @@ type Client = RedisClient<
 >;
 
 type Subscriptions = Record<
-  PUBSUB_TYPE['CHANNELS'] | PUBSUB_TYPE['PATTERNS'],
+  PUBSUB_TYPE['CHANNELS'] | PUBSUB_TYPE['PATTERNS'] | PUBSUB_TYPE['SHARDED'],
   PubSubTypeListeners
 >;
 
@@ -71,7 +71,8 @@ export class PubSubProxy extends EventEmitter {
         if (withSubscriptions && this.#subscriptions) {
           await Promise.all([
             client.extendPubSubListeners(PUBSUB_TYPE.CHANNELS, this.#subscriptions[PUBSUB_TYPE.CHANNELS]),
-            client.extendPubSubListeners(PUBSUB_TYPE.PATTERNS, this.#subscriptions[PUBSUB_TYPE.PATTERNS])
+            client.extendPubSubListeners(PUBSUB_TYPE.PATTERNS, this.#subscriptions[PUBSUB_TYPE.PATTERNS]),
+            client.extendPubSubListeners(PUBSUB_TYPE.SHARDED, this.#subscriptions[PUBSUB_TYPE.SHARDED])
           ]);
         }
 
@@ -116,7 +117,8 @@ export class PubSubProxy extends EventEmitter {
     if (this.#state.connectPromise === undefined) {
       this.#subscriptions = {
         [PUBSUB_TYPE.CHANNELS]: this.#state.client.getPubSubListeners(PUBSUB_TYPE.CHANNELS),
-        [PUBSUB_TYPE.PATTERNS]: this.#state.client.getPubSubListeners(PUBSUB_TYPE.PATTERNS)
+        [PUBSUB_TYPE.PATTERNS]: this.#state.client.getPubSubListeners(PUBSUB_TYPE.PATTERNS),
+        [PUBSUB_TYPE.SHARDED]: this.#state.client.getPubSubListeners(PUBSUB_TYPE.SHARDED)
       };
 
       this.#state.client.destroy();
@@ -195,6 +197,24 @@ export class PubSubProxy extends EventEmitter {
     return this.#unsubscribe(client => client.PUNSUBSCRIBE(patterns, listener, bufferMode));
   }
 
+  sSubscribe<T extends boolean = false>(
+    channels: string | Array<string>,
+    listener: PubSubListener<T>,
+    bufferMode?: T
+  ) {
+    return this.#executeCommand(
+      client => client.SSUBSCRIBE(channels, listener, bufferMode)
+    );
+  }
+
+  async sUnsubscribe<T extends boolean = false>(
+    channels?: string | Array<string>,
+    listener?: PubSubListener<T>,
+    bufferMode?: T
+  ) {
+    return this.#unsubscribe(client => client.SUNSUBSCRIBE(channels, listener, bufferMode));
+  }
+
   destroy() {
     this.#subscriptions = undefined;
     if (this.#state === undefined) return;