Skip to content

Commit

Permalink
Add disablePatternSubscriptions option
Browse files Browse the repository at this point in the history 
Turning on this option will disable the use of redis pattern subscriptions
for receiving messages. All messages will be received through regular
subscriptions.

Added new test scripts for running against all the redis clients with the
option enabled.
  • Loading branch information
@stevebaum23
stevebaum23 committed Apr 5, 2023
1 parent 2d65d0a commit 4097451
Show file tree
Hide file tree
Showing 3 changed files with 95 additions and 37 deletions.
116 changes: 82 additions & 34 deletions lib/index.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,20 @@ export interface RedisAdapterOptions {
* This option defaults to using `notepack.io`, a MessagePack implementation.
*/
parser: Parser;
/**
* Whether or not to disable pattern subscriptions. Pattern subscriptions are not optimized in redis and
* having a large number of them can cause redis latency. The amount of latency depends on the number of
* publishes and the number of pattern subscriptions.
*
* If you are using rooms and you enable this option, you will lose some CPU optimization when sending a
* message to a single room.
*
* You wouldn't turn this option on unless you notice higher amounts of redis latency and you have a large
* number of namespaces.
*
* @default false
*/
disablePatternSubscriptions: boolean;
}

/**
Expand All @@ -98,6 +112,7 @@ export class RedisAdapter extends Adapter {
public readonly requestsTimeout: number;
public readonly publishOnSpecificResponseChannel: boolean;
public readonly parser: Parser;
public readonly disablePatternSubscriptions: boolean;

private readonly channel: string;
private readonly requestChannel: string;
Expand Down Expand Up @@ -129,6 +144,7 @@ export class RedisAdapter extends Adapter {
this.requestsTimeout = opts.requestsTimeout || 5000;
this.publishOnSpecificResponseChannel = !!opts.publishOnSpecificResponseChannel;
this.parser = opts.parser || msgpack;
this.disablePatternSubscriptions = !!opts.disablePatternSubscriptions;

const prefix = opts.key || "socket.io";

Expand All @@ -137,6 +153,16 @@ export class RedisAdapter extends Adapter {
this.responseChannel = prefix + "-response#" + this.nsp.name + "#";
this.specificResponseChannel = this.responseChannel + this.uid + "#";

const subscribeChannels = [
this.requestChannel,
this.responseChannel,
this.specificResponseChannel,
];

if (this.disablePatternSubscriptions) {
subscribeChannels.push(this.channel);
}

const isRedisV4 = typeof this.pubClient.pSubscribe === "function";
if (isRedisV4) {
this.redisListeners.set("psub", (msg, channel) => {
Expand All @@ -147,35 +173,33 @@ export class RedisAdapter extends Adapter {
this.onrequest(channel, msg);
});

this.subClient.pSubscribe(
this.channel + "*",
this.redisListeners.get("psub"),
true
);
if (!this.disablePatternSubscriptions) {
this.subClient.pSubscribe(
this.channel + "*",
this.redisListeners.get("psub"),
true
);
}

this.subClient.subscribe(
[
this.requestChannel,
this.responseChannel,
this.specificResponseChannel,
],
subscribeChannels,
this.redisListeners.get("sub"),
true
);
} else {
this.redisListeners.set("pmessageBuffer", this.onmessage.bind(this));
this.redisListeners.set("messageBuffer", this.onrequest.bind(this));

this.subClient.psubscribe(this.channel + "*");
this.subClient.on(
"pmessageBuffer",
this.redisListeners.get("pmessageBuffer")
);
if (!this.disablePatternSubscriptions) {
this.subClient.psubscribe(this.channel + "*");

this.subClient.subscribe([
this.requestChannel,
this.responseChannel,
this.specificResponseChannel,
]);
this.subClient.on(
"pmessageBuffer",
this.redisListeners.get("pmessageBuffer")
);
}

this.subClient.subscribe(subscribeChannels);
this.subClient.on(
"messageBuffer",
this.redisListeners.get("messageBuffer")
Expand Down Expand Up @@ -246,6 +270,11 @@ export class RedisAdapter extends Adapter {

if (channel.startsWith(this.responseChannel)) {
return this.onresponse(channel, msg);
} else if (
this.disablePatternSubscriptions &&
channel.startsWith(this.channel)
) {
return this.onmessage(undefined, channel, msg);
} else if (!channel.startsWith(this.requestChannel)) {
return debug("ignore different channel");
}
Expand Down Expand Up @@ -629,7 +658,11 @@ export class RedisAdapter extends Adapter {
};
const msg = this.parser.encode([this.uid, packet, rawOpts]);
let channel = this.channel;
if (opts.rooms && opts.rooms.size === 1) {
if (
!this.disablePatternSubscriptions &&
opts.rooms &&
opts.rooms.size === 1
) {
channel += opts.rooms.keys().next().value + "#";
}
debug("publishing message to channel %s", channel);
Expand Down Expand Up @@ -939,12 +972,21 @@ export class RedisAdapter extends Adapter {

close(): Promise<void> | void {
const isRedisV4 = typeof this.pubClient.pSubscribe === "function";

if (isRedisV4) {
this.subClient.pUnsubscribe(
this.channel + "*",
this.redisListeners.get("psub"),
true
);
if (this.disablePatternSubscriptions) {
this.subClient.unsubscribe(
this.channel,
this.redisListeners.get("sub"),
true
);
} else {
this.subClient.pUnsubscribe(
this.channel + "*",
this.redisListeners.get("psub"),
true
);
}

// There is a bug in redis v4 when unsubscribing multiple channels at once, so we'll unsub one at a time.
// See https://github.com/redis/node-redis/issues/2052
Expand All @@ -964,17 +1006,23 @@ export class RedisAdapter extends Adapter {
true
);
} else {
this.subClient.punsubscribe(this.channel + "*");
this.subClient.off(
"pmessageBuffer",
this.redisListeners.get("pmessageBuffer")
);

this.subClient.unsubscribe([
const unsubscribeChannels = [
this.requestChannel,
this.responseChannel,
this.specificResponseChannel,
]);
];

if (this.disablePatternSubscriptions) {
unsubscribeChannels.push(this.channel);
} else {
this.subClient.punsubscribe(this.channel + "*");
this.subClient.off(
"pmessageBuffer",
this.redisListeners.get("pmessageBuffer")
);
}

this.subClient.unsubscribe(unsubscribeChannels);
this.subClient.off(
"messageBuffer",
this.redisListeners.get("messageBuffer")
Expand Down
5 changes: 4 additions & 1 deletion package.json
View file
Original file line number Diff line number Diff line change
Expand Up @@ -13,11 +13,14 @@
"main": "./dist/index.js",
"types": "./dist/index.d.ts",
"scripts": {
"test": "npm run format:check && tsc && npm run test:redis-v4 && npm run test:redis-v4-specific-channel && npm run test:redis-v3 && npm run test:ioredis",
"test": "npm run format:check && tsc && npm run test:redis-v4 && npm run test:redis-v4-specific-channel && npm run test:redis-v3 && npm run test:ioredis && npm run test:redis-v4-disable-pattern-subs && npm run test:redis-v3-disable-pattern-subs && npm run test:ioredis-disable-pattern-subs",
"test:redis-v4": "nyc mocha --bail --require ts-node/register test/index.ts",
"test:redis-v4-specific-channel": "SPECIFIC_CHANNEL=1 nyc mocha --bail --require ts-node/register test/index.ts",
"test:redis-v3": "REDIS_CLIENT=redis-v3 nyc mocha --bail --require ts-node/register test/index.ts",
"test:ioredis": "REDIS_CLIENT=ioredis nyc mocha --bail --require ts-node/register test/index.ts",
"test:redis-v4-disable-pattern-subs": "DISABLE_PATTERN_SUBS=1 nyc mocha --bail --require ts-node/register test/index.ts",
"test:redis-v3-disable-pattern-subs": "REDIS_CLIENT=redis-v3 DISABLE_PATTERN_SUBS=1 nyc mocha --bail --require ts-node/register test/index.ts",
"test:ioredis-disable-pattern-subs": "REDIS_CLIENT=ioredis DISABLE_PATTERN_SUBS=1 nyc mocha --bail --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"
Expand Down
11 changes: 9 additions & 2 deletions test/index.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -237,8 +237,13 @@ describe(`socket.io-redis with ${

// Depending on the version of redis this may be 3 (redis < v5) or 1 (redis > v4)
// Older versions subscribed multiple times on the same pattern. Newer versions only sub once.
expect(info.pubsub_patterns).to.be.greaterThan(0);
expect(info.pubsub_channels).to.eql(5); // 2 shared (request/response) + 3 unique for each namespace
if (process.env.DISABLE_PATTERN_SUBS) {
expect(info.pubsub_patterns).to.eql(0);
expect(info.pubsub_channels).to.eql(6); // 2 shared (request/response) + 4 unique for each namespace
} else {
expect(info.pubsub_patterns).to.be.greaterThan(0);
expect(info.pubsub_channels).to.eql(5); // 2 shared (request/response) + 3 unique for each namespace
}

namespace1.adapter.close();
namespace2.adapter.close();
Expand Down Expand Up @@ -529,6 +534,8 @@ function _create() {
createAdapter(await createClient(), await createClient(), {
publishOnSpecificResponseChannel:
process.env.SPECIFIC_CHANNEL !== undefined,
disablePatternSubscriptions:
process.env.DISABLE_PATTERN_SUBS !== undefined,
})
);
httpServer.listen((err) => {
Expand Down

0 comments on commit 4097451

Please sign in to comment.