refactor: minor cleanup

Author: darrachequesne

lib/index.ts

@@ -1,6 +1,7 @@
 import uid2 = require("uid2");
 import msgpack = require("notepack.io");
 import { Adapter, BroadcastOptions, Room } from "socket.io-adapter";
+import { parseNumSubResponse, sumValues } from "./util";
 
 const debug = require("debug")("socket.io-redis");
 
@@ -688,7 +689,7 @@ export class RedisAdapter extends Adapter {
    */
   public async allRooms(): Promise<Set<Room>> {
     const localRooms = new Set(this.rooms.keys());
-    const numSub = await this.getNumSub();
+    const numSub = await this.serverCount();
     debug('waiting for %d responses to "allRooms" request', numSub);
 
     if (numSub <= 1) {
@@ -732,7 +733,7 @@ export class RedisAdapter extends Adapter {
       return localSockets;
     }
 
-    const numSub = await this.getNumSub();
+    const numSub = await this.serverCount();
     debug('waiting for %d responses to "fetchSockets" request', numSub);
 
     if (numSub <= 1) {
@@ -849,7 +850,7 @@ export class RedisAdapter extends Adapter {
 
   private async serverSideEmitWithAck(packet: any[]) {
     const ack = packet.pop();
-    const numSub = (await this.getNumSub()) - 1; // ignore self
+    const numSub = (await this.serverCount()) - 1; // ignore self
 
     debug('waiting for %d responses to "serverSideEmit" request', numSub);
 
@@ -889,13 +890,7 @@ export class RedisAdapter extends Adapter {
     this.pubClient.publish(this.requestChannel, request);
   }
 
-  /**
-   * Get the number of subscribers of the request channel
-   *
-   * @private
-   */
-
-  private getNumSub(): Promise<number> {
+  override serverCount(): Promise<number> {
     if (
       this.pubClient.constructor.name === "Cluster" ||
       this.pubClient.isCluster
@@ -904,39 +899,27 @@ export class RedisAdapter extends Adapter {
       const nodes = this.pubClient.nodes();
       return Promise.all(
         nodes.map((node) =>
-          node.send_command("pubsub", ["numsub", this.requestChannel])
+          node
+            .send_command("pubsub", ["numsub", this.requestChannel])
+            .then(parseNumSubResponse)
         )
-      ).then((values) => {
-        let numSub = 0;
-        values.forEach((value) => {
-          numSub += parseInt(value[1], 10);
-        });
-        return numSub;
-      });
+      ).then(sumValues);
     } else if (typeof this.pubClient.pSubscribe === "function") {
       // node-redis client
       const isCluster = Array.isArray(this.pubClient.masters);
       if (isCluster) {
         const nodes = this.pubClient.masters;
         return Promise.all(
           nodes.map((node) => {
-            return node.client.sendCommand([
-              "pubsub",
-              "numsub",
-              this.requestChannel,
-            ]);
+            return node.client
+              .sendCommand(["pubsub", "numsub", this.requestChannel])
+              .then(parseNumSubResponse);
           })
-        ).then((values) => {
-          let numSub = 0;
-          values.map((value) => {
-            numSub += parseInt(value[1], 10);
-          });
-          return numSub;
-        });
+        ).then(sumValues);
       } else {
         return this.pubClient
           .sendCommand(["pubsub", "numsub", this.requestChannel])
-          .then((res) => parseInt(res[1], 10));
+          .then(parseNumSubResponse);
       }
     } else {
       // ioredis or node-redis v3 client
@@ -946,17 +929,13 @@ export class RedisAdapter extends Adapter {
           ["numsub", this.requestChannel],
           (err, numSub) => {
             if (err) return reject(err);
-            resolve(parseInt(numSub[1], 10));
+            resolve(parseNumSubResponse(numSub));
           }
         );
       });
     }
   }
 
-  serverCount(): Promise<number> {
-    return this.getNumSub();
-  }
-
   close(): Promise<void> | void {
     const isRedisV4 = typeof this.pubClient.pSubscribe === "function";
     if (isRedisV4) {

lib/sharded-adapter.ts

@@ -1,6 +1,6 @@
 import { ClusterAdapter, ClusterMessage, MessageType } from "./cluster-adapter";
 import { decode, encode } from "notepack.io";
-import { hasBinary } from "./util";
+import { hasBinary, parseNumSubResponse, sumValues } from "./util";
 import debugModule from "debug";
 
 const debug = debugModule("socket.io-redis");
@@ -60,7 +60,6 @@ class ShardedRedisAdapter extends ClusterAdapter {
   private readonly opts: Required<ShardedRedisAdapterOptions>;
   private readonly channel: string;
   private readonly responseChannel: string;
-  private readonly cleanup: () => void;
 
   constructor(nsp, pubClient, subClient, opts: ShardedRedisAdapterOptions) {
     super(nsp);
@@ -196,19 +195,15 @@ class ShardedRedisAdapter extends ClusterAdapter {
     ) {
       return Promise.all(
         this.pubClient.nodes().map((node) => {
-          return node.sendCommand(["PUBSUB", "SHARDNUMSUB", this.channel]);
+          return node
+            .sendCommand(["PUBSUB", "SHARDNUMSUB", this.channel])
+            .then(parseNumSubResponse);
         })
-      ).then((values) => {
-        let numSub = 0;
-        values.forEach((value) => {
-          numSub += parseInt(value[1], 10);
-        });
-        return numSub;
-      });
+      ).then(sumValues);
     } else {
       return this.pubClient
         .sendCommand(["PUBSUB", "SHARDNUMSUB", this.channel])
-        .then((res) => parseInt(res[1], 10));
+        .then(parseNumSubResponse);
     }
   }
 }

lib/util.ts

@@ -34,3 +34,13 @@ export function hasBinary(obj: any, toJSON?: boolean): boolean {
 export function randomId() {
   return randomBytes(8).toString("hex");
 }
+
+export function parseNumSubResponse(res) {
+  return parseInt(res[1], 10);
+}
+
+export function sumValues(values) {
+  return values.reduce((acc, val) => {
+    return acc + val;
+  }, 0);
+}