Adapters & transports

The @babelqueue/core codec is framework-agnostic. Thin packages wire it into common Node stacks: @babelqueue/bullmq (BullMQ jobs) and @babelqueue/nestjs (a NestJS module built on the BullMQ adapter), plus broker transports — @babelqueue/redis (Redis), @babelqueue/rabbitmq (RabbitMQ), @babelqueue/sqs (Amazon SQS), @babelqueue/azure-service-bus (Azure Service Bus), @babelqueue/pulsar (Apache Pulsar), @babelqueue/kafka (Apache Kafka) and @babelqueue/artemis (Apache ActiveMQ Artemis).

BullMQ — `@babelqueue/bullmq`

npm install @babelqueue/bullmq bullmq

bullmq ^5 is a peer dependency; @babelqueue/core is pulled in for you.

It exports two functions:

publish(queue, urn, data, options?) → Promise<string> — adds a BullMQ job whose name is the URN and whose data is the canonical envelope; returns meta.id. options is { traceId?, jobsOptions? } (jobsOptions is BullMQ’s own — delay, attempts, backoff…).
processor(handlers, options?) → (job) => Promise — a BullMQ processor function that validates each envelope, resolves its URN and routes to handlers[urn]. options.onUnknownUrn(envelope, job) handles URNs with no mapped handler.

Produce

import { Queue } from "bullmq";
import { publish } from "@babelqueue/bullmq";

const queue = new Queue("orders", { connection: { host: "localhost", port: 6379 } });

const id = await publish(queue, "urn:babel:orders:created", { order_id: 1042 });

Consume

import { Worker } from "bullmq";
import { processor } from "@babelqueue/bullmq";

new Worker(
  "orders",
  processor(
    {
      "urn:babel:orders:created": async (env, job) => {
        console.log(env.data.order_id, env.trace_id);
      },
    },
    { onUnknownUrn: (env, job) => console.warn("no handler for", job.name) },
  ),
  { connection: { host: "localhost", port: 6379 } },
);

A handler is (envelope, job) => unknown | Promise<unknown>. A non-conformant envelope is rejected (BullMQ then retries/fails per its options); an unmapped URN throws unless onUnknownUrn is supplied.

NestJS — `@babelqueue/nestjs`

npm install @babelqueue/nestjs @nestjs/common bullmq

@nestjs/common ^10 || ^11 and bullmq ^5 are peers; it builds on @babelqueue/bullmq.

import { Module } from "@nestjs/common";
import { BabelQueueModule } from "@babelqueue/nestjs";

@Module({
  imports: [
    BabelQueueModule.forRoot({
      queue: "orders",
      connection: { host: "localhost", port: 6379 },
    }),
  ],
})
export class AppModule {}

import { Injectable } from "@nestjs/common";
import { BabelQueuePublisher } from "@babelqueue/nestjs";

@Injectable()
export class Orders {
  constructor(private readonly babelQueue: BabelQueuePublisher) {}

  create() {
    return this.babelQueue.publish("urn:babel:orders:created", { order_id: 1042 });
  }
}

forRoot({ queue, connection, queueOptions? }) provides an injectable BabelQueuePublisher (publish(urn, data, { traceId? }) → Promise<string>) over the BullMQ queue. For consuming, build a plain BullMQ Worker with the processor re-exported from @babelqueue/nestjs:

import { Worker } from "bullmq";
import { processor } from "@babelqueue/nestjs";

new Worker(
  "orders",
  processor({ "urn:babel:orders:created": async (env) => { /* ... */ } }),
  { connection: { host: "localhost", port: 6379 } },
);

Redis — `@babelqueue/redis`

npm install @babelqueue/redis ioredis

ioredis is an optional peer — you provide the client (an ioredis instance satisfies the adapter structurally). It implements the §1 reliable-queue pattern: the list element is the canonical envelope JSON, byte-for-byte, with no wrapping (unlike @babelqueue/bullmq, which uses BullMQ’s own job layout) and no property projection — Redis lists carry no native metadata, so routing and tracing read the body directly. Produce is RPUSH; consume reserves the head into a <queue>:processing list (BRPOPLPUSH, so an in-flight message survives a crash), routes by URN, and LREMs it on success.

Produce

import Redis from "ioredis";
import { RedisPublisher } from "@babelqueue/redis";

const client = new Redis("redis://localhost:6379/0");

const id = await RedisPublisher.create(client, "orders")
  .publish("urn:babel:orders:created", { order_id: 1042 });

publish(urn, data, { traceId? }) returns the message meta.id.

Consume

import { RedisConsumer, type BabelHandlers } from "@babelqueue/redis";

const handlers: BabelHandlers = {
  "urn:babel:orders:created": (envelope, raw) => {
    console.log(envelope.data.order_id, envelope.trace_id);
  },
};

const consumer = new RedisConsumer(client, "orders", handlers, {
  maxTries: 3, // requeue with attempts+1, then <queue>.dlq
  onError: (err) => console.error(err),
});

await consumer.run(() => true); // reserve → process → LREM, until you stop it

A successful handler LREMs the element from <queue>:processing. A throwing handler requeues the envelope with attempts + 1 (at-least-once) up to maxTries, then dead-letters to <queue>.dlq with a dead_letter block — the body owns the attempt count. Unknown-URN strategy is one of fail / delete / release / dead_letter.

This is a Node-owned reliable queue. Full parity with Laravel’s reserved-sorted-set reservation on a shared Redis queue is a separate task — for a mixed PHP+Node fleet, prefer a queue this consumer owns end-to-end. See the Redis binding.

RabbitMQ — `@babelqueue/rabbitmq`

npm install @babelqueue/rabbitmq amqplib

amqplib is an optional peer — you provide the channel (an amqplib Channel satisfies the adapter structurally). It implements §2 of the broker-bindings contract: the envelope JSON is the message body, and the contract fields are projected onto native AMQP 0-9-1 properties so a consumer routes without decoding the body — type = URN, correlation_id = trace_id, message_id = meta.id, app_id = babelqueue, plus the native-typed x-schema-version / x-source-lang / x-attempts headers (AMQP field-tables carry typed values, so integers stay integers). Consume is basic.get + manual ack (at-least-once).

Produce

import amqp from "amqplib";
import { RabbitMQPublisher } from "@babelqueue/rabbitmq";

const conn = await amqp.connect("amqp://guest:guest@localhost:5672/");
const channel = await conn.createChannel();
await channel.assertQueue("orders", { durable: true });

const id = await RabbitMQPublisher.create(channel, "orders")
  .publish("urn:babel:orders:created", { order_id: 1042 });

publish(urn, data, { traceId? }) returns the message meta.id. Messages are persistent (delivery_mode = 2).

Consume

import { RabbitMQConsumer, type BabelHandlers } from "@babelqueue/rabbitmq";

const handlers: BabelHandlers = {
  "urn:babel:orders:created": (envelope, message) => {
    console.log(envelope.data.order_id, envelope.trace_id);
  },
};

const consumer = new RabbitMQConsumer(channel, "orders", handlers, {
  maxTries: 3,
  onError: (err) => console.error(err),
});

await consumer.run(() => true); // basic.get → process → ack, until you stop it

A successful handler acks the message. A throwing handler republishes the envelope with attempts + 1 (at-least-once) up to maxTries, then dead-letters to <queue>.dlq with a dead_letter block. The consumer routes on properties.type (falling back to the body URN). Unknown-URN strategy is one of fail / delete / release / dead_letter. See the RabbitMQ binding.

Amazon SQS — `@babelqueue/sqs`

npm install @babelqueue/sqs @aws-sdk/client-sqs

@aws-sdk/client-sqs is an optional peer — you provide the SQS client (the aggregated SQS class satisfies the transport structurally). It sends the canonical envelope as the MessageBody with the §3 MessageAttributes, and consumes by routing each message to a handler by URN.

Produce

import { SQS } from "@aws-sdk/client-sqs";
import { SqsPublisher } from "@babelqueue/sqs";

const sqs = new SQS({ region: "eu-central-1" });
const url = "https://sqs.eu-central-1.amazonaws.com/123456789012/orders";

const id = await new SqsPublisher(sqs, url).publish("urn:babel:orders:created", { order_id: 1042 });

publish(urn, data, { traceId? }) returns the message meta.id. FIFO queues: new SqsPublisher(sqs, url, { fifo: true }) (the queue URL must end in .fifo).

Consume

import { SqsConsumer } from "@babelqueue/sqs";

const consumer = new SqsConsumer(
  sqs,
  url,
  {
    "urn:babel:orders:created": async (env, message) => {
      console.log(env.data.order_id, env.trace_id);
    },
  },
  { onUnknownUrn: (env, msg) => {}, onError: (err, env, msg) => {} },
);

await consumer.poll(); // receive one batch, route, delete handled; loop this

A throwing handler leaves the message for SQS to redeliver after the visibility timeout (at-least-once); attempts is reconciled to ApproximateReceiveCount − 1. Point the client’s endpoint at LocalStack/ElasticMQ for local testing.

Azure Service Bus — `@babelqueue/azure-service-bus`

npm install @babelqueue/azure-service-bus @azure/service-bus

@azure/service-bus is an optional peer — you provide the sender/receiver (a ServiceBusSender / ServiceBusReceiver satisfies the adapter structurally). It sends the canonical envelope as the message body with the native §4 projection (subject = URN, correlationId = trace_id, messageId = meta.id, plus the bq- application properties), and consumes by routing each message to a handler by URN.

import { ServiceBusClient } from "@azure/service-bus";
import { AsbPublisher, AsbConsumer } from "@babelqueue/azure-service-bus";

const client = new ServiceBusClient(connectionString); // or (namespace, credential)

// produce
const id = await new AsbPublisher(client.createSender("orders"))
  .publish("urn:babel:orders:created", { order_id: 1042 });

// consume (PeekLock)
const consumer = new AsbConsumer(
  client.createReceiver("orders"),
  {
    "urn:babel:orders:created": async (env, message) => {
      console.log(env.data.order_id, env.trace_id);
    },
  },
  { onError: (err) => console.error(err) },
);
await consumer.run();

Delayed delivery: publish(urn, data, { delayMs: 300000 }) → native scheduledEnqueueTimeUtc. A throwing handler abandons the message (the broker redelivers, incrementing deliveryCount); attempts is reconciled to max(body.attempts, deliveryCount − 1). See the Azure Service Bus binding.

Apache Pulsar — `@babelqueue/pulsar`

npm install @babelqueue/pulsar pulsar-client

pulsar-client is an optional peer — you provide the producer/consumer (a Producer / Consumer satisfies the adapter structurally). It sends the canonical envelope as the message payload with the §5 property projection (bq-job = URN, bq-trace-id = trace_id, bq-message-id = meta.id, plus bq-schema-version / bq-source-lang / bq-attempts, all string→string), and consumes by routing each message to a handler by URN.

import Pulsar from "pulsar-client";
import { PulsarPublisher, PulsarConsumer } from "@babelqueue/pulsar";

const client = new Pulsar.Client({ serviceUrl: "pulsar://localhost:6650" });

// produce
const producer = await client.createProducer({ topic: "orders" });
const id = await new PulsarPublisher(producer)
  .publish("urn:babel:orders:created", { order_id: 1042 });

// consume (Shared subscription)
const sub = await client.subscribe({
  topic: "orders",
  subscription: "babelqueue",
  subscriptionType: "Shared",
});
const consumer = new PulsarConsumer(
  sub,
  {
    "urn:babel:orders:created": async (env, message) => {
      console.log(env.data.order_id, env.trace_id);
    },
  },
  { onError: (err) => console.error(err) },
);
await consumer.run();

Delayed delivery: publish(urn, data, { delayMs: 300000 }) → native deliverAfter. A throwing handler negativeAcknowledges the message (the broker redelivers, incrementing getRedeliveryCount()); attempts is reconciled to max(body.attempts, redeliveryCount) — the redelivery count is 0-based, so no −1. See the Apache Pulsar binding.

Apache Kafka — `@babelqueue/kafka`

npm install @babelqueue/kafka kafkajs

kafkajs is an optional peer — you provide the producer/consumer. Kafka has no native ack/delay/DLQ/delivery-counter, so the adapter absorbs all four: the record value is the envelope, the contract fields go to bq- headers (bq-job routes), the record timestamp mirrors meta.created_at, and bq-attempts is the authoritative attempt counter. Consume is process-then-commit (manual commit).

import { Kafka } from "kafkajs";
import { KafkaPublisher, KafkaConsumer, RetryTopics } from "@babelqueue/kafka";

const kafka = new Kafka({ brokers: ["localhost:9092"] });

// produce
const producer = kafka.producer();
await producer.connect();
const id = await KafkaPublisher.create(producer, "orders")
  .publish("urn:babel:orders:created", { order_id: 1042 });

// consume (manual commit)
const consumer = kafka.consumer({ groupId: "orders-workers" });
await consumer.connect();
await consumer.subscribe({ topic: "orders" });

const retry = new RetryTopics("orders", [5_000, 60_000]); // orders.retry.1/.2 + orders.dlq
const babel = new KafkaConsumer(
  consumer,
  { "urn:babel:orders:created": async (env, message) => { console.log(env.data.order_id, env.trace_id); } },
  { producer, retryTopics: retry, maxTries: 3, onError: (err) => console.error(err) },
);
await babel.run();

A throwing handler republishes the envelope to the next <topic>.retry.<n> tier with bq-attempts + 1, then commits; at maxTries it goes to <topic>.dlq with a dead_letter block. A delay with no retry topics throws (Kafka has no native delay). See the Apache Kafka binding.

Apache ActiveMQ Artemis — `@babelqueue/artemis`

npm install @babelqueue/artemis rhea

rhea is an optional peer — you provide the sender/receiver. Artemis speaks AMQP 1.0 (not RabbitMQ’s 0-9-1), with native settlement, scheduled delivery, a delivery counter and a dead-letter address. The envelope is the message body; the contract fields ride the slots a JMS peer reads — x-opt-jms-type = URN (routes), correlation-id = trace_id, creation-time = meta.created_at — plus the string bq_ application properties (underscored, since JMS property names must be valid Java identifiers).

import { Container } from "rhea";
import { ArtemisPublisher, ArtemisConsumer } from "@babelqueue/artemis";

const connection = new Container().connect({ host: "localhost", port: 5672 });

// produce
const sender = connection.open_sender("orders");
const id = await ArtemisPublisher.create(sender, "orders")
  .publish("urn:babel:orders:created", { order_id: 1042 });

// consume (the consumer owns the disposition — autoaccept off)
const receiver = connection.open_receiver({ source: "orders", autoaccept: false, credit_window: 10 });
const dlqSender = connection.open_sender("orders.dlq");

const babel = new ArtemisConsumer(
  { "urn:babel:orders:created": (env, message) => { console.log(env.data.order_id, env.trace_id); } },
  { deadLetterSender: dlqSender, maxTries: 3, onError: (err) => console.error(err) },
);
babel.listen(receiver); // wires receiver.on("message") → accept / release / dead-letter

A successful handler accepts the message; a throwing handler releases it (the broker redelivers and bumps delivery-count); at maxTries the envelope goes to <queue>.dlq with a dead_letter block. attempts reconciles to max(body, delivery-count) — the AMQP counter is 0-based, so no −1. See the Apache ActiveMQ Artemis binding.

Whatever a Node service produces is the canonical envelope, so it is consumed natively by any other BabelQueue SDK — see the wire contract.

Adapters & transports

BullMQ — @babelqueue/bullmq

Produce

Consume

NestJS — @babelqueue/nestjs

Redis — @babelqueue/redis

Produce

Consume

RabbitMQ — @babelqueue/rabbitmq

Produce

Consume

Amazon SQS — @babelqueue/sqs

Produce

Consume

Azure Service Bus — @babelqueue/azure-service-bus

Apache Pulsar — @babelqueue/pulsar

Apache Kafka — @babelqueue/kafka

Apache ActiveMQ Artemis — @babelqueue/artemis

BullMQ — `@babelqueue/bullmq`

NestJS — `@babelqueue/nestjs`

Redis — `@babelqueue/redis`

RabbitMQ — `@babelqueue/rabbitmq`

Amazon SQS — `@babelqueue/sqs`

Azure Service Bus — `@babelqueue/azure-service-bus`

Apache Pulsar — `@babelqueue/pulsar`

Apache Kafka — `@babelqueue/kafka`

Apache ActiveMQ Artemis — `@babelqueue/artemis`