amqplib

What is amqplib?

The amqplib npm package is a library for Node.js that provides an interface for interacting with AMQP 0-9-1 compatible message brokers, such as RabbitMQ. It allows for the sending and receiving of messages over queues and exchanges, and supports various messaging patterns.

What are amqplib's main functionalities?

Connecting to a RabbitMQ server

This code demonstrates how to connect to a RabbitMQ server using amqplib. It establishes a connection and creates a channel for sending and receiving messages.

const amqp = require('amqplib');

async function connect() {
  try {
    const connection = await amqp.connect('amqp://localhost');
    const channel = await connection.createChannel();
    console.log('Connected to RabbitMQ');
    // Additional code to interact with the channel
  } catch (error) {
    console.error('Connection failed', error);
  }
}

connect();

Sending a message to a queue

This code snippet shows how to send a message to a specific queue. It connects to the server, creates a channel, asserts the queue, sends the message, and then closes the channel and connection.

const amqp = require('amqplib');

async function sendMessage(queue, message) {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();
  await channel.assertQueue(queue, { durable: false });
  channel.sendToQueue(queue, Buffer.from(message));
  console.log(`Message sent: ${message}`);
  await channel.close();
  await connection.close();
}

sendMessage('myQueue', 'Hello World!');

Receiving messages from a queue

This example demonstrates how to receive messages from a queue. It sets up a consumer that listens for messages on the specified queue and acknowledges them after processing.

const amqp = require('amqplib');

async function receiveMessages(queue) {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();
  await channel.assertQueue(queue, { durable: false });
  channel.consume(queue, (msg) => {
    if (msg !== null) {
      console.log(`Received message: ${msg.content.toString()}`);
      channel.ack(msg);
    }
  });
}

receiveMessages('myQueue');

Setting up exchanges and routing

This code sets up an exchange and a queue, then binds them with a routing key. It also sets up a consumer to receive messages sent to the exchange with the specific routing key.

const amqp = require('amqplib');

async function setupExchangeAndRouting() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();
  await channel.assertExchange('myExchange', 'direct', { durable: false });
  const { queue } = await channel.assertQueue('', { exclusive: true });
  channel.bindQueue(queue, 'myExchange', 'myRoutingKey');
  channel.consume(queue, (msg) => {
    if (msg !== null) {
      console.log(`Received: ${msg.content.toString()}`);
      channel.ack(msg);
    }
  });
}

setupExchangeAndRouting();

Other packages similar to amqplib

rabbit.js

rabbit.js is a messaging library for Node.js that is built on top of amqplib. It provides a simpler API for common messaging patterns. However, it may not be as actively maintained or feature-rich as amqplib.

rascal

rascal is an advanced message bus for Node.js that wraps amqplib. It provides additional features like configuration-based setup, automatic reconnection, and message redelivery. It's a higher-level abstraction compared to amqplib.

seneca-amqp-transport

seneca-amqp-transport is a plugin for the Seneca microservices toolkit that allows Seneca-based microservices to communicate over AMQP using RabbitMQ. It's more specialized for use with Seneca, whereas amqplib is a general-purpose AMQP library.

README

AMQP 0-9-1 library and client for Node.JS

npm install amqplib

• Change log
• GitHub pages
• API reference
• Examples from RabbitMQ tutorials

A library for making AMQP 0-9-1 clients for Node.JS, and an AMQP 0-9-1 client for Node.JS v0.8-0.12, v4-v9, and the intervening io.js releases.

This library does not implement AMQP 1.0 or AMQP 0-10.

Project status:

• Expected to work
• Complete high-level and low-level APIs (i.e., all bits of the protocol)
• Stable APIs
• A fair few tests
• Measured test coverage
• Ports of the RabbitMQ tutorials as examples
• Used in production

Still working on:

• Getting to 100% (or very close to 100%) test coverage

Callback API example

var q = 'tasks';

function bail(err) {
  console.error(err);
  process.exit(1);
}

// Publisher
function publisher(conn) {
  conn.createChannel(on_open);
  function on_open(err, ch) {
    if (err != null) bail(err);
    ch.assertQueue(q);
    ch.sendToQueue(q, Buffer.from('something to do'));
  }
}

// Consumer
function consumer(conn) {
  var ok = conn.createChannel(on_open);
  function on_open(err, ch) {
    if (err != null) bail(err);
    ch.assertQueue(q);
    ch.consume(q, function(msg) {
      if (msg !== null) {
        console.log(msg.content.toString());
        ch.ack(msg);
      }
    });
  }
}

require('amqplib/callback_api')
  .connect('amqp://localhost', function(err, conn) {
    if (err != null) bail(err);
    consumer(conn);
    publisher(conn);
  });

Promise API example

var q = 'tasks';

var open = require('amqplib').connect('amqp://localhost');

// Publisher
open.then(function(conn) {
  return conn.createChannel();
}).then(function(ch) {
  return ch.assertQueue(q).then(function(ok) {
    return ch.sendToQueue(q, Buffer.from('something to do'));
  });
}).catch(console.warn);

// Consumer
open.then(function(conn) {
  return conn.createChannel();
}).then(function(ch) {
  return ch.assertQueue(q).then(function(ok) {
    return ch.consume(q, function(msg) {
      if (msg !== null) {
        console.log(msg.content.toString());
        ch.ack(msg);
      }
    });
  });
}).catch(console.warn);

Running tests

npm test

To run the tests RabbitMQ is required. Either install it with your package manager, or use docker to run a RabbitMQ instance.

docker run -d --name amqp.test -p 5672:5672 rabbitmq

If prefer not to run RabbitMQ locally it is also possible to use a instance of RabbitMQ hosted elsewhere. Use the URL environment variable to configure a different amqp host to connect to. You may also need to do this if docker is not on localhost; e.g., if it's running in docker-machine.

One public host is dev.rabbitmq.com:

URL=amqp://dev.rabbitmq.com npm test

NB You may experience test failures due to timeouts if using the dev.rabbitmq.com instance.

You can run it under different versions of Node.JS using nave:

nave use 0.8 npm test

or run the tests on all supported versions of Node.JS in one go:

make test-all-nodejs

(which also needs nave installed, of course).

Lastly, setting the environment variable LOG_ERRORS will cause the tests to output error messages encountered, to the console; this is really only useful for checking the kind and formatting of the errors.

LOG_ERRORS=true npm test

Test coverage

make coverage
open file://`pwd`/coverage/lcov-report/index.html

Changelog

Changes in v0.5.5

git log v0.5.3..v0.5.5

NB this includes a minor but possibly breaking change: after PR 498, all confirmation promises still unresolved will be rejected when their associated channel is closed.

• Generate defs in npm prepare rather than npm prepublish so that e.g., amqplib can be installed via git (part of PR 498)
• Reject all pending confirmations when the channel is closed (PR 498); thanks @johanneswuerbach
• Update supported NodeJS versions in package.json (PR 525); thanks @tingwai