Skip to content

Commit 2113e8d

Browse files
authoredMar 13, 2024··
fix(sharded): fix SSUBSCRIBE memory leak with ioredis (#529)
This pull request introduces a change to the sharded adapter's `SSUBSCRIBE` logic: Previously, for each dynamic channel/room, a unique listener was added to the client's `smessageBuffer`. This approach led to a large number of listeners (resulting in `MaxListenersExceededWarning`), especially in scenarios with many dynamic channels. Further, listeners were not being removed when unsubscribing, leading to a memory leak. The new implementation replaces the multiple listeners with a single `smessageBuffer` listener. This listener is registered once and handles all dynamic channels by maintaining specific channel handlers in a `Map`. Listeners are added to this `Map` in `SSUBSCRIBE` and removed from the `Map` in `SUNSUBSCRIBE`. Related: #528
1 parent bd32763 commit 2113e8d

File tree

1 file changed

+17
-6
lines changed

1 file changed

+17
-6
lines changed
 

‎lib/util.ts

+17-6
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,8 @@ function isRedisV4Client(redisClient: any) {
5858
return typeof redisClient.sSubscribe === "function";
5959
}
6060

61+
const kHandlers = Symbol("handlers");
62+
6163
export function SSUBSCRIBE(
6264
redisClient: any,
6365
channel: string,
@@ -66,13 +68,17 @@ export function SSUBSCRIBE(
6668
if (isRedisV4Client(redisClient)) {
6769
redisClient.sSubscribe(channel, handler, RETURN_BUFFERS);
6870
} else {
71+
if (!redisClient[kHandlers]) {
72+
redisClient[kHandlers] = new Map();
73+
redisClient.on("smessageBuffer", (rawChannel, message) => {
74+
redisClient[kHandlers].get(rawChannel.toString())?.(
75+
message,
76+
rawChannel
77+
);
78+
});
79+
}
80+
redisClient[kHandlers].set(channel, handler);
6981
redisClient.ssubscribe(channel);
70-
71-
redisClient.on("smessageBuffer", (rawChannel, message) => {
72-
if (rawChannel.toString() === channel) {
73-
handler(message, rawChannel);
74-
}
75-
});
7682
}
7783
}
7884

@@ -81,6 +87,11 @@ export function SUNSUBSCRIBE(redisClient: any, channel: string | string[]) {
8187
redisClient.sUnsubscribe(channel);
8288
} else {
8389
redisClient.sunsubscribe(channel);
90+
if (Array.isArray(channel)) {
91+
channel.forEach((c) => redisClient[kHandlers].delete(c));
92+
} else {
93+
redisClient[kHandlers].delete(channel);
94+
}
8495
}
8596
}
8697

0 commit comments

Comments
 (0)
Please sign in to comment.