diff options
| -rw-r--r-- | streaming/index.js | 12 |
1 files changed, 9 insertions, 3 deletions
diff --git a/streaming/index.js b/streaming/index.js index 270ed6f70..5afdd5961 100644 --- a/streaming/index.js +++ b/streaming/index.js @@ -115,7 +115,7 @@ const startWorker = (workerId) => { const subs = {}; - redisSubscribeClient.on('pmessage', (_, channel, message) => { + redisSubscribeClient.on('message', (channel, message) => { const callbacks = subs[channel]; log.silly(`New message on channel ${channel}`); @@ -127,8 +127,6 @@ const startWorker = (workerId) => { callbacks.forEach(callback => callback(message)); }); - redisSubscribeClient.psubscribe(`${redisPrefix}timeline:*`); - const subscriptionHeartbeat = (channel) => { const interval = 6*60; const tellSubscribed = () => { @@ -144,12 +142,20 @@ const startWorker = (workerId) => { 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); + } subs[channel].push(callback); }; const unsubscribe = (channel, callback) => { log.silly(`Removing listener for ${channel}`); subs[channel] = subs[channel].filter(item => item !== callback); + if (subs[channel].length === 0) { + log.verbose(`Unsubscribe ${channel}`); + redisSubscribeClient.unsubscribe(channel); + } }; const allowCrossDomain = (req, res, next) => { |