Ready to take your JavaScript development to the next level? Meet npm Enterprise - the ultimate in enterprise JavaScript. Learn more »

wrappitmq

1.0.4 • Public • Published

wrappitmq – it wraps amqplib for us lazy people!

Build Status Coverage Status Dependencies Status

npm install wrappitmq

This is a Node.JS library for Inter-Process Communication with RabbitMQ based on amqplib (AMQP 0-9-1). It intends to take care of all the complicated stuff and offers you a simple API that just works out of the box and does what you expect it to (don't nail me down on this).

It makes heavy use of async/await and requires node v7.

WorkQueue

Producer enqueues a task to be worked on. You may block execution until the Broker confirms that it has received the message. That task will be picked up by one Worker and it will be acknowledged if the consuming function resolves. If an error occurs within the consuming function the error event will be emitted on the queue and the task will be re-enqueued for another try.

const WorkQueue = require('wrappitmq').WorkQueue;
 
// Set-up.
const queue = new WorkQueue({
  // All these options are optional.
  prefetch: 1, // How many messages to work on asynchronously (Default: 1)
  queue: 'workqueue', // Name of the queue to use. (Default: workqueue)
  url: 'amqp://localhost:5672' // Can also be specified on connect()
});
 
// URL here is optional, can also be set as above.
await queue.connect('amqp://localhost:5672');
 
queue.on('error', (err) => {
  // Something went wrong and you should log it.
});
queue.on('close', (err) => {
  // The connection was closed and the queue is no longer usable.
  // An err is given if the connection was closed due to an error.
});
 
// Enqueue a task.
await queue.enqueue({ do: 'it' });
 
// Consume tasks.
const cancel = await queue.consume(async (task) => {
  await work(task);
  // Will be acknowledged when work() is done.
});
 
// Cancel consumer to stop receiving tasks.
await cancel();
 
// Close connection.
await queue.close();

PubSub

Publisher publishes a message on a certain topic. You may block execution until the Broker confirms that it has received the message. The message will be broadcast to all Subscribers listening on that topic. There is no acknowledgement and persistence handling for these exchanges. Messages will be lost if nobody has subscribed to the topic.

const PubSub = require('wrappitmq').PubSub;
 
// Set-up.
const pubsub = new PubSub({
  // All these options are optional.
  exchange: 'pubsub', // Name of the queue to use. (Default: pubsub)
  url: 'amqp://localhost:5672' // Can also be specified on connect()
});
 
// URL here is optional, can also be set as above.
await pubsub.connect('amqp://localhost:5672');
 
pubsub.on('error', (err) => {
  // Something went wrong and you should log it.
});
pubsub.on('close', (err) => {
  // The connection was closed and the queue is no longer usable.
  // An err is given if the connection was closed due to an error.
});
 
// Publish a message.
await pubsub.publish('login', { userId: 1 });
 
// Subscribe to a topic.
const unsub = await pubsub.subscribe('login', async (message) => {
  await loggedIn(message.userId);
});
 
// Cancel subscription to stop receiving messages.
await unsub();
 
// Close connection.
await pubsub.close();

Tests

Run the libraries integration tests with:

$ AMQP_URL="amqp://user:pass@localhost:5672" npm test

To determine test coverage run:

$ AMQP_URL="amqp://user:pass@localhost:5672" npm run coverage

Contributions

Are always welcome!

  • Initial development by @patrickd-
  • Name suggested by @leschekfm :D

install

npm i wrappitmq

Downloadsweekly downloads

116

version

1.0.4

license

MIT

homepage

github.com

repository

Gitgithub

last publish

collaborators

  • avatar
Report a vulnerability