amqpemitter v1.2.0

AMQPEmitter

Reliable distributed AMQP EventEmitter with a customized API supporting both request/response and publish/subscribe patterns

Building on top of eventemitter2, a specialized API is exposed for request/response and publish/subscribe patterns.

Installation

$ npm install --save amqpemitter

Getting Started

const EventEmitter = require('amqpemitter');
const emitter = new EventEmitter(options);

(async () => {
  await emitter.connect();
  await emitter.onAll('example.message', (msg) => {
    console.log('Received message:', msg);
  });
  await emitter.emitAll('example.message', 'Hello World');
})()

// Prints 'Received message: Hello World' after the library connected to the local AMQP server

For a more complete example, take a look at the examples directory

API

new AMQPEmitter(options): Takes an options object and constructs a new AMQPEmitter.

const emitter = new AMQPEmitter({
  // Information about the server to connect to
  server: {
    url: 'amqp://localhost', // URL parameter passed to amqp.node, can also be an object describing the connection
    options: {}, // Further options passed to amqp.node, for example for specifying a CA certificate. For more information, take a look at examples/config.js
  },

  exchange: 'eventemitter', // The default exchange name to be used for passing publish/subscribe messages

  requestTimeout: 30, // Timeout in seconds after which an emitOne() request is discarded

  reconnectTimeout: 5, // Seconds to wait until reconnecting to the AMQP server if a connection is lost
});

connect(): (Re)connects to the AMQPEmitter. Returns a promise that resolves when the connection has been established and is usable. Also fires connected event after a connection has been established.

await emitter.connect();

disconnect(reason): Disconnects from the AMQP server and fires the disconnect event with a given reason.

await emitter.disconnect();

Internal Events

on(event, listener): Subscribe to internal events error, connected, disconnected. Returns a promise which resolves after successfully registering the listener.

await emitter.on('error', (err) => {
  console.error(err);
});

once(event, listener): Unsubscribes automatically after listener has been called once.

many(event, amount, listener): Unsubscribes automatically after listener has been called amount times.

off(event, listener): Unsubscribes from a previously subscribed event.

emit(event, payload): Fires internal event. Internal events are not sent over the wire, only evaluated locally.

await emitter.emit('error', new Error('Something terrible happened'));

Request/Response

Note: listeners are callback functions whose first argument is always the actual event that was broadcasted, in string form. This differs from the registered event if wildcards were used

onOne(event, listener): Subscribes to a request/response style event. Emitted messages are delivered to exactly one registered listener across the whole network, its return value is delivered back to emitOne(). Returns a promise which resolves as soon as the event is usable, prior to that messages are not guaranteed to pass through it.

await emitter.onOne('example.sendEmail', async (event, {name, email}) => {
  await internal.sendEmail(name, email);
  return `An email has been sent to ${email}.`;
});

onceOne(event, listener): Unsubscribes automatically after listener has been called once.

manyOne(event, amount, listener): Unsubscribes automatically after listener has been called amount times.

offOne(event, listener): Unsubscribes from a previously subscribed event.

emitOne(event, payload): Fires a request/response style event. Only the very first argument is sent, serialized as JSON data. Returns a promise which resolves to the return value of the listener that handles this request.

console.log(await emitter.emitOne('example.sendEmail', {name: 'John Doe', email: 'john@example.com'}));
// Prints 'An email has been sent to john@example.com.'

Publish/Subscribe

Note: listeners are callback functions whose first argument is always the actual event that was broadcast, in string form. This differs from the registered event if wildcards were used

onAll(event, listener): Subscribes to a publish/subscribe style event. Emitted messages are delivered to all registered listeners, as such its return value is discarded. Returns a promise which resolves as soon as the event is usable, prior to that messages are not guaranteed to pass through it.

await emitter.onAll('example.log', async (event, msg) => {
  console.log(msg);
});

onceAll(event, listener): Unsubscribes automatically after listener has been called once.

manyAll(event, amount, listener): Unsubscribes automatically after listener has been called amount times.

offAll(event, listener): Unsubscribes from a previously subscribed event.

emitAll(event, payload): Fires a publish/subscribe style event. Only the very first payload argument is sent, serialized as JSON data. Returns a promise which resolves as soon as the AMQP server indicates the message has been handled.

await emitter.emitAll('example.log', 'User John logged in');
await emitter.emitAll('example.log', 'User John added last name Doe to his profile');
// Prints both log messages