redis/node-redis
1
0
Fork 0
mirror of https://github.com/redis/node-redis.git synced 2025-08-20 18:01:21 +03:00
Code Activity

fix PubSub UNSUBSCRIBE/PUNSUBSCRIBE without channel and/or listener

Browse Source
This commit is contained in:
leibale
2021-06-11 17:49:22 -04:00
parent c29c1bb7d2
commit 463fad11e8
1 changed files with 22 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
lib/commands-queue.ts
View File

@@ -207,9 +207,8 @@ export default class RedisCommandsQueue {
unsubscribe(command: PubSubUnsubscribeCommands, channels?: string | Array<string>, listener?: PubSubListener) { unsubscribe(command: PubSubUnsubscribeCommands, channels?: string | Array<string>, listener?: PubSubListener) {
const listeners = command === PubSubUnsubscribeCommands.UNSUBSCRIBE ? this.#pubSubListeners.channels : this.#pubSubListeners.patterns; const listeners = command === PubSubUnsubscribeCommands.UNSUBSCRIBE ? this.#pubSubListeners.channels : this.#pubSubListeners.patterns;
if (!channels) { if (!channels) {
const keys = [...listeners.keys()];
listeners.clear(); listeners.clear();
return this.#pushPubSubCommand(command, keys); return this.#pushPubSubCommand(command);
} }
const channelsToUnsubscribe = []; const channelsToUnsubscribe = [];
@@ -232,24 +231,36 @@ export default class RedisCommandsQueue {
return this.#pushPubSubCommand(command, channelsToUnsubscribe); return this.#pushPubSubCommand(command, channelsToUnsubscribe);
} }
#pushPubSubCommand(command: PubSubSubscribeCommands | PubSubUnsubscribeCommands, channels: Array<string>): Promise<void> { #pushPubSubCommand(command: PubSubSubscribeCommands | PubSubUnsubscribeCommands, channels?: Array<string>): Promise<void> {
if (!channels.length) return Promise.resolve(); if (channels?.length === 0) return Promise.resolve();
return new Promise((resolve, reject) => { return new Promise((resolve, reject) => {
const isSubscribe = command === PubSubSubscribeCommands.SUBSCRIBE || command === PubSubSubscribeCommands.PSUBSCRIBE, const isSubscribe = command === PubSubSubscribeCommands.SUBSCRIBE || command === PubSubSubscribeCommands.PSUBSCRIBE,
inProgressKey = isSubscribe ? 'subscribing' : 'unsubscribing'; inProgressKey = isSubscribe ? 'subscribing' : 'unsubscribing',
commandArgs: Array<string> = [command];
let channelsCounter: number;
if (channels) {
commandArgs.push(...channels);
channelsCounter = channels.length;
} else {
channelsCounter = (
command[0] === 'P' ?
this.#pubSubListeners.patterns :
this.#pubSubListeners.channels
).size;
}
this.#pubSubState[inProgressKey] += channels.length; this.#pubSubState[inProgressKey] += channelsCounter;
this.#waitingToBeSent.push({ this.#waitingToBeSent.push({
encodedCommand: RedisCommandsQueue.encodeCommand([command, ...channels]), encodedCommand: RedisCommandsQueue.encodeCommand(commandArgs),
channelsCounter: channels.length, channelsCounter,
resolve: () => { resolve: () => {
this.#pubSubState[inProgressKey] -= channels.length; this.#pubSubState[inProgressKey] -= channelsCounter;
this.#pubSubState.subscribed += channels.length * (isSubscribe ? 1 : -1); this.#pubSubState.subscribed += channelsCounter * (isSubscribe ? 1 : -1);
resolve(); resolve();
}, },
reject: () => { reject: () => {
this.#pubSubState[inProgressKey] -= channels.length; this.#pubSubState[inProgressKey] -= channelsCounter;
reject(); reject();
} }
}); });