Yoctol Message Queue

The package provide two pattern of message queue:

• Queue
• Pubsub

Usage

$ yarn add @yoctol/message-queue

Queue

The queue has two driver: 'memory' and 'rabbitmq'.

Memory Queue

import { connectQueue } from '@yoctol/message-queue';

const queue = await connectQueue({ driver: 'memory', queueName: 'queueName' });

RabbitMQ Queue

import { connectQueue } from '@yoctol/message-queue';

const queue = await connectQueue({
  driver: 'rabbitmq',
  queueName: 'queueName',
  amqpConfig: {
    url: 'amqp://localhost',
  },
  deadLetterConfig: {
    waitExchange: 'wait-exchange',
    maxRetryCount: 3,
    delayFactor: 1000,
    retryJitter: true,
  }
});

Enabling Dead Letter Handling in RabbitMQ Queue

When given deadLetterConfig, the dead letter handling flow will be enabled. Enabling this feature will automatic retry dead letter messages(which throws errors during executing consume handler) with exponential back off with/without jitter.

To enable dead letter handling, you can follow the settings:

const queue = await connectQueue({
  driver: 'rabbitmq',
  amqpConfig: {
    url: 'amqp://localhost',
  },
  queueName: 'queueName',
  deadLetterConfig: {
    waitExchange: 'wait-exchange', // The exchange will be created with the name given
    maxRetryCount: 3, // If retry amount exceed the maxRetryCount, the reachMaxRetryHandler will be called
    delayFactor: 1000, // The final delay time (in milliseconds) is generate by this formula: delayFactor * delay base time (2^retryCount if retry jitter is disabled)
    retryJitter: true, // If retryJitter is enabled, the delay base time will be random integer from 1 to 2^retryCount
  }
});

// If retry amount exceed the maxRetryCount, the reachMaxRetryHandler will be called with (err, bufferMsg)
queue.consume(handler, reachMaxRetryHandler)

PubSub

The pubsub has three driver: 'memory' and 'postgres', and 'rabbitmq'.

Memory Pubsub

import { connectPubsub } from '@yoctol/message-queue';

const pubsub = await connectPubsub({ driver: 'memory' });

Postgres Pubsub

We use pg as client, see https://github.com/brianc/node-postgres for config info.

import { connectPubsub } from '@yoctol/message-queue';

// for more info, check https://github.com/brianc/node-postgres
const config = {
  host: process.env.DATABASE_HOST,
  port: process.env.DATABASE_PORT,
  database: process.env.DATABASE_NAME,
  user: process.env.DATABASE_USER,
  password: process.env.DATABASE_PASSWORD,
};
const pubsub = await connectPubsub({
  driver: 'postgres',
  pgConfig: config,
});

RabbitMQ Pubsub

import { connectPubsub } from '@yoctol/message-queue';

const pubsub = await connectPubsub({
  driver: 'rabbitmq',
  amqpConfig: {
    url: 'amqp://localhost',
    exchange: 'amqp.exchange',
    connectOptions: {
      //node-amqp-connection-manager connect() options 
      heartbeatIntervalInSeconds: 5,
      ...
      connectionOptions: {
        //amqplib connection options
        ...
      },
    },
    createChannelOptions: {
      //node-amqp-connection-manager createChannel() options 
      name: 'channelName',
      setup: ...,
      json: true,
    },
 },
});