Fix streaming server sometimes silently dropping subscriptions (#17841)

This commit is contained in:
Claire 2022-03-21 19:08:29 +01:00 committed by GitHub
parent 7eb2e791ee
commit f29458da1d
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23

View file

@ -167,6 +167,11 @@ const startWorker = async (workerId) => {
const redisPrefix = redisNamespace ? `${redisNamespace}:` : ''; const redisPrefix = redisNamespace ? `${redisNamespace}:` : '';
/**
* @type {Object.<string, Array.<function(string): void>>}
*/
const subs = {};
const redisSubscribeClient = await redisUrlToClient(redisParams, process.env.REDIS_URL); const redisSubscribeClient = await redisUrlToClient(redisParams, process.env.REDIS_URL);
const redisClient = await redisUrlToClient(redisParams, process.env.REDIS_URL); const redisClient = await redisUrlToClient(redisParams, process.env.REDIS_URL);
@ -191,23 +196,55 @@ const startWorker = async (workerId) => {
}; };
/** /**
* @param {string} message
* @param {string} channel * @param {string} channel
* @param {function(string): void} callback
*/ */
const subscribe = (channel, callback) => { const onRedisMessage = (message, channel) => {
log.silly(`Adding listener for ${channel}`); const callbacks = subs[channel];
redisSubscribeClient.subscribe(channel, callback); log.silly(`New message on channel ${channel}`);
if (!callbacks) {
return;
}
callbacks.forEach(callback => callback(message));
}; };
/** /**
* @param {string} channel * @param {string} channel
* @param {function(string): void} callback * @param {function(string): void} callback
*/ */
const subscribe = (channel, callback) => {
log.silly(`Adding listener for ${channel}`);
subs[channel] = subs[channel] || [];
if (subs[channel].length === 0) {
log.verbose(`Subscribe ${channel}`);
redisSubscribeClient.subscribe(channel, onRedisMessage);
}
subs[channel].push(callback);
};
/**
* @param {string} channel
*/
const unsubscribe = (channel, callback) => { const unsubscribe = (channel, callback) => {
log.silly(`Removing listener for ${channel}`); log.silly(`Removing listener for ${channel}`);
redisSubscribeClient.unsubscribe(channel, callback); if (!subs[channel]) {
return;
}
subs[channel] = subs[channel].filter(item => item !== callback);
if (subs[channel].length === 0) {
log.verbose(`Unsubscribe ${channel}`);
redisSubscribeClient.unsubscribe(channel);
delete subs[channel];
}
}; };
const FALSE_VALUES = [ const FALSE_VALUES = [