redis/node-redis
1
0
Fork 0
mirror of https://github.com/redis/node-redis.git synced 2025-08-07 13:22:56 +03:00
Code Activity

fix #1764 - fix PubSub resubscribe

Browse Source
This commit is contained in:
leibale
2021-12-06 20:29:59 -05:00
parent 7202f3582f
commit 82920aef0b
2 changed files with 119 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
packages/client/lib/client/commands-queue.ts
View File

@@ -294,9 +294,9 @@ export default class RedisCommandsQueue {
} }
resolve(); resolve();
}, },
reject: () => { reject: err => {
pubSubState[inProgressKey] -= channelsCounter * (isSubscribe ? 1 : -1); pubSubState[inProgressKey] -= channelsCounter * (isSubscribe ? 1 : -1);
reject(); reject(err);
} }
}); });
}); });
@@ -307,11 +307,32 @@ export default class RedisCommandsQueue {
return; return;
} }
// TODO: acl error on one channel/pattern will reject the whole command this.#pubSubState.subscribed = 0;
return Promise.all([
this.#pushPubSubCommand(PubSubSubscribeCommands.SUBSCRIBE, [...this.#pubSubState.listeners.channels.keys()]), const promises = [],
this.#pushPubSubCommand(PubSubSubscribeCommands.PSUBSCRIBE, [...this.#pubSubState.listeners.patterns.keys()]) { channels, patterns } = this.#pubSubState.listeners;
]);
if (channels.size) {
promises.push(
this.#pushPubSubCommand(
PubSubSubscribeCommands.SUBSCRIBE,
[...channels.keys()]
)
);
}
if (patterns.size) {
promises.push(
this.#pushPubSubCommand(
PubSubSubscribeCommands.PSUBSCRIBE,
[...patterns.keys()]
)
);
}
if (promises.length) {
return Promise.all(promises);
}
} }
getCommandToSend(): RedisCommandArguments | undefined { getCommandToSend(): RedisCommandArguments | undefined {

151
packages/client/lib/client/index.spec.ts
View File

@@ -560,73 +560,104 @@ describe('Client', () => {
); );
}, GLOBAL.SERVERS.OPEN); }, GLOBAL.SERVERS.OPEN);
testUtils.testWithClient('PubSub', async publisher => { describe('PubSub', () => {
function assertStringListener(message: string, channel: string) { testUtils.testWithClient('should be able to publish and subscribe to messages', async publisher => {
assert.ok(typeof message === 'string'); function assertStringListener(message: string, channel: string) {
assert.ok(typeof channel === 'string'); assert.ok(typeof message === 'string');
} assert.ok(typeof channel === 'string');
}
function assertBufferListener(message: Buffer, channel: Buffer) { function assertBufferListener(message: Buffer, channel: Buffer) {
assert.ok(Buffer.isBuffer(message)); assert.ok(Buffer.isBuffer(message));
assert.ok(Buffer.isBuffer(channel)); assert.ok(Buffer.isBuffer(channel));
} }
const subscriber = publisher.duplicate(); const subscriber = publisher.duplicate();
await subscriber.connect(); await subscriber.connect();
try { try {
const channelListener1 = spy(assertBufferListener), const channelListener1 = spy(assertBufferListener),
channelListener2 = spy(assertStringListener), channelListener2 = spy(assertStringListener),
patternListener = spy(assertStringListener); patternListener = spy(assertStringListener);
await Promise.all([ await Promise.all([
subscriber.subscribe('channel', channelListener1, true), subscriber.subscribe('channel', channelListener1, true),
subscriber.subscribe('channel', channelListener2), subscriber.subscribe('channel', channelListener2),
subscriber.pSubscribe('channel*', patternListener) subscriber.pSubscribe('channel*', patternListener)
]); ]);
await Promise.all([ await Promise.all([
waitTillBeenCalled(channelListener1), waitTillBeenCalled(channelListener1),
waitTillBeenCalled(channelListener2), waitTillBeenCalled(channelListener2),
waitTillBeenCalled(patternListener), waitTillBeenCalled(patternListener),
publisher.publish(Buffer.from('channel'), Buffer.from('message')) publisher.publish(Buffer.from('channel'), Buffer.from('message'))
]); ]);
assert.ok(channelListener1.calledOnceWithExactly(Buffer.from('message'), Buffer.from('channel'))); assert.ok(channelListener1.calledOnceWithExactly(Buffer.from('message'), Buffer.from('channel')));
assert.ok(channelListener2.calledOnceWithExactly('message', 'channel')); assert.ok(channelListener2.calledOnceWithExactly('message', 'channel'));
assert.ok(patternListener.calledOnceWithExactly('message', 'channel')); assert.ok(patternListener.calledOnceWithExactly('message', 'channel'));
await subscriber.unsubscribe('channel', channelListener1, true); await subscriber.unsubscribe('channel', channelListener1, true);
await Promise.all([ await Promise.all([
waitTillBeenCalled(channelListener2), waitTillBeenCalled(channelListener2),
waitTillBeenCalled(patternListener), waitTillBeenCalled(patternListener),
publisher.publish('channel', 'message') publisher.publish('channel', 'message')
]); ]);
assert.ok(channelListener1.calledOnce); assert.ok(channelListener1.calledOnce);
assert.ok(channelListener2.calledTwice); assert.ok(channelListener2.calledTwice);
assert.ok(channelListener2.secondCall.calledWithExactly('message', 'channel')); assert.ok(channelListener2.secondCall.calledWithExactly('message', 'channel'));
assert.ok(patternListener.calledTwice); assert.ok(patternListener.calledTwice);
assert.ok(patternListener.secondCall.calledWithExactly('message', 'channel')); assert.ok(patternListener.secondCall.calledWithExactly('message', 'channel'));
await subscriber.unsubscribe('channel'); await subscriber.unsubscribe('channel');
await Promise.all([ await Promise.all([
waitTillBeenCalled(patternListener), waitTillBeenCalled(patternListener),
publisher.publish('channel', 'message') publisher.publish('channel', 'message')
]); ]);
assert.ok(channelListener1.calledOnce); assert.ok(channelListener1.calledOnce);
assert.ok(channelListener2.calledTwice); assert.ok(channelListener2.calledTwice);
assert.ok(patternListener.calledThrice); assert.ok(patternListener.calledThrice);
assert.ok(patternListener.thirdCall.calledWithExactly('message', 'channel')); assert.ok(patternListener.thirdCall.calledWithExactly('message', 'channel'));
await subscriber.pUnsubscribe(); await subscriber.pUnsubscribe();
await publisher.publish('channel', 'message'); await publisher.publish('channel', 'message');
assert.ok(channelListener1.calledOnce); assert.ok(channelListener1.calledOnce);
assert.ok(channelListener2.calledTwice); assert.ok(channelListener2.calledTwice);
assert.ok(patternListener.calledThrice); assert.ok(patternListener.calledThrice);
// should be able to send commands when unsubsribed from all channels (see #1652) // should be able to send commands when unsubsribed from all channels (see #1652)
await assert.doesNotReject(subscriber.ping()); await assert.doesNotReject(subscriber.ping());
} finally { } finally {
await subscriber.disconnect(); await subscriber.disconnect();
} }
}, GLOBAL.SERVERS.OPEN); }, GLOBAL.SERVERS.OPEN);
testUtils.testWithClient('should resubscribe', async publisher => {
const subscriber = publisher.duplicate();
await subscriber.connect();
try {
const listener = spy();
await subscriber.subscribe('channel', listener);
subscriber.on('error', err => {
console.error('subscriber err', err.message);
});
await Promise.all([
once(subscriber, 'error'),
publisher.sendCommand(['CLIENT', 'KILL', 'SKIPME', 'yes'])
]);
await once(subscriber, 'ready');
await Promise.all([
waitTillBeenCalled(listener),
publisher.publish('channel', 'message')
]);
} finally {
await subscriber.disconnect();
}
}, GLOBAL.SERVERS.OPEN);
});
testUtils.testWithClient('ConnectionTimeoutError', async client => { testUtils.testWithClient('ConnectionTimeoutError', async client => {
const promise = assert.rejects(client.connect(), ConnectionTimeoutError), const promise = assert.rejects(client.connect(), ConnectionTimeoutError),