Force slots refresh on MOVED error when using ssubscribe (#2013)

Co-authored-by: Hristo Temelski <hristo.temelski@redis.com>

Author: PavelPashov

lib/DataHandler.ts

@@ -80,6 +80,11 @@ export default class DataHandler {
       args: item.command.args,
     };
 
+    if (item.command.name == "ssubscribe" && err.message.includes("MOVED")) {
+      this.redis.emit("moved");
+      return;
+    }
+
     this.redis.handleReconnection(err, item);
   }
 

lib/cluster/ClusterSubscriber.ts

@@ -160,6 +160,10 @@ export default class ClusterSubscriber {
     // Ignore the errors since they're handled in the connection pool.
     this.subscriber.on("error", noop);
 
+    this.subscriber.on("moved", () => {
+      this.emitter.emit("forceRefresh");
+    });
+
     // The node we lost connection to may not come back up in a
     // reasonable amount of time (e.g. a slave that's taken down
     // for maintainence), we could potentially miss many published

lib/cluster/ClusterSubscriberGroup.ts

@@ -31,7 +31,7 @@ export default class ClusterSubscriberGroup {
      *
      * @param cluster
      */
-    constructor(private cluster: Cluster) {
+    constructor(private cluster: Cluster, refreshSlotsCacheCallback: () => void) {
 
         cluster.on("+node", (redis) => {
             this._addSubscriber(redis);
@@ -44,6 +44,10 @@ export default class ClusterSubscriberGroup {
         cluster.on("refresh", () => {
             this._refreshSlots(cluster);
         });
+
+        cluster.on("forceRefresh", () => {
+          refreshSlotsCacheCallback();
+        });
     }
 
 

lib/cluster/index.ts

@@ -126,7 +126,7 @@ class Cluster extends Commander {
     this.options = defaults({}, options, DEFAULT_CLUSTER_OPTIONS, this.options);
 
     if (this.options.shardedSubscribers == true)
-      this.shardedSubscribers = new ClusterSubscriberGroup(this);
+      this.shardedSubscribers = new ClusterSubscriberGroup(this, this.refreshSlotsCache.bind(this));
 
     if (
       this.options.redisOptions &&