Features

• [x] Delayed jobs
• [x] Retries
• [x] Dead queue
• [x] Priority
• [x] Concurrency
• [x] Pause/resume processing
• [x] Optional Topic based publishing (publish into multiple queues)
• [x] Low CPU usage
• [x] Able to process around 1000 messages per second (tested on Macbook Pro 13-inch, 2017, concurrency set to 150)

Install

npm install monkeycymbal --save

or

yarn add monkeycymbal

Requirements: Monkey Cymbal requires MongoDB

Quick Guide

Basic Usage

import { Queue } from 'monkeycymbal';

// Initialize queue
const queue = new Queue('mongodb://localhost/myDb', 'videoTranscoding');

// subscribe to the queue
queue.subscribe((msg) => {
  // process the message
  // ...
  
  // we can return a result that will be saved in the message
  
  return 'transcoded';
});

// Add a message to the queue
const [msgId] = await queue.add({ video: 'http://example.com/video1.mov' });

Pause / Resume

A queue can be paused and resumed globally (pass true to pause processing for just this worker):

await queue.pause()

// queue is paused now

await queue.resume()

// queue is resumed now

Events

A queue emits also some useful events, for example...

queue.on('added', msgId => {
  // A message has been added to the queue
})

queue.on('active', msg => {
  // The message is being processed
});

queue.on('completed', (msg, result) => {
  // The message has been processed succesfully
})

queue.on('error', (msg, error) => {
  // An error occurred while processing the message.
  // If maxRetries is set, it will be re-processed after a visibility timeout
})

queue.on('dead', msg => {
  // The message is failed permanently.
  // If a dead queue is configured, the message will be copied there.
})

For more information on events, including the full list of events that are fired, check out the Events reference

Documentation

• Queue

  • Queue.subscribe
  • Queue.add
  • Queue.pause
  • Queue.resume
  • Queue.ping

• Channel

  • Channel.publish
  • Channel.subscribe

Queue

new Queue(connectionUrlOrMongoClient, queueName, options)

This is the Queue constructor. It creates a new Queue that is persisted in MongoDB.

SubscriptionOptions

Queue.subscribe

queue.subscribe(handler);

Defines a processing function for the jobs in a given Queue and start processing. The handler function receive msg as argument.

Queue.add

queue.add(msg, AddMessageOptions);

Adds a message to the queue.

AddMessageOptions

Queue.pause

queue.pause();

Pause a queue. A paused queue will not process new jobs until resumed.

Queue.resume

queue.resume();

Resume a queue after being paused.

Queue.ping

queue.ping(msg.ack);

Ping a message to keep it's visibility open for long-running tasks.

Queue.totalCount

queue.totalCount();

Returns the total number of records in the collection.

Queue.waitingCount

queue.waitingCount();

Returns the total number of messages that are waiting to be processed.

Queue.inFlightCount

queue.inFlightCount();

Returns the total number of messages that are currently being processed.

Channel

new Channel(connectionUrlOrMongoClient, topic, options)

This is the Channel constructor. It creates a new Channel.

Channel.publish

channel.publish(msg, PublishOptions);

Publish messages to the queues subscribed to the topic.

PublishOptions

Channel.subscribe

channel.subscribe(handler, queueName, SubscriptionOptions): Queue;

Convenience method that returns an instance of a queue bound to the channel.