mq-one

A lightweight TypeScript library for building reliable asynchronous message processing pipelines.

This library provides a reusable processing pipeline for asynchronous messages, including:

• Message validation
• Automatic retries
• Retry queues
• Dead-letter handling
• Logging
• JSON deserialization
• Error handling

It is designed to work with any messaging system, including:

• ActiveMQ
  • activemq, to wrap and simplify amqplib
• RabbitMQ
  • rabbitmq-ext, to wrap and simplify amqplib
• Apache Kafka
  • kafka-plus, to wrap and simplify kafkajs
• Google Pub/Sub
  • google-pubsub, to wrap and simplify @google-cloud/pubsub
• NATS
  • nats-plus, to wrap and simplify nats
• IBM MQ
  • ibmmq-plus, to wrap and simplify ibmmq
• Amazon SQS
• Azure Service Bus
• Custom message transports

The transport library is responsible for receiving and sending messages. This library is responsible for processing them.

Examples:

• rabbitmq-sample: An example to consume message from rabbitmq.
• activemq-sample: An example to consume message from activemq.
• pubsub-sample: An example to consume message from pubsub.
• ibmmq-sample: An example to consume message from ibmmq.
• kafka-sample: An example to consume message from kafka.
• nats-sample: An example to consume message from nats.
• sqs-sample: An example to consume message from AWS sqs.

Features

• Transport independent
• Generic TypeScript API
• Validation before processing
• Immediate retry with configurable delays
• Retry queue support
• Dead-letter/error handler support
• Retry count stored in message headers
• Configurable logging
• Lightweight with no messaging-system dependency

Architecture

                Message
                   │
                   ▼
            Parse (optional)
                   │
                   ▼
                Validate
                   │
         ┌─────────┴──────────┐
         │                    │
       Invalid              Valid
         │                    │
         ▼                    ▼
       Ignore            Business Logic
                              │
                      ┌───────┴────────┐
                      │                │
                  Success          Failure
                      │                │
                      ▼                ▼
                    Finish      Retry Strategy
                                      │
                    ┌─────────────────┴──────────────────┐
                    │                                    │
            Immediate Retry                         Retry Queue
                    │                                    │
              Retry Delays                       Re-publish Message
                    │                                    │
                    └─────────────────┬──────────────────┘
                                      │
                                      ▼
                             Retry Limit Reached
                                      │
                                      ▼
                             Dead Letter Handler

Installation

npm install mq-one

or

yarn add mq-one

Concepts

The framework separates responsibilities into independent components.

| Component | Responsibility | |-----------|----------------| | Handler | Coordinates the processing pipeline | | validate | Validates incoming messages | | write | Executes business logic | | retry | Sends failed messages to a retry queue | | handleError | Handles permanently failed messages | | RetryWriter | Retries immediately with delays | | RetrySender | Retries sending operations |

Basic Usage

import { Handler } from "mq-one";

const handler = new Handler<Order>(
    saveOrder,
    validateOrder,
    [],
    saveDeadLetter,
    console.error,
    console.log
);

await handler.handle(message);

Validation

Validation runs before business logic.

async function validateOrder(order: Order) {
    const errors = [];

    if (!order.id) {
        errors.push({
            field: "id",
            code: "required"
        });
    }

    return errors;
}

If validation returns errors, business processing is skipped.

Business Logic

Business logic only focuses on processing.

async function saveOrder(order: Order): Promise<number> {
    await repository.save(order);
    return 1;
}

The framework manages retries and error handling.

Retry Queue

Instead of retrying immediately, failed messages can be sent back to another queue.

const handler = new Handler(
    saveOrder,
    validateOrder,
    [],
    saveDeadLetter,
    console.error,
    console.log,
    retrySender,
    5
);

Whenever processing fails:

     Consumer

         ↓

  Business Failed

         ↓

Publish to Retry Queue

         ↓

 Consume Again Later

This prevents consumers from being blocked for long periods.

Immediate Retry

Immediate retry is useful for temporary failures such as:

• Database unavailable
• Temporary network issues
• HTTP timeout
• Cloud services, such as Google Firestore
  • When Application cannot write data to Google Firestore, normally the network between Application and Google Cloud is down
  • If Application retry after 1 second, normally it is still down
  • The best approach is to retry after 30 seconds, then 60 seconds, then 180 seconds

const delays = [
    30000,
    60000,
    180000
];

await writeWithRetry(
    order,
    saveOrder,
    delays
);

Retry sequence:

     Try

      ↓

 Wait 30 seconds

      ↓

    Retry

      ↓

 Wait 60 seconds

      ↓

    Retry

      ↓

Wait 180 seconds

      ↓

    Retry

RetryWriter

RetryWriter wraps any write operation with retry logic.

const writer = new RetryWriter(
    saveOrder,
    [30000, 60000, 180000]
);

await writer.write(order);

RetrySender

RetrySender wraps any send operation.

const sender = new RetrySender(
    publishOrder,
    [30000, 60000]
);

await sender.send(order);

Retry Metadata

Retry count is stored in message headers.

{
    retryCount: "3"
}

This keeps retry information separate from the business payload.

The header name is configurable.

new Handler(
    write,
    validate,
    [],
    deadLetter,
    console.error,
    console.log,
    retry,
    10,
    "retry"
);

Dead Letter Handler

After the retry limit is exceeded, the framework invokes the configured error handler.

async function saveDeadLetter(
    order: Order,
    headers?: StringMap
) {
    await repository.save(order);
}

Possible implementations include:

• Dead Letter Queue
• Database
• Log file
• Email notification
• Monitoring system

Logging

Optional logging callbacks provide visibility into processing.

const handler = new Handler(
    write,
    validate,
    [],
    deadLetter,
    console.error,
    console.log
);

The framework logs events such as:

• Message received
• Validation failure
• Retry started
• Retry completed
• Retry exhausted
• Dead-letter processing

Helper Functions

write()

Executes business logic and optionally sends failed messages to a retry queue.

await write(
    processOrder,
    order,
    headers,
    deadLetter,
    retrySender,
    5
);

writeWithRetry()

Retries immediately using configurable delays.

await writeWithRetry(
    order,
    processOrder,
    [30000, 60000, 180000]
);

createRetry()

Creates a retry delay array from a numbered configuration object.

const delays = createRetry({
    1: 30000,
    2: 60000,
    3: 180000
});

Produces:

[30000, 60000, 180000]

Design Philosophy

The library follows a simple processing pipeline.

Receive Message

      ↓

   Validate

      ↓

Business Logic

      ↓

Retry Strategy

      ↓

  Dead Letter

Each stage has a single responsibility and can be replaced independently.

The messaging system is intentionally abstracted away, allowing the same processing pipeline to be reused across different transports.

Typical Integrations

This library can be combined with:

• RabbitMQ
• Apache Kafka
• Amazon SQS
• Azure Service Bus
• Google Pub/Sub
• NATS
• Redis Streams
• Custom messaging systems

Only the transport layer changes; the processing pipeline remains the same.

License

MIT