Force slots refresh on MOVED error when using ssubscribe

Author: htemelski-redis

lib/DataHandler.ts

@@ -74,6 +74,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);
   }
 
@@ -119,7 +124,11 @@ export default class DataHandler {
       case "message":
         if (this.redis.listeners("message").length > 0) {
           // Check if there're listeners to avoid unnecessary `toString()`.
-          this.redis.emit("message", reply[1].toString(), reply[2] ? reply[2].toString() : '');
+          this.redis.emit(
+            "message",
+            reply[1].toString(),
+            reply[2] ? reply[2].toString() : ""
+          );
         }
         this.redis.emit("messageBuffer", reply[1], reply[2]);
         break;

lib/cluster/ClusterSubscriber.ts

@@ -95,6 +95,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");
+    });
+
     // Re-subscribe previous channels
     const previousChannels = { subscribe: [], psubscribe: [], ssubscribe: [] };
     if (lastActiveSubscriber) {

lib/cluster/ClusterSubscriberGroup.ts

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

lib/cluster/index.ts

@@ -110,7 +110,10 @@ class Cluster extends EventEmitter {
     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)
+      );
 
     // validate options
     if (