Monkey Cymbal

Production ready, MongoDB-based queue for Node.

---

Features

• Delayed jobs
• Retries
• Dead queue
• Priority
• Concurrency
• Pause/resume processing
• Optional Topic based publishing (publish into multiple queues)
• Low CPU usage
• 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.

Name	Type	Description
connectionUrlOrMongoClient required	`MongoClient	string`
queueName required	string	The name of the queue
options	SubscriptionOptions

SubscriptionOptions

Arguments	Type	Default	Description
visibility	number (seconds)	10	After a message is received to prevent other consumers from processing the message again, Monkeycymbal sets a visibility timeout, a period of time during which Monkeycymbal prevents other consumers from receiving and processing the message.
delay	number (seconds)		if you set delay to be 10, then every message will only be available for retrieval 10s after being added.
maxRetries	number	5	Maximum number of attempts to retry processing a message. If deadQueue is set, the message will be moved to the dead queue. Otherwise it will be acked.
expireAfterSeconds	number (seconds)		The processed messages will be removed from the collection after the specified number of seconds.
concurrency	number	1	The max number of messages that will be processed in parallel.
pollInterval	number (seconds)	10	The amount of time the subscriber waits before checking for new messages.
deadQueue	string or Queue instance		Messages that have been retried over maxRetries will be pushed to this queue for later inspection.

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

Arguments	Type	Default	Description
priority	number	1	Optional priority value. It ranges from -Infinity to +Infinity

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.

Name	Type	Description
connectionUrlOrMongoClient required	`MongoClient	string`
topic required	topic	The name of the channel.
options	SubscriptionOptions

Channel.publish

channel.publish(msg, PublishOptions);

Publish messages to the queues subscribed to the topic.

PublishOptions

Arguments	Type	Default	Description
priority	number	1	Optional priority value. It ranges from -Infinity to +Infinity

Channel.subscribe

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

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