Queue#

Background job processing with priority, delays, retries, and backoff strategies. Built-in drivers for Memory, Redis (ioredis), Bun native Redis, BullMQ, and RabbitMQ.

Built-in Drivers#

Driver	Package	Use Case
Memory	built-in	Default, development
Redis	ioredis	Most common, production
Bun Redis	bun (native)	Zero-dep on Bun
BullMQ	bullmq	Enterprise, rate limiting
RabbitMQ	amqplib	Fan-out, multi-language

Driver Imports#

For production use, import driver classes directly:

typescript

import { Queue } from "@buntok/core/queue";
import {
  RedisQueueDriver,
  BunRedisQueueDriver,
  BullmqQueueDriver,
  RabbitmqQueueDriver,
} from "@buntok/core/queue-drivers";

// Pass driver instance directly
const redisDriver = new RedisQueueDriver({ url: "redis://localhost:6379" });
const queue = new Queue("email", redisDriver);

Basic Usage#

typescript

import { Queue } from "@buntok/core/queue";

// Memory driver (default - development)
const emailQueue = new Queue<{ to: string; subject: string }>("emails");

// Add a job
await emailQueue.add({ to: "user@example.com", subject: "Welcome!" });

// Process jobs
emailQueue.process(async (job) => {
  console.log(`Processing ${job.id}`);
  await sendEmail(job.data.to, job.data.subject);
});

Warning

The name ("emails") is the first argument and is required. Each queue must have a unique name. This name is used for logging, debugging, and driver isolation.

Redis Driver (ioredis)#

Production-ready Redis driver using ioredis. Install ioredis first: bun add ioredis

typescript

import { Queue } from "@buntok/core/queue";
import Redis from "ioredis";

const redis = new Redis();

const emailQueue = new Queue<{ to: string }>("emails", {
  driver: "redis",
  client: redis,
  maxRetries: 3,
  retryDelay: 1000,
  backoff: "exponential",
});

// Or use URL directly
const queue = new Queue<Job>("tasks", {
  driver: "redis",
  url: "redis://localhost:6379",
});

Bun Native Redis Driver#

Uses Bun's built-in Redis client (zero dependencies, requires Bun >= 1.3).

typescript

import { Queue } from "@buntok/core/queue";

const emailQueue = new Queue<{ to: string }>("emails", {
  driver: "bun-redis",
  url: "redis://localhost:6379",
});

BullMQ Driver#

Enterprise Redis-backed queue with rate limiting, job scheduling, and monitoring. Install bullmq first: bun add bullmq

typescript

import { Queue } from "@buntok/core/queue";

const emailQueue = new Queue<{ to: string }>("emails", {
  driver: "bullmq",
  connection: { host: "localhost", port: 6379 },
  defaultJobOptions: {
    attempts: 3,
    backoff: { type: "exponential", delay: 1000 },
    removeOnComplete: { age: 86400 },   // keep completed for 24h
    removeOnFail: { age: 604800 },      // keep failed for 7 days
  },
});

Info

BullMQ provides built-in support for concurrency, rate limiting, repeatable jobs, and a monitoring dashboard. See BullMQ docs for details.

RabbitMQ Driver#

AMQP-based driver for fan-out patterns and multi-language workers. Install amqplib first: bun add amqplib

typescript

import { Queue } from "@buntok/core/queue";

const emailQueue = new Queue<{ to: string }>("emails", {
  driver: "rabbitmq",
  url: "amqp://guest:guest@localhost:5672",
  prefetch: 1,
  maxRetries: 3,
  retryDelay: 1000,
});

Queue API#

Method	Description
queue.add(data, opts?)	Add a job (opts: priority, delay)
queue.process(handler)	Register a handler for jobs
queue.size()	Number of pending jobs
queue.clear()	Remove all pending jobs

Job Object#

Property	Type	Description
job.id	string	Auto-generated UUID
job.data	T	Payload passed by the caller
job.priority	number	Higher = processed sooner (default: 0)
job.delay	number	ms to wait before processing
job.attempt	number	Current attempt (0 = first)
job.createdAt	number	Timestamp when added

Priority & Delay#

typescript

// High priority job (processed first)
await queue.add({ task: "urgent" }, { priority: 10 });

// Delayed job (process after 5 seconds)
await queue.add({ task: "later" }, { delay: 5000 });

// Both
await queue.add({ task: "important-later" }, {
  priority: 5,
  delay: 10_000,
});

Retries & Backoff#

typescript

// Retry failed jobs up to 3 times
const queue = new Queue("tasks", {
  maxRetries: 3,
  retryDelay: 1000,      // 1 second between retries
  backoff: "fixed",       // or "exponential"
});

// Exponential backoff: 1s, 2s, 4s, 8s...
const queue = new Queue("tasks", {
  maxRetries: 4,
  retryDelay: 1000,
  backoff: "exponential",
});

Info

Jobs that exceed maxRetries are permanently failed and logged to console. The queue continues processing remaining jobs.

Full Example#

typescript

import { Buntok } from "@buntok/core";
import { Queue } from "@buntok/core/queue";

const emailQueue = new Queue<{
  to: string;
  subject: string;
  body: string;
}>("emails", { maxRetries: 3, backoff: "exponential" });

// Process emails in background
emailQueue.process(async (job) => {
  console.log(`[${job.attempt + 1}] Sending to ${job.data.to}`);
  await transporter.sendMail({
    to: job.data.to,
    subject: job.data.subject,
    html: job.data.body,
  });
});

// HTTP endpoint adds jobs
app.post("/send-email", async (ctx) => {
  const { to, subject, body } = await ctx.body();
  await emailQueue.add({ to, subject, body });
  return ctx.json({ queued: true });
});

app.listen(1212);

Custom Driver#

Implement the QueueDriver interface for custom backends:

typescript

import type { QueueDriver, Job, JobHandler } from "@buntok/core/queue";

class MyCustomDriver implements QueueDriver<{ to: string }> {
  async add(data: { to: string }, opts?: { priority?: number; delay?: number }): Promise<void> {
    // Your implementation
  }
  process(handler: JobHandler<{ to: string }>): void {
    // Your implementation
  }
  size(): number { return 0; }
  clear(): void {}
}

const queue = new Queue("email", new MyCustomDriver());

PreviousTemplate Engine Next Scheduler