Examples

Produce and consume end to end for every provider. Each snippet follows the same shape: create the queue, publish, consume with an explicit acknowledgement, close cleanly. All snippets run against the local broker stack in the repository (docker compose up -d).

BullMQ

Job queue

npm install @mohamedhabibwork/queuekit bullmq ioredis

BullMQ no longer bundles a Redis client — when connection is a URL string, also install ioredis.

Produce

import { createQueue } from '@mohamedhabibwork/queuekit';

const jobs = await createQueue({
  type: 'bullmq',
  connection: 'redis://localhost:6379',
});

const result = await jobs.publish('emails', {
  type: 'welcome',
  payload: { userId: 'u_1' },
}, {
  delay: 5_000,
  priority: 2,
  native: { attempts: 5, backoff: { type: 'exponential', delay: 1_000 } },
});

console.log(result.messageId); // BullMQ job id
await jobs.close();

Consume

const consumer = await jobs.consume<{ userId: string }>('emails', async ({ message }) => {
  await sendWelcome(message.payload.userId);
  // returning normally completes the job; throwing fails it
}, { concurrency: 4 });

await consumer.close();

Ack: returning from the worker completes the job — the normalized acknowledgement is a no-op.

Kafka

Event stream

npm install @mohamedhabibwork/queuekit kafkajs

Produce

import { createQueue } from '@mohamedhabibwork/queuekit';

const kafka = await createQueue({
  type: 'kafka',
  clientId: 'orders-api',
  brokers: ['localhost:9092'],
});

const result = await kafka.publish('orders.created', {
  type: 'order.placed',
  payload: { orderId: 'ord_123' },
  traceId: 'trace_abc',
}, {
  native: { partition: 2, headers: { source: 'api' } },
});

console.log(result.partition, result.offset); // kafkajs RecordMetadata stays on result.native
await kafka.close();

Consume

const consumer = await kafka.consume<{ orderId: string }>('orders.created', async ({ message, ack }) => {
  await fulfil(message.payload.orderId);
  await ack.complete(); // commits the next offset for this message
}, {
  native: { groupId: 'fulfilment-workers', fromBeginning: true },
});

await consumer.close();

native.groupId is required. Set autoAck: true to commit after every successful handler instead of calling ack.complete() yourself.

RabbitMQ

Message queue

npm install @mohamedhabibwork/queuekit amqplib

Produce

import { createQueue } from '@mohamedhabibwork/queuekit';

const rabbit = await createQueue({ type: 'rabbitmq', url: 'amqp://localhost' });

await rabbit.publish('emails', {
  payload: { to: 'person@example.com' },
  correlationId: 'corr_1',
}, {
  priority: 5,
  ttl: 60_000,
  native: { persistent: true },
});

await rabbit.close();

Consume

const consumer = await rabbit.consume<{ to: string }>('emails', async ({ message, ack }) => {
  const delivered = await send(message.payload.to);
  if (delivered) await ack.complete();
  else await ack.retry?.();          // nack with requeue — the broker redelivers
}, { native: { prefetch: 10 } });

await consumer.close();

Ack: complete, retry, and reject map to ack, nack, and reject.

Redis Streams

Stream

npm install @mohamedhabibwork/queuekit redis   # or: ioredis

Choose the SDK with client: 'redis' (default) or client: 'ioredis'. Both work against Redis and Valkey servers — the protocol is identical.

Produce

import { createQueue } from '@mohamedhabibwork/queuekit';

const stream = await createQueue({
  type: 'redis',
  mode: 'streams',
  url: 'redis://localhost:6379',
  group: 'billing',
  consumer: 'worker-a',
  client: 'ioredis', // or 'redis' (default)
});

const result = await stream.publish('payments', {
  payload: { paymentId: 'pay_1' },
});
console.log(result.messageId); // the XADD entry id, e.g. "1730000000000-0"

await stream.close();

Consume, retry, and dead-letter

const consumer = await stream.consume<{ paymentId: string }>('payments', async ({ message, ack }) => {
  const ok = await capture(message.payload.paymentId);
  if (ok) await ack.complete();      // XACK — the entry leaves the pending list
  else await ack.retry?.();          // XADD a new entry with attempt + 1, then XACK the old one
}, {
  native: {
    deadLetter: 'payments:dead',     // reject() without requeue lands here
    blockMs: 1_000,
    count: 10,
  },
});

await consumer.close();

On ack.reject() without requeue, the entry is first XADDed to the dead-letter stream — keeping its payload, attempt, and a deadLetterOf: { stream, id, group } trace of where it came from — and only then acknowledged. The dead-letter stream is itself a stream, so it can be consumed, replayed, or re-driven with the same driver.

The consumer group is created at id 0, so entries published before the first consumer attaches are still delivered — a new group always starts with the stream's backlog.

Redis Pub/Sub

Pub/Sub

import { createQueue } from '@mohamedhabibwork/queuekit';

const pubsub = await createQueue({ type: 'redis', mode: 'pubsub', url: 'redis://localhost:6379' });

// Core pub/sub only reaches live subscribers — subscribe before publishing.
const consumer = await pubsub.consume<{ temperature: number }>('sensor.readings', ({ message }) => {
  console.log(message.payload.temperature);
});

const result = await pubsub.publish('sensor.readings', { payload: { temperature: 21 } });
console.log(result.native); // 1 — the number of subscribers that received the fan-out

await consumer.close();
await pubsub.close();

Ack: pub/sub is fire-and-forget — acknowledgements are no-ops. Use Streams mode when you need persistence, groups, or retries.

NATS

Pub/Sub

npm install @mohamedhabibwork/queuekit nats

import { createQueue } from '@mohamedhabibwork/queuekit';

const nats = await createQueue({ type: 'nats', servers: ['nats://localhost:4222'], mode: 'core' });

// Plain subscribe: every subscriber sees every message.
const all = await nats.consume<{ userId: string }>('user.signed.up', ({ message }) => {
  audit(message.payload.userId);
});

// Queue group: each message goes to exactly one subscriber — cheap horizontal scaling.
const worker = await nats.consume<{ userId: string }>('user.signed.up', ({ message }) => {
  sendEmail(message.payload.userId);
}, { native: { queue: 'mailer-workers' } });

await nats.publish('user.signed.up', { payload: { userId: 'u_1' } });

await all.close();
await worker.close();
await nats.close();

Ack: core NATS messages are ephemeral and acknowledgements are no-ops. JetStream publishing is supported via mode: 'jetstream'; for JetStream pull consumers use native() directly in v0.2.

Amazon SQS

Message queue

npm install @mohamedhabibwork/queuekit @aws-sdk/client-sqs

Produce

import { createQueue } from '@mohamedhabibwork/queuekit';

const sqs = await createQueue({ type: 'sqs', region: 'eu-central-1' });

// The destination is the queue URL (or set config.queueUrl once).
await sqs.publish(process.env.QUEUE_URL!, {
  payload: { taskId: 'task-1' },
}, {
  delay: 10,
  idempotencyKey: 'task-1', // → MessageDeduplicationId on FIFO queues
  native: { MessageGroupId: 'jobs' },
});

await sqs.close();

Consume

const consumer = await sqs.consume<{ taskId: string }>(process.env.QUEUE_URL!, async ({ message, ack }) => {
  const done = await runTask(message.payload.taskId);
  if (done) await ack.complete();               // deletes the message
  else await ack.retry?.({ delay: 30_000 });    // visibility timeout back to 30s
}, {
  native: { MaxNumberOfMessages: 5, WaitTimeSeconds: 10 },
});

await consumer.close();

Multi-provider manager

Composition

import { createQueueManager } from '@mohamedhabibwork/queuekit';

const queues = createQueueManager({
  default: 'jobs',
  providers: {
    jobs: { type: 'bullmq', connection: 'redis://localhost:6379' },
    events: { type: 'kafka', clientId: 'api', brokers: ['localhost:9092'] },
  },
});

// Providers are created lazily — the first .provider() call loads that SDK.
await (await queues.provider('jobs')).publish('emails', { type: 'welcome', payload: { userId: 'u_1' } });
await (await queues.default()).health();
await queues.warmup();  // initialize everything, e.g. at boot
await queues.close();   // closes all initialized consumers and clients

Typed registry

Type safety

import { createTypedQueue } from '@mohamedhabibwork/queuekit';

type Events = {
  'user.signedUp': { userId: string };
  'order.placed': { orderId: string; total: number };
};

const events = createTypedQueue<Events>(kafka);

await events.publish('user.signedUp', { userId: 'u_1' }); // payload type checked per topic
await events.publish('order.placed', { orderId: 'ord_1', total: 99 });

Middleware

Cross-cutting

const kafka = await createQueue({ type: 'kafka', clientId: 'api', brokers: ['localhost:9092'] });

kafka.use(async (context, next) => {
  const started = Date.now();
  await next();
  console.log(`${context.provider} ${context.operation} ${context.destination} in ${Date.now() - started}ms`);
});

Middleware wraps every publish and publishMany with the operation context: provider, operation, destination, trace, and correlation ids.

Testing with the fake driver

Testing

import { createFakeQueue } from '@mohamedhabibwork/queuekit/testing';

const testQueue = createFakeQueue();

// autoAck defaults to true on the fake so simple assertions need no ack dance.
const received: string[] = [];
await testQueue.consume<{ email: string }>('emails', ({ message }) => {
  received.push(message.payload.email);
});

// Messages published before a consumer attaches are queued and delivered FIFO.
await testQueue.publish('emails', { type: 'welcome', payload: { email: 'person@example.com' } });
await testQueue.waitUntilIdle();                    // await all in-flight handler work

expect(received).toEqual(['person@example.com']);
expect(testQueue.pending('emails')).toHaveLength(0);

Deterministic controls

await testQueue.publish('emails', { payload: {} }, { delay: 60_000 });
await testQueue.waitUntilIdle();
testQueue.pending('emails');            // still queued — the delay has not elapsed
await testQueue.flush();                // force delayed messages out immediately

testQueue.pause(); testQueue.resume();  // hold and release delivery
testQueue.acknowledged('emails');       // settled messages
testQueue.deadLetters('emails');        // rejected / exhausted messages
testQueue.failNext(new Error('broker down')); // inject the next publish failure

The fake is a first-class driver, so createQueue({ type: 'memory' }) and createQueueManager({ providers: { jobs: { type: 'memory' } } }) work with the same config shape used in production. Handlers that throw are retried up to setMaxAttempts() (default 3) and then dead-lettered.

Every produce → consume → acknowledge path shown here runs against real brokers in the repository's end-to-end suite — see tests/e2e and docker compose up -d.