---
title: "Adapters & transports"
description: "The npm packages beyond core: the BullMQ and NestJS framework adapters and the Amazon SQS transport — canonical-envelope jobs/messages, URN routing, and an injectable publisher."
source: https://babelqueue.com/docs/babelqueue-node/1.x/adapters/
updated: 2026-06-14T00:00:00.000Z
---

# 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`

```bash
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

```ts
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

```ts
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`

```bash
npm install @babelqueue/nestjs @nestjs/common bullmq
```

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

Register the module and inject the publisher:

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

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

```ts
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`:

```ts
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`

```bash
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](/docs/spec/1.x/broker-bindings/#redis):
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 `LREM`s it on success.

### Produce

```ts
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

```ts
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 `LREM`s 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](/docs/spec/1.x/broker-bindings/#redis).

## RabbitMQ — `@babelqueue/rabbitmq`

```bash
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](/docs/spec/1.x/broker-bindings/#rabbitmq-amqp-0-9-1):
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

```ts
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

```ts
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 `ack`s 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](/docs/spec/1.x/broker-bindings/#rabbitmq-amqp-0-9-1).

## Amazon SQS — `@babelqueue/sqs`

```bash
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`](/docs/spec/1.x/broker-bindings/#amazon-sqs),
and consumes by routing each message to a handler by URN.

### Produce

```ts
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

```ts
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`

```bash
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.

```ts
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 `abandon`s the message (the broker redelivers,
incrementing `deliveryCount`); `attempts` is reconciled to
`max(body.attempts, deliveryCount − 1)`. See the
[Azure Service Bus binding](/docs/spec/1.x/broker-bindings/#azure-service-bus).

## Apache Pulsar — `@babelqueue/pulsar`

```bash
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.

```ts
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 `negativeAcknowledge`s 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](/docs/spec/1.x/broker-bindings/#apache-pulsar).

## Apache Kafka — `@babelqueue/kafka`

```bash
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).

```ts
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](/docs/spec/1.x/broker-bindings/#apache-kafka).

## Apache ActiveMQ Artemis — `@babelqueue/artemis`

```bash
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).

```ts
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 `accept`s the message; a throwing handler `release`s 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](/docs/spec/1.x/broker-bindings/#apache-activemq-artemis).

Whatever a Node service produces is the canonical envelope, so it is consumed
natively by any other BabelQueue SDK — see the [wire contract](/docs/spec/1.x/envelope/).