socket.io-redis - npm Package Compare versions

Comparing version 6.0.1 to 6.1.0

CHANGELOG.md

@@ -0,1 +1,14 @@
+# [6.1.0](https://github.com/socketio/socket.io-redis/compare/6.0.1...6.1.0) (2021-03-12)
+
+
+### Features
+
+* implement utility methods from Socket.IO v4 ([468c3c8](https://github.com/socketio/socket.io-redis/commit/468c3c8008ddd0c89b2fc2054d874e9e706f0948))
+
+
+### Performance Improvements
+
+* remove one round-trip for the requester ([6c8d770](https://github.com/socketio/socket.io-redis/commit/6c8d7701962bee4acf83568f8e998876d3549fb8))
+
+
 ## [6.0.1](https://github.com/socketio/socket.io-redis/compare/6.0.0...6.0.1) (2020-11-14)
@@ -2,0 +15,0 @@

dist/index.d.ts

@@ -115,2 +115,6 @@ import { Adapter, BroadcastOptions, Room, SocketId } from "socket.io-adapter";
     remoteDisconnect(id: SocketId, close?: boolean): Promise<void>;
+    fetchSockets(opts: BroadcastOptions): Promise<any[]>;
+    addSockets(opts: BroadcastOptions, rooms: Room[]): void;
+    delSockets(opts: BroadcastOptions, rooms: Room[]): void;
+    disconnectSockets(opts: BroadcastOptions, close: boolean): void;
     /**
@@ -117,0 +121,0 @@ * Get the number of subscribers of the request channel

dist/index.js

@@ -20,2 +20,3 @@ "use strict";
     RequestType[RequestType["REMOTE_DISCONNECT"] = 4] = "REMOTE_DISCONNECT";
+    RequestType[RequestType["REMOTE_FETCH"] = 5] = "REMOTE_FETCH";
 })(RequestType || (RequestType = {}));
@@ -129,2 +130,5 @@ function createRedisClient(uri, opts) {
             case RequestType.SOCKETS:
+                if (this.requests.has(request.requestId)) {
+                    return;
+                }
                 const sockets = await super.sockets(new Set(request.rooms));
@@ -138,2 +142,5 @@ response = JSON.stringify({
             case RequestType.ALL_ROOMS:
+                if (this.requests.has(request.requestId)) {
+                    return;
+                }
                 response = JSON.stringify({
@@ -146,2 +153,9 @@ requestId: request.requestId,
             case RequestType.REMOTE_JOIN:
+                if (request.opts) {
+                    const opts = {
+                        rooms: new Set(request.opts.rooms),
+                        except: new Set(request.opts.except),
+                    };
+                    return super.addSockets(opts, request.rooms);
+                }
                 socket = this.nsp.sockets.get(request.sid);
@@ -158,2 +172,9 @@ if (!socket) {
             case RequestType.REMOTE_LEAVE:
+                if (request.opts) {
+                    const opts = {
+                        rooms: new Set(request.opts.rooms),
+                        except: new Set(request.opts.except),
+                    };
+                    return super.delSockets(opts, request.rooms);
+                }
                 socket = this.nsp.sockets.get(request.sid);
@@ -170,2 +191,9 @@ if (!socket) {
             case RequestType.REMOTE_DISCONNECT:
+                if (request.opts) {
+                    const opts = {
+                        rooms: new Set(request.opts.rooms),
+                        except: new Set(request.opts.except),
+                    };
+                    return super.disconnectSockets(opts, request.close);
+                }
                 socket = this.nsp.sockets.get(request.sid);
@@ -181,2 +209,22 @@ if (!socket) {
                 break;
+            case RequestType.REMOTE_FETCH:
+                if (this.requests.has(request.requestId)) {
+                    return;
+                }
+                const opts = {
+                    rooms: new Set(request.opts.rooms),
+                    except: new Set(request.opts.except),
+                };
+                const localSockets = await super.fetchSockets(opts);
+                response = JSON.stringify({
+                    requestId: request.requestId,
+                    sockets: localSockets.map((socket) => ({
+                        id: socket.id,
+                        handshake: socket.handshake,
+                        rooms: [...socket.rooms],
+                        data: socket.data,
+                    })),
+                });
+                this.pubClient.publish(this.responseChannel, response);
+                break;
             default:
@@ -209,2 +257,3 @@ debug("ignoring unknown request type: %s", request.type);
             case RequestType.SOCKETS:
+            case RequestType.REMOTE_FETCH:
                 request.msgCount++;
@@ -214,3 +263,8 @@ // ignore if response does not contain 'sockets' key
                     return;
-                response.sockets.forEach((s) => request.sockets.add(s));
+                if (request.type === RequestType.SOCKETS) {
+                    response.sockets.forEach((s) => request.sockets.add(s));
+                }
+                else {
+                    response.sockets.forEach((s) => request.sockets.push(s));
+                }
                 if (request.msgCount === request.numSub) {
@@ -284,5 +338,9 @@ clearTimeout(request.timeout);
     async sockets(rooms) {
-        const requestId = uid2(6);
+        const localSockets = await super.sockets(rooms);
         const numSub = await this.getNumSub();
         debug('waiting for %d responses to "sockets" request', numSub);
+        if (numSub <= 1) {
+            return Promise.resolve(localSockets);
+        }
+        const requestId = uid2(6);
         const request = JSON.stringify({
@@ -305,4 +363,4 @@ requestId,
                 timeout,
-                msgCount: 0,
-                sockets: new Set(),
+                msgCount: 1,
+                sockets: localSockets,
             });
@@ -318,5 +376,9 @@ this.pubClient.publish(this.requestChannel, request);
     async allRooms() {
-        const requestId = uid2(6);
+        const localRooms = new Set(this.rooms.keys());
         const numSub = await this.getNumSub();
         debug('waiting for %d responses to "allRooms" request', numSub);
+        if (numSub <= 1) {
+            return localRooms;
+        }
+        const requestId = uid2(6);
         const request = JSON.stringify({
@@ -338,4 +400,4 @@ requestId,
                 timeout,
-                msgCount: 0,
-                rooms: new Set(),
+                msgCount: 1,
+                rooms: localRooms,
             });
@@ -450,2 +512,85 @@ this.pubClient.publish(this.requestChannel, request);
     }
+    async fetchSockets(opts) {
+        var _a;
+        const localSockets = await super.fetchSockets(opts);
+        if ((_a = opts.flags) === null || _a === void 0 ? void 0 : _a.local) {
+            return localSockets;
+        }
+        const numSub = await this.getNumSub();
+        debug('waiting for %d responses to "fetchSockets" request', numSub);
+        if (numSub <= 1) {
+            return localSockets;
+        }
+        const requestId = uid2(6);
+        const request = JSON.stringify({
+            requestId,
+            type: RequestType.REMOTE_FETCH,
+            opts: {
+                rooms: [...opts.rooms],
+                except: [...opts.except],
+            },
+        });
+        return new Promise((resolve, reject) => {
+            const timeout = setTimeout(() => {
+                if (this.requests.has(requestId)) {
+                    reject(new Error("timeout reached while waiting for fetchSockets response"));
+                    this.requests.delete(requestId);
+                }
+            }, this.requestsTimeout);
+            this.requests.set(requestId, {
+                type: RequestType.REMOTE_FETCH,
+                numSub,
+                resolve,
+                timeout,
+                msgCount: 1,
+                sockets: localSockets,
+            });
+            this.pubClient.publish(this.requestChannel, request);
+        });
+    }
+    addSockets(opts, rooms) {
+        var _a;
+        if ((_a = opts.flags) === null || _a === void 0 ? void 0 : _a.local) {
+            return super.addSockets(opts, rooms);
+        }
+        const request = JSON.stringify({
+            type: RequestType.REMOTE_JOIN,
+            opts: {
+                rooms: [...opts.rooms],
+                except: [...opts.except],
+            },
+            rooms: [...rooms],
+        });
+        this.pubClient.publish(this.requestChannel, request);
+    }
+    delSockets(opts, rooms) {
+        var _a;
+        if ((_a = opts.flags) === null || _a === void 0 ? void 0 : _a.local) {
+            return super.delSockets(opts, rooms);
+        }
+        const request = JSON.stringify({
+            type: RequestType.REMOTE_LEAVE,
+            opts: {
+                rooms: [...opts.rooms],
+                except: [...opts.except],
+            },
+            rooms: [...rooms],
+        });
+        this.pubClient.publish(this.requestChannel, request);
+    }
+    disconnectSockets(opts, close) {
+        var _a;
+        if ((_a = opts.flags) === null || _a === void 0 ? void 0 : _a.local) {
+            return super.disconnectSockets(opts, close);
+        }
+        const request = JSON.stringify({
+            type: RequestType.REMOTE_DISCONNECT,
+            opts: {
+                rooms: [...opts.rooms],
+                except: [...opts.except],
+            },
+            close,
+        });
+        this.pubClient.publish(this.requestChannel, request);
+    }
     /**
@@ -452,0 +597,0 @@ * Get the number of subscribers of the request channel

package.json

 {
   "name": "socket.io-redis",
-  "version": "6.0.1",
+  "version": "6.1.0",
   "description": "",
@@ -16,15 +16,17 @@ "license": "MIT",
   "scripts": {
-    "test": "npm run format:check && tsc && nyc mocha test/index.js",
-    "format:check": "prettier --parser typescript --check 'lib/**/*.ts' 'test/**/*.js'",
-    "format:fix": "prettier --parser typescript --write 'lib/**/*.ts' 'test/**/*.js'",
+    "test": "npm run format:check && tsc && nyc mocha --require ts-node/register test/index.ts",
+    "format:check": "prettier --parser typescript --check 'lib/**/*.ts' 'test/**/*.ts'",
+    "format:fix": "prettier --parser typescript --write 'lib/**/*.ts' 'test/**/*.ts'",
     "prepack": "tsc"
   },
   "dependencies": {
-    "debug": "~4.1.0",
+    "debug": "~4.3.1",
     "notepack.io": "~2.2.0",
     "redis": "^3.0.0",
-    "socket.io-adapter": "~2.0.0",
+    "socket.io-adapter": "^2.2.0",
     "uid2": "0.0.3"
   },
   "devDependencies": {
+    "@types/expect.js": "^0.3.29",
+    "@types/mocha": "^8.2.1",
     "@types/node": "^14.14.7",
@@ -36,4 +38,5 @@ "expect.js": "0.3.1",
     "prettier": "^2.1.2",
-    "socket.io": "^3.0.1",
-    "socket.io-client": "^3.0.1",
+    "socket.io": "^4.0.0",
+    "socket.io-client": "^4.0.0",
+    "ts-node": "^9.1.1",
     "typescript": "^4.0.5"
@@ -40,0 +43,0 @@ },

README.md

 # socket.io-redis
 
-[![Build Status](https://travis-ci.org/socketio/socket.io-redis.svg?branch=master)](https://travis-ci.org/socketio/socket.io-redis)
+[![Build Status](https://github.com/socketio/socket.io-redis/workflows/CI/badge.svg?branch=master)](https://github.com/socketio/socket.io-redis/actions)
 [![NPM version](https://badge.fury.io/js/socket.io-redis.svg)](http://badge.fury.io/js/socket.io-redis)
@@ -13,2 +13,3 @@
 - [Compatibility table](#compatibility-table)
+- [How does it work under the hood?](#how-does-it-work-under-the-hood)
 - [API](#api)
@@ -82,3 +83,3 @@ - [adapter(uri[, opts])](#adapteruri-opts)
 
-will properly be broadcast to the clients through the Redis Pub/Sub mechanism.
+will properly be broadcast to the clients through the Redis [Pub/Sub mechanism](https://redis.io/topics/pubsub).
 
@@ -94,4 +95,60 @@ If you need to emit events to socket.io instances from a non-socket.io
 | 5.x                   | 2.x                      |
-| 6.x                   | 3.x                      |
+| 6.0.x                 | 3.x                      |
+| 6.1.x and above       | 4.x                      |
 
+## How does it work under the hood?
+
+This adapter extends the [in-memory adapter](https://github.com/socketio/socket.io-adapter/) that is included by default with the Socket.IO server.
+
+The in-memory adapter stores the relationships between Sockets and Rooms in two Maps.
+
+When you run `socket.join("room21")`, here's what happens:
+
+```
+console.log(adapter.rooms); // Map { "room21" => Set { "mdpk4kxF5CmhwfCdAHD8" } }
+console.log(adapter.sids); // Map { "mdpk4kxF5CmhwfCdAHD8" => Set { "mdpk4kxF5CmhwfCdAHD8", room21" } }
+// "mdpk4kxF5CmhwfCdAHD8" being the ID of the given socket
+```
+
+Those two Maps are then used when broadcasting:
+
+- a broadcast to all sockets (`io.emit()`) loops through the `sids` Map, and send the packet to all sockets
+- a broadcast to a given room (`io.to("room21").emit()`) loops through the Set in the `rooms` Map, and sends the packet to all matching sockets
+
+The Redis adapter extends the broadcast function of the in-memory adapter: the packet is also [published](https://redis.io/topics/pubsub) to a Redis channel (see [below](#protocol) for the format of the channel name).
+
+Each Socket.IO server receives this packet and broadcasts it to its own list of connected sockets.
+
+To check what's happening on your Redis instance:
+
+```
+$ redis-cli
+127.0.0.1:6379> PSUBSCRIBE *
+Reading messages... (press Ctrl-C to quit)
+1) "psubscribe"
+2) "*"
+3) (integer) 1
+
+1) "pmessage"
+2) "*"
+3) "socket.io#/#" (a broadcast to all sockets or to a list of rooms)
+4) <the packet content>
+
+1) "pmessage"
+2) "*"
+3) "socket.io#/#room21#" (a broadcast to a single room)
+4) <the packet content>
+```
+
+Note: **no data** is stored in Redis itself
+
+There are 3 Redis subscriptions per namespace:
+
+- main channel: `<prefix>#<namespace>#*` (glob)
+- request channel: `<prefix>-request#<namespace>#`
+- response channel: `<prefix>-response#<namespace>#`
+
+The request and response channels are used in the additional methods exposed by the Redis adapter, like [RedisAdapter#allRooms()](#redisadapterallrooms).
+
+
 ## API
@@ -142,3 +199,3 @@
 // this method is also exposed by the Server instance
-const sockets = io.in('room3').allSockets();
+const sockets = await io.in('room3').allSockets();
 console.log(sockets); // a Set containing the socket ids in 'room3'
@@ -145,0 +202,0 @@ ```

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

41414

increased by25.62%

Lines of code

736

increased by25.38%

Number of lines in readme file

353

increased by19.26%

Worsened metrics

Dev dependency count

12

increased by33.33%

Dependency changes

• + Addeddebug@4.3.7(transitive)

• + Addedsocket.io-adapter@2.5.5(transitive)

• + Addedws@8.17.1(transitive)

• - Removeddebug@4.1.1(transitive)

• - Removedsocket.io-adapter@2.0.3(transitive)

• Updateddebug@~4.3.1

• Updatedsocket.io-adapter@^2.2.0