@axah/mq

@axah/mq a Node library that provides a convenient and easy-to-use wrapper for Rascal, which itself is a wrapper for AMQP (Advanced Message Queuing Protocol) client library amqplib to be used in the convenience plattform .

Concept

This library aims to simplify the process of setting up a Rascal config and provide an easy way to subscribe and publish to a queue. In case the default config is not suitable there is always the option to use the full feature set of rascal. In the best case scenario it takes a queue name and generates a Rascal vhost configuration with a default exchange, the specified queue name, a dead letter queue, bindings, subscriptions and publications.

Using the following configuration:

{
"queueName": "document-extraction-request.main",
"options": { "withSubscription": true }
}

will generate the following rascal config.

{
  "vhosts": {
    "rabbit": {
      "connection": {
        "hostname": "localhost",
        "management": {
          "hostname": "localhost",
          "port": "15672",
          "protocol": "http",
          "user": "admin",
          "password": "*****",
          "vhost": "/"
        },
        "port": "5672",
        "protocol": "amqp",
        "user": "admin",
        "password": "*****",
        "vhost": ""
      },
      "exchanges": [
        {
          "assert": true,
          "name": "document-extraction-request.exchange",
          "type": "direct"
        }
      ],
      "queues": [
        {
          "assert": true,
          "name": "document-extraction-request.main",
          "options": {
            "arguments": {
              "x-dead-letter-exchange": "document-extraction-request.exchange",
              "x-dead-letter-routing-key": "document-extraction-request.backout"
            }
          }
        },
        {
          "assert": true,
          "name": "document-extraction-request.backout"
        }
      ],
      "bindings": [
        {
          "bindingKeys": [
            "document-extraction-request.main"
          ],
          "destination": "document-extraction-request.main",
          "destinationType": "queue",
          "source": "document-extraction-request.exchange"
        },
        {
          "bindingKey": "document-extraction-request.backout",
          "destination": "document-extraction-request.backout",
          "destinationType": "queue",
          "source": "document-extraction-request.exchange"
        }
      ],
      "publications": {},
      "subscriptions": {
        "document-extraction-request-subscription": {
          "contentType": "application/json",
          "queue": "document-extraction-request.main",
          "vhost": "rabbit"
        }
      }
    }
  } 
}

Getting Started

To use @axah/mq in your Node.js project, you need to install it first via pnpm:

pnpm install @axah/mq
pnpm install -D  @types/rascal

After installation, you can use the library in your code:

import {mq} from '@axah/mq';

const config = {
    mq: {
        paymentQueue: {
            queueName: 'payment-service-address-update.main',
            options: {
                withSubscription: true,
                withPublication: true,
            },
        },
    }
}

const addressUpdateMessageSchema = z.object({id: z.string()});
type AddressUpdateMessage = z.infer<typeof addressUpdateMessageSchema>;

//setup and assert queue
await mq.initBroker(config.mq, log);

//publish to queue 
await mq.publish({
    vHostConfig: config.mq.paymentQueue,
    contentValidator: addressUpdateMessageSchema,
    content: {id: "b3b2273b-25a7-42bd-9b6d-bf49615eddef"},
    log,
});

//publish to queue 
await mq.subscribe({
    vHostConfig: config.mq.paymentQueue,
    contentValidator: addressUpdateMessageSchema,
    messageHandler: createOpenPositionMessageHandler,
    log,
});

const updateAddressConsumer = async ({ content} : Message<AddressUpdateMessage>) => {
 //handle message
}