Mq-mongo NPM | npm.io

mq-mongo

Easy message queue on mongo

Install

npm i mq-mongo

Usage

const mq = require('mq-mongo');
const q = mq(db, options);
q.get().then(msg => {/* Do something */});

db - mongo db object (or promise resolves to). Only required parameter.

options

options.name - name of mongo collection for queue. Defaults to 'mq'.
options.ttl - time-to-live (ms) for taken message before it will be acked or returned to queue. Defaults to 30000.
options.tries - just how many times single message may be taken and returned to queue without ack. Defaults to 10.
options.clean - if true previous messages in this queue (in fact documents in collection) will be deleted. Defaults to false.
options.strict - if true method get will work in strict mode (throws error instead of returning null). Defaults to false.
options.insistent - if true then get will begins from last failed (returned to queue without ack) messages. If false then get follow 'FIFO' rule.
options.items - message or array of messages for adding to queue on start.

methods (all asinc)

q.add(something) - adds single message or array of messages to queue. Returns array of _id strings for added messages.
q.get(ttl) - gets message from queue. Optional parameter is individual ttl for that specific message. Returns message object if there is messages ready. If no waiting messages returns null or (if options.strict set to true) throws error.
q.ack(tag) - deletes successfully handled message (specified by tag field) from queue. Returns _id string of deleted message or null (if no message with such tag or if ttl expires).
q.ping(tag, ttl) - prolong ttl of message specified by tag field. Optional parameter ttl defaults to options.ttl of queue.
q.waiting() - returns quantity of messages in queue.
q.active() - returns quantity of messages in work (waiting for ack).
q.failed() - returns quantity of failed messages (all tries is over).
q.total() - returns total quantity of messages (sum of three above).
q.stats() - returns object with quantities of waiting, active and failed messages.
q.options() - getter, returns object with some options ({ttl, tries, insistent}).

message fields

msg._id - mongo objectID of message.
msg.data - payload data of message.
msg.created - time (unix TS, number) when message was added to queue.
msg.expires - time (unix TS, number) when message will returns to queue.
msg.tries - just how many times this message was getted from queue.
msg.tag - unique tag for this try (for use in ack and ping).

Example

const mq = require('mq-mongo');

(async () => {
    const q = mq(db, {
        name = 'mq', // default mongo collection name
        ttl = 30000, // default message ttl
        tries = 10,  // default tries to handle message
    });
    await q.add('test');
    const msg = await q.get();
    await doSomethingWithData(msg.data);
    await q.ack(msg.tag);
})();

See tests for more complicated examples.