BullMQ

Swap the in-memory publish queue for a Redis-backed BullMQ queue, so log and span events survive restarts and retry on transient failures.

The pino, winston, console-capture, and OpenTelemetry integrations buffer their events in memory by default. BullMQPublisher replaces that buffer with a BullMQ queue on Redis: submit() becomes a Redis write, a separate worker process does the publishing, and nothing is lost when the producer restarts.

This is the TypeScript counterpart of the Python SDK’s RqPublisher. Tested against bullmq@^5, an optional peer dependency.

Install

npm install @axonpush/sdk bullmq

You also need a Redis instance both the producer and the worker can reach.

Producer

Set mode: "bullmq" and pass bullmqOptions to any integration that takes an IntegrationConfig:

import IORedis from "ioredis";
import pino from "pino";
import { AxonPush } from "@axonpush/sdk";
import { createAxonPushPinoStream } from "@axonpush/sdk/integrations/pino";

const connection = new IORedis(process.env.REDIS_URL!, { maxRetriesPerRequest: null });
const client = new AxonPush();

const stream = createAxonPushPinoStream({
  client,
  channelId: process.env.AXONPUSH_CHANNEL_ID!,
  serviceName: "my-api",
  mode: "bullmq",
  bullmqOptions: { connection, queueName: "axonpush" },
});

const log = pino({ level: "info" }, stream);

mode: "bullmq" without bullmqOptions throws at construction rather than silently falling back, and bullmqOptions without a connection throws too. bullmq itself is imported lazily on the first enqueue, so a misconfigured install surfaces as a clear “install bullmq” error.

Worker

Nothing reaches axonpush until a worker drains the queue. Run this in a separate process:

import IORedis from "ioredis";
import { AxonPush, createBullMQWorker } from "@axonpush/sdk";

const connection = new IORedis(process.env.REDIS_URL!, { maxRetriesPerRequest: null });
const client = new AxonPush();

const worker = await createBullMQWorker({ client, connection, queueName: "axonpush" });

process.on("SIGTERM", () => void worker.close());

The worker pulls each job and calls client.events.publish(job.data). The job payload is the full publish params object the integration built, so the worker needs no knowledge of which integration produced it, one worker drains every queue-mode integration in your fleet.

The publishing client lives in the worker

BullMQPublisher takes an AxonPush instance for interface symmetry with the in-memory publisher, but never publishes with it, it only enqueues. The credentials that matter are the worker’s. The producer needs Redis; the worker needs Redis and a valid axonpush API key.

Options

Option	Type	Default	Description
`connection`	IORedis instance or options	required	Passed through to BullMQ.
`queueName`	`string`	`"axonpush"`	Must match the worker’s.
`jobOptions`	`Record<string, unknown>`	see below	Merged over the publisher defaults.

Default job options: attempts: 3, removeOnComplete: true, and removeOnFail: { age: 86400 }, failed jobs are kept for 24 hours so you can inspect them, successful ones are dropped immediately. Anything you put in jobOptions overrides these; every BullMQ job option is available.

Direct use

You can also drive the publisher yourself, outside any integration:

import { BullMQPublisher } from "@axonpush/sdk";

const publisher = new BullMQPublisher(client, { connection });
publisher.submit({ identifier: "job.done", channelId, payload: { ok: true } });

await publisher.flush(2000);
await publisher.close();

It mirrors the in-memory publisher’s surface, submit, flush, close, droppedCount, and Symbol.asyncDispose, so the two are interchangeable behind a common holder.

Delivery semantics

submit() is still fire-and-forget on the caller’s path: it starts an async enqueue and returns. flush(timeoutMs?) waits only for those in-flight enqueues to reach Redis, once a job is in the queue it is durable, so there is nothing further to drain locally. close() flushes, then closes the BullMQ queue; submissions after close are counted in droppedCount.

An enqueue that fails, Redis down, connection refused, is logged at warning level and swallowed, matching the SDK’s fail-open posture. Publishing itself is retried by BullMQ according to attempts.