fix(cluster): subscriber connection leaks

Closes #1325

lib/cluster/ClusterSubscriber.ts

@@ -50,6 +50,10 @@ export default class ClusterSubscriber {
       lastActiveSubscriber.disconnect();
     }
 
+    if (this.subscriber) {
+      this.subscriber.disconnect();
+    }
+
     const sampleNode = sample(this.connectionPool.getNodes());
     if (!sampleNode) {
       debug(
@@ -113,7 +117,10 @@ export default class ClusterSubscriber {
                 this.lastActiveSubscriber = this.subscriber;
               }
             })
-            .catch(noop);
+            .catch(() => {
+              // TODO: should probably disconnect the subscriber and try again.
+              debug("failed to %s %d channels", type, channels.length);
+            });
         }
       }
     } else {

test/functional/cluster/ClusterSubscriber.ts

@@ -0,0 +1,36 @@
+import ConnectionPool from "../../../lib/cluster/ConnectionPool";
+import ClusterSubscriber from "../../../lib/cluster/ClusterSubscriber";
+import { EventEmitter } from "events";
+import MockServer from "../../helpers/mock_server";
+import { expect } from "chai";
+
+describe("ClusterSubscriber", () => {
+  it("cleans up subscribers when selecting a new one", async () => {
+    const pool = new ConnectionPool({});
+    const subscriber = new ClusterSubscriber(pool, new EventEmitter());
+
+    let rejectSubscribes = false;
+    const server = new MockServer(30000, (argv) => {
+      if (rejectSubscribes && argv[0] === "subscribe") {
+        return new Error("Failed to subscribe");
+      }
+      return "OK";
+    });
+
+    pool.findOrCreate({ host: "127.0.0.1", port: 30000 });
+
+    subscriber.start();
+    await subscriber.getInstance().subscribe("foo");
+    rejectSubscribes = true;
+
+    subscriber.start();
+    await subscriber.getInstance().echo("hello");
+
+    subscriber.start();
+    await subscriber.getInstance().echo("hello");
+
+    expect(server.getAllClients()).to.have.lengthOf(1);
+    subscriber.stop();
+    pool.reset([]);
+  });
+});

test/helpers/mock_server.ts

@@ -173,4 +173,8 @@ export default class MockServer extends EventEmitter {
         return getConnectionName(client) === name;
       });
   }
+
+  getAllClients(): Socket[] {
+    return this.clients.filter(Boolean);
+  }
 }