Push/Pull

Implement a Push/Pull mecanism using Redis as backend. This pattern is typically used to handle pool of workers.

Installation

npm install --save pushpull

If you don't already use Redis and still use npm ≤ 2.x, you'll have to install redis too:

npm install --save redis

You may want to install hiredis to, whenever possible:

npm install --save-optional hiredis

Usage

Look at (and run) sample.js for a working example.

// Worker
 
var Pull = require("./").Pull;
 
var worker = new Pull(options);
worker.on("data", function (data) {
  // do something with data
});
 
// Sender
 
var Push = require("./").Push;
 
var sender = new Push(options);
sender.emit("data", {"some": "data"});

Pull API

Pull instances are valid readable streams (object mode).

• new Pull(options): constructor, see Options below
• data event: emitted when data has been pulled from queue
• error event: emitter when an error occurs (seriously)
• pause(): stop querying for more data
  • IMPORTANT: this method cannot interrupt current fetch, which means one job can be pulled before puller is actually paused. This data will be buffered and emitted again as soon as you call resume().
• resume(): quit pause
• end(): close the underlying Redis client, which obviously prevents any further query

Push API

Push instances are valid writable streams (object mode).

• new Push(options): constructor, see Options below
• write(data): emit this event to push a job to queue
• pushed event: emitted when data has been pushed successfully
• error event: emitter when an error occurs (seriously)
• end(): close the underlying Redis client, which obviously prevents any further query

Options

• queue (mandatory): queue name, to be shared between worker(s) and sender(s)
  • Note: a Redis list with this name will be created. No decoration added, if you want to ensure unicity of the name, it's up to you to add prefix or suffix
• timeout (Pull only, default = 30 seconds): timeout between two BLPOP commands, the lower the value, the higher the chance not to grab data when calling pause()
• client: the Redis client to be used
  • IMPORTANT: a Pull instance will run BLPOP commands, which BLOCK the client. It's highly advised to use a unique client for each puller.
• host (default = localhost): Redis host (if client is not set)
• port (default = 6379): Redis port (if client is not set)
• database: Redis database to select once connected, if set

Internals

In both Push and Pull instances, you have access to some semi-privates:

• _redisClient is the internal Redis client instance
• _queue is the name of the queue
  • Note: changing its value will have expected effect, although you should only do it if you know what you're doing

Why documenting those internals? Because I use them in a project using this module, and I want to make sure those features don't get away with no warning.