XediDC XediDC - 1 month ago 11
Node.js Question

Better way to write a simple Node redis loop (using ioredis)?

So, I'm stilling learning the JS/Node way from a long time in other languages.

I have a tiny micro-service that reads from a redis channel, temp stores it in a working channel, does the work, removes it, and moves on. If there is more in the channel it re-runs immediately. If not, it sets a timeout and checks again in 1 second.

It works fine...but timeout polling doesn't seem to be the "correct" way to approach this. And I haven't found much about using BRPOPLPUSH to try to block (vs. RPOPLPUSH) and wait in Node....or other options like that. (Pub/Sub isn't an option here...this is the only listener, and it may not always be listening.)

Here's the short essence of what I'm doing:

var Redis = require('ioredis');
var redis = new Redis();

var redisLoop = function () {
redis.rpoplpush('channel', 'channel-working').then(function (result) {
if (result) {
processJob(result); //do stuff

//delete the item from the working channel, and check for another item
redis.lrem('channel-working', 1, result).then(function (result) { });
redisLoop();
} else {
//no items, wait 1 second and try again
setTimeout(redisLoop, 1000);
}
});
};

redisLoop();


I feel like I'm missing something really obvious. Thanks!

Answer

BRPOPLPUSH doesn't block in Node, it blocks in the client. In this instance I think it's exactly what you need to get rid of the polling.

var Redis = require('ioredis');
var redis = new Redis();

var redisLoop = function () {
    redis.brpoplpush('channel', 'channel-working', 0).then(function (result) {
        // because we are using BRPOPLPUSH, the client promise will not resolve
        // until a 'result' becomes available
        processJob(result);

        // delete the item from the working channel, and check for another item
        redis.lrem('channel-working', 1, result).then(redisLoop);
    });
};

redisLoop();

Note that redis.lrem is asynchronous, so you should use lrem(...).then(redisLoop) to ensure that your next tick executes only after the item is successfully removed from channel-working.