@thecolvinco/nodejs-amqplib

Colvin nodejs rabbitmq utils

A RabbitMQ package with some utils like retryables to use across diferents projects.

Usage

Consumer/worker example

import { amqpConnect, retryable, worker, deadLetter } from '@thecolvinco/nodejs-amqplib';

const connectionString = 'amqp://rabbitmq:rabbitmq@localhost:5672';
const args = process.argv.slice(2);
const queueName = args[0] || 'send-invoice-on-shopify-order-created';
const bindingKey = args[1] || 'blom.superapp.1.event.shopify.order-created';
const dlxExchange = args[2] || 'blom.superapp.dlx.direct';
const dlxRoutingKey = args[3] || 'blom.superapp.dlx';
const dlxQueueName = args[4] || 'blom-superapp-dlx-queue';

const workable = (channel) => {
  return async (msg) => {
    console.info(`Message received ${msg}`);
    
    // Simulate some retryables
    if (msg.content.toString() === 'retry') {
      await retryable({
        channel,
        message: msg,
        queue: {
          name: queueName,
        },
        retryExchange: {
          name: 'retries.exchange',
        },
        maxRetries: 3,
        delay: 1000,
      });
    } else {
      channel.ack(msg);
    };
  };
};

try {
  const connection = await amqpConnect(connectionString);
  const channel = await conn.createChannel();

  const exchangeData = {
    name: 'blom.exchange.topic',
    type: 'topic',
  };
    
  const queueData = {
    name: queueName,
    bindingKey,
    options: {
      durable: true,
      deadLetterExchange: dlxExchange,
      deadLetterRoutingKey: dlxRoutingKey
    },
    dlx: {
      func: deadLetter,
      params: {
        channel,
        dlxQueue: {
          name: dlxQueueName,
        },
        dlxExchange: {
          name: dlxExchange,
          type: 'direct',
        },
      },
    },
  };

  const consumerData = {
    onMessage: workable(channel),
    options: {
      noAck: false
    },
  };

  await worker({
    channel,
    exchange: exchangeData,
    queue: queueData,
    consumer: consumerData,
  });

  console.log("Wating for messages....");
} catch(error) {
  console.log(error)
}

Producer example

import { amqpConnect, producer } from '@thecolvinco/nodejs-amqplib';

const connectionString = 'amqp://rabbitmq:rabbitmq@localhost:5672';
const args = process.argv.slice(2);

try {
  const connection = await amqpConnect(connectionString);
  const channel = await conn.createChannel();

  const message = args[0] || 'Hello world';
  const key = args[1] || 'blom.superapp.1.event.shopify.order-created';
  const exchange = {
    name: 'blom.exchange.topic',
    type: 'topic',
  };

  await producer({
    channel,
    message,
    key,
    exchange,
  });

  console.log(`[x] Sent ${message} to ${exchange.name} with key ${key}`);

  connection.close();
} catch (error) {
  console.log(error);
}

About

This package is maintained by TheColvinCo

LICENSE

Code is licensed under the MIT License.