scamp v0.0.1

scamp

An advance Node.js RabbitMQ client from the author of Rascal. It is in the early stages of development, has an unstable API and is not production ready.

Connection Topologies

Scamp allows you to choose your connection topology by providing a range of pluggable connection and channel sources. For example...

Dedicated connection with a dedicated channel

  ┌─────────────────────────┐                                            ┌─────────────────────────┐
  │                         │                 Connection                 │                         │
  │                         ├────────────────────────────────────────────┤                         │
  │                         │                                            │                         │
  │   Producer / Consumer   ╠══════════════════Channel═══════════════════╣      Virtual Host       │
  │                         │                                            │                         │
  │                         ├────────────────────────────────────────────┤                         │
  │                         │                                            │                         │
  │                         │                                            │                         │
  └─────────────────────────┘                                            └─────────────────────────┘

const amqplib = require('amqplib');
const { DedicatedConnectionSource, DedicatedChannelSource } = require('scamp');
const connectionOptions = { hostname: 'rabbitmq.example.com' };
const socketOptions = { timeout: 10000 };
const connectionSource = new DedicatedConnectionSource({ amqplib, connectionOptions, socketOptions });
const channelSource = new DedicatedChannelSource({ connectionSource });

const channel = await channelSource.getChannel();

Dedicated connection with a channel pool

  ┌─────────────────────────┐                                            ┌─────────────────────────┐
  │                         │                 Connection                 │                         │
  │                         ├────────────────────────────────────────────┤                         │
  │                         │                                            │                         │
  │                         ╠═════════════════Channel 1══════════════════╣                         │
  │                         │                                            │                         │
  │        Producer         │                                            │      Virtual Host       │
  │                         │                                            │                         │
  │                         ╠═════════════════Channel 2══════════════════╣                         │
  │                         │                                            │                         │
  │                         ├────────────────────────────────────────────┤                         │
  │                         │                                            │                         │
  └─────────────────────────┘                                            └─────────────────────────┘

const amqplib = require('amqplib');
const { DedicatedConnectionSource, PooledChannelSource } = require('scamp');
const connectionOptions = { hostname: 'rabbitmq.example.com' };
const socketOptions = { timeout: 10000 };
const connectionSource = new DedicatedConnectionSource({ amqplib, connectionOptions, socketOptions });
const channelSource = new PooledChannelSource({ size: 2, connectionSource });

const channel = await channelSource.getChannel();

Connection pool with dedicated channels

  ┌─────────────────────────┐                Connection 1                ┌─────────────────────────┐
  │                         ├────────────────────────────────────────────┤                         │
  │                         │                                            │                         │
  │                         ╠════════════════Channel 1═══════════════════╣                         │
  │                         │                                            │                         │
  │                         ├────────────────────────────────────────────┤                         │
  │        Producer         │                                            │      Virtual Host       │
  │                         │                Connection 2                │                         │
  │                         ├────────────────────────────────────────────┤                         │
  │                         │                                            │                         │
  │                         ╠═════════════════Channel 2══════════════════╣                         │
  │                         │                                            │                         │
  │                         ├────────────────────────────────────────────┤                         │
  └─────────────────────────┘                                            └─────────────────────────┘

const amqplib = require('amqplib');
const { DedicatedConnectionSource, DedicatedChannelSource, MultiChannelSource } = require('scamp');
const connectionOptions = { hostname: 'rabbitmq.example.com' };
const socketOptions = { timeout: 10000 };
const channelSources = [
  new DedicatedChannelSource({ connectionSource: new DedicatedConnectionSource({ amqplib, connectionOptions, socketOptions }) }),
  new DedicatedChannelSource({ connectionSource: new DedicatedConnectionSource({ amqplib, connectionOptions, socketOptions }) }),
];
const channelSource = new MultiChannelSource({ channelSources });

const channel = await channelSource.getChannel();

Dedicated active/passive connection with dedicated channel

  ┌─────────────────────────┐                Connection 1                ┌─────────────────────────┐
  │                         ├────────────────────────────────────────────┤                         │
  │                         │                 Channel 1                  │                         │
  │                         ╠════════════════════════════════════════════╣      Virtual Host       │
  │                         │                                            │                         │
  │                         ├────────────────────────────────────────────┤                         │
  │   Producer / Consumer   │                                            └─────────────────────────┘
  │                         │           Connection 2 (passive)           ┌─────────────────────────┐
  │                         ├ ─ ─ ─ ─ ─  ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ │                         │
  │                         │                                            │                         │
  │                         ├ ─ ─ ─ ─ ─ ─ Channel 2 (passive) ─ ─ ─ ─ ─ ─│      Virtual Host       │
  │                         │                                            │                         │
  │                         ├ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─│                         │
  └─────────────────────────┘                                            └─────────────────────────┘

const amqplib = require('amqplib');
const { DedicatedConnectionSource, DedicatedChannelSource } = require('scamp');
const optionSets = [
  {
    connectionOptions: { hostname: 'rabbitmq-primary.example.com' },
    socketOptions: { timeout: 10000 },
  },
  {
    connectionOptions: { hostname: 'rabbitmq-secondary.example.com' },
    socketOptions: { timeout: 10000 },
  },
];
const connectionSource = new HighAvailabilityConnectionSource({ amqplib, optionSets });
const channelSource = new DedicatedChannelSource({ connectionSource });

const channel = await channelSource.getChannel();

Broker

The broker is a container for one or more virtual hosts and repository for common config such as encryption profiles and content parsers.

const broker = new Broker({
  encryption: {
    'profile-1': {
      key: 'blah-blah-blah',
      algorithm: 'aes256',
      ivLength: 16,
    }
  },
  parsers: {
    'application/json': new JsonParser(),
    'application/xml', new XmlParser(),
  }
});

You can shutdown all vhosts managed by the broker using broker.shutdown. You can also nuke, or purge matching queues and exchanges, which is useful for testing. e.g.

let broker;

beforeAll(async () => {
  broker = await initBroker();
});

beforeEach(async () => {
  await broker.purge();
});

afterAll(async () => {
  await broker.nuke(/^test_/);
});

Vhosts

Virtual Hosts are obtained by connecting to a broker.

const connectionSource = new SimpleConnectionSource();
const vhost = await broker.connect({
  connectionSource,
  options: {
    host: 'localhost',
    port: 5672,
    name: 'vh1',
    user: 'bob',
    password: 'secret',
    params: {
      heartbeat: 10,
    },
    socket: {
      timeout: 10000,
    }
  }
});

Options

Exchanges

Exchanges are obtained from a virtual host using vhost.declareExchange. You can use the passive option to determine whether the exchange should be created if it doesn't already exist. Declaring an exchange passively which does not already exist will result in an error. Attempting to redeclare an exchange with different attributes will also result in an error. Once you have an instance of an exchange you can create a producer and start publishing messages.

const exchange = await vhost.declareExchange({
  name: 'ex1',
  type: 'topic',
  passive: false,
  arguments: {
    'x-dead-letter-exchange': 'dlx',
  },
});

Options

Queues

Like exchanges, queues are also obtained from a virtual host using vhost.declareQueue. You can use the passive option to determine whether the queue should be created if it doesn't already exist. Declaring a queue passively which does not already exist will result in an error. Attempting to redeclare a queue with different attributes will also result in an error. Once you have an instance of an queue you can create a producer and start publishing messages.

const queue = await vhost.declareQueue({
  name: 'q1',
  passive: false,
  arguments: {
    'x-message-ttl': 1000,
  },
});

Binding

You can bind queues to exchanges as follows...

// Fanout exchange
await queue.bind(exchange);

// Topic or direct exchange
await queue.bind(exchange, { keys: ['a.b.c'] });

// Headers exchange
await queue.bind(exchange, { arguments { format: 'pdf', type: 'report', 'x-match': 'all' } });

Or bind two exchanges like this...

// Fanout exchange
await exchange1.bind(exchange2);

// Topic or direct exchange
await exchange1.bind(exchange2, { keys: ['a.b.c'] });

// Header exchange
await exchange1.bind(exchange2, { arguments: { format: 'pdf', type: 'report', 'x-match': 'all' } });

You can also unbind, but you must be careful to use binding keys which actually exist.

await queue.unbind(exchange { keys });
await exchange.unbind(exchange2, { keys });

Options

Producers

Producers are obtained from queues or exchanges. They require a channel source for obtaining channels. For example...

const channelSource = new SimpleChannelSource();
const producer = queue.getProducer({ channelSource })
  .setMessageId(uuid)
  .useConfirmChannel(true)
  .detectContentType(true)
  .encryptContent('profile-1'})
  .timeout(1000);

await new Promise((resolve, reject) => {
  producer.publish('hello world')
    .on('success', message => {
      resolve();
    })
    .on('returned', message => {
      // Already resolved
      console.warn(`Message was returned ${message.properties.messageId}`);
    })
    .on('error', (err, message) => {
      reject(err);
    })
    .on('busy', (message) => {
      // May want to apply back pressure to whatever is calling the publisher
    })
    .on('ready', () => {
      // May want to relax back pressure
    });
});

await producer.close();
await vhost.disconnect();

Consumers

Consumers are obtained from a queue. They require a channel source for obtaining channels. For example...

const channelSource = new SimpleChannelSource();
const consumer = queue.createConsumer({ channelSource })
  .setPrefetch(10)
  .setContentType('application/xml')
  .decryptContent(true)
  .parseContent(true);

// The consumer will only start consuming messages once the on message handler is registered
consumer.subscribe()
  .on('message', (message) => {
    console.log(message.content);
    await message.ack();
  }).on('error', (err, message) => {
    console.error(err);
  });