shape-of-q

shape-of-q

A simple and fast redis based queue, supports both fifo and lifo.

Install

npm i shape-of-q

Usage

const q = require('shape-of-q')('myqueue')
q.on('error', console.log)

// push a new message
q.push('hello')

q.pull({ polling: true }, (msg, done) => {
  console.log(msg)
  done()
})

Async await is supported as well!

q.pull({ polling: true }, async msg => {
  console.log(msg)
})

API

Constructor

Create a new queue, the queue name parameter is mandatory.

Options:

const q = require('shape-of-q')('myqueue', {
  encoding: 'json', // default: null
  type: 'lifo', // default: 'fifo'
  client: Object, // custom redis client
  host: '127.0.0.1' // redis host for the internal client,
  encoder: msgpack.encode, // default null
  decoder: msgpack.decode, // default null
  binaryData: true // default false
})

shape-of-q is an event emitter and you should listen for the error event.

If you are working with objects and want to speed up the serialization you can use the encoder and decoder option, both of them must be sync functions.
If the encoder produces binary data make sure to pass { binaryData: true } as option.

push

Add a new message to the queue.

If the encoding has been set to 'json' you can pass plain objects.

q.push('hello')

pull

Retrieve a single message from the queue.

To enable polling, pass { polling: true } as option and pollingInterval if you want to customize the interval (must be expressed in seconds).
The api supports both classic callbacks and async await.

// callbacks
q.pull({ polling: true }, (msg, done) => {
  console.log(msg)
  done()
})

// async-await
q.pull({ polling: true }, async msg => {
  console.log(msg)
})

If you pass an error to done or return an error in the async version the message will be put back in the queue.

list

Get all elements in the queue.

q.list((err, msg) => {
  console.log(msg)
})

stop

Stops the polling and closes the connection to redis.

q.stop()

flush

Flushes a queue removing all its elements.

q.flush((err) => {
  if (err) console.log(err)
})

License

MIT

Copyright © 2018 Tomas Della Vedova