bunmq

bunMQ

A BullMQ alternative using Bun's Worker API instead of Redis. bunMQ provides a simple, fast, and lightweight job queue system for background processing.

Features

• 🚀 Fast: Built on Bun's optimized Worker API
• 🔄 Retry Logic: Configurable retry attempts with exponential backoff
• ⏰ Delayed Jobs: Schedule jobs to run in the future
• 📊 Job Monitoring: Track job status, completion, and failures
• 🧵 Multi-threaded: Process multiple jobs concurrently
• 💾 Memory Efficient: Automatic cleanup of old jobs
• 🛡️ Error Handling: Robust error handling and recovery

Installation

bun add bunmq

Quick Start

import { Queue } from 'bunmq';

// Create a queue with 2 workers
const queue = new Queue(2);

// Define a function to process
function sendEmail(to: string, subject: string, body: string) {
  console.log(`Sending email to ${to}: ${subject}`);
  return Promise.resolve({ success: true });
}

// Enqueue a job
const jobId = queue.enqueue(
  sendEmail,
  { 
    delay: 1000,        // Wait 1 second before processing
    attempts: 3,         // Retry up to 3 times
    backoff: { 
      type: 'exponential', 
      delay: 1000 
    }
  },
  '[e-mail]',   // to parameter
  'Welcome!',           // subject parameter
  'Welcome message'      // body parameter
);

console.log('Job enqueued:', jobId);

API Reference

Queue

Constructor

new Queue(workerCount?: number)

• workerCount: Number of worker threads (default: 1)

Methods

enqueue<T>(fn: T, options?: JobOptions, ...params: Parameters<T>): string

Enqueue a function to be executed in a worker thread.

Parameters:

• fn: Function to execute
• options: Job configuration options
• ...params: Parameters to pass to the function

Returns: Job ID string

getJob(jobId: string): JobData | undefined

Get job information by ID.

getJobs(status?: JobStatus): JobData[]

Get all jobs, optionally filtered by status.

getStats(): QueueStats

Get queue statistics.

close(): Promise<void>

Gracefully close the queue and terminate all workers.

Event Handlers

onJobStart(handler: (job: JobData) => void): void

Called when a job starts processing.

onJobComplete(handler: (job: JobData, result: any) => void): void

Called when a job completes successfully.

onJobError(handler: (job: JobData, error: string) => void): void

Called when a job encounters an error.

onJobRetry(handler: (job: JobData, attempt: number) => void): void

Called when a job is being retried.

onJobFailed(handler: (job: JobData, error: string) => void): void

Called when a job fails after all retry attempts.

onQueueEmpty(handler: () => void): void

Called when the queue becomes empty (no waiting or active jobs).

onQueueError(handler: (error: Error) => void): void

Called when a queue-level error occurs.

setEvents(events: QueueEvents): void

Set multiple event handlers at once.

JobOptions

interface JobOptions {
  delay?: number;                    // Delay before processing (ms)
  attempts?: number;                 // Number of retry attempts
  backoff?: {                       // Retry backoff strategy
    type: 'fixed' | 'exponential';
    delay: number;
  };
  removeOnComplete?: number;        // Keep N completed jobs
  removeOnFail?: number;            // Keep N failed jobs
  priority?: number;                  // Job priority (higher = more priority)
}

JobData

interface JobData {
  id: string;                        // Unique job ID
  functionName: string;              // Function name
  params: any[];                     // Function parameters
  options: JobOptions;              // Job options
  createdAt: number;                // Creation timestamp
  attempts: number;                 // Current attempt number
  status: 'waiting' | 'active' | 'completed' | 'failed' | 'delayed';
}

Examples

Basic Usage

import { Queue } from 'bunmq';

const queue = new Queue();

function processData(data: string) {
  console.log('Processing:', data);
  return `Processed: ${data}`;
}

// Enqueue a job
const jobId = queue.enqueue(processData, {}, 'Hello World');

With Retry Logic

function unreliableTask() {
  if (Math.random() < 0.5) {
    throw new Error('Random failure');
  }
  return 'Success!';
}

const jobId = queue.enqueue(
  unreliableTask,
  {
    attempts: 5,
    backoff: { type: 'exponential', delay: 1000 }
  }
);

Delayed Jobs

function sendReminder() {
  console.log('Sending reminder email');
}

// Send reminder in 1 hour
queue.enqueue(
  sendReminder,
  { delay: 60 * 60 * 1000 }
);

Monitoring Jobs

// Get queue statistics
const stats = queue.getStats();
console.log('Queue stats:', stats);
// { waiting: 2, active: 1, completed: 5, failed: 0, total: 8 }

// Get specific job
const job = queue.getJob('job_123');
console.log('Job status:', job?.status);

// Get all completed jobs
const completed = queue.getCompletedJobs();
console.log('Completed jobs:', completed.length);

Lifecycle Events

const queue = new Queue(2);

// Set up event handlers
queue.onJobStart((job) => {
  console.log(`Job started: ${job.id}`);
});

queue.onJobComplete((job, result) => {
  console.log(`Job completed: ${job.id} with result:`, result);
});

queue.onJobError((job, error) => {
  console.log(`Job error: ${job.id} - ${error}`);
});

queue.onJobRetry((job, attempt) => {
  console.log(`Job retry: ${job.id} (attempt ${attempt})`);
});

queue.onJobFailed((job, error) => {
  console.log(`Job failed: ${job.id} after all retries`);
});

queue.onQueueEmpty(() => {
  console.log('All jobs processed!');
});

// Or set multiple events at once
queue.setEvents({
  onJobStart: (job) => console.log(`Started: ${job.id}`),
  onJobComplete: (job, result) => console.log(`Completed: ${job.id}`),
  onQueueEmpty: () => console.log('Queue empty!')
});

Performance

bunMQ leverages Bun's optimized Worker API and includes several performance optimizations:

• Fast Message Passing: Optimized postMessage with fast paths for strings and simple objects
• Concurrent Processing: Multiple workers process jobs in parallel
• Memory Efficient: Automatic cleanup of old jobs
• Low Latency: Direct worker communication without external dependencies

Comparison with BullMQ

Feature	bunMQ	BullMQ
Dependencies	None (Bun only)	Redis required
Setup	Instant	Redis server needed
Performance	Very fast	Fast (with Redis)
Persistence	In-memory	Redis-backed
Clustering	Single process	Multi-process with Redis

Running the Example

# Run the example
bun run example.ts

License

MIT