guide · queue

The producer side.

A Queue is where work goes in. It is the same object whether it writes to SQLite in your process or talks over TCP to a memory-, SQLite-, or PostgreSQL-backed server, so client code does not change when the deployment does.

Start here: create the queue, pick a mode, and know which options belong to the constructor rather than to individual jobs.

Create a queue

Bun

import { Queue } from 'bunqueue/client';

const queue = new Queue('my-queue', { embedded: true });

await queue.add('job-name', { key: 'value' });

Node.js / Deno

import { Queue } from 'bunqueue-client';

const queue = new Queue('my-queue', { embedded: false });

await queue.add('job-name', { key: 'value' });

Python

from bunqueue import Queue

queue = Queue("my-queue")  # connects to localhost:6789

queue.add("job-name", {"key": "value"})

PHP

use Bunqueue\Queue;

$queue = new Queue('my-queue'); // connects to localhost:6789

$queue->add('job-name', ['key' => 'value']);

Go

queue := bunqueue.NewQueue("my-queue", bunqueue.Options{}) // localhost:6789
defer queue.Close()

queue.Add("job-name", map[string]any{"key": "value"}, nil)

Rust

use bunqueue_client::{ConnectionOptions, JobOptions, Queue, Value};

let queue = Queue::new("my-queue", ConnectionOptions::default()); // localhost:6789
let data = Value::Map(vec![(Value::from("key"), Value::from("value"))]);
queue.add("job-name", data, JobOptions::default())?;

Elixir

queue = Bunqueue.queue("my-queue")  # connects to localhost:6789

{:ok, _job} = Bunqueue.Queue.add(queue, "job-name", %{key: "value"})

Embedded vs TCP

embedded: true runs the queue inside your process. Without it, the Queue connects to a bunqueue server on localhost:6789 (see Server Mode). The Queue and its Worker must use the same mode.

Useful variations:

Bun

// Typed queue: job.data is type-checked
interface TaskData {
  userId: number;
  action: string;
}
const typedQueue = new Queue<TaskData>('tasks', { embedded: true });

// Default options applied to every job
const emailQueue = new Queue('emails', {
  embedded: true,
  defaultJobOptions: {
    attempts: 3,
    backoff: 1000,
    removeOnComplete: true,
  },
});

// TCP mode with a custom connection
const remoteQueue = new Queue('tasks', {
  connection: {
    host: '192.168.1.100',
    port: 6789,
    token: 'secret-token', // If AUTH_TOKENS is set on the server
    poolSize: 4, // Connection pool size
  },
});

Node.js / Deno

// Typed queue: job.data is type-checked
interface TaskData {
  userId: number;
  action: string;
}
const typedQueue = new Queue<TaskData>('tasks', { embedded: false });

// Default options applied to every job
const emailQueue = new Queue('emails', {
  embedded: false,
  defaultJobOptions: {
    attempts: 3,
    backoff: 1000,
    removeOnComplete: true,
  },
});

// TCP mode with a custom connection
const remoteQueue = new Queue('tasks', {
  connection: {
    host: '192.168.1.100',
    port: 6789,
    token: 'secret-token', // If AUTH_TOKENS is set on the server
    poolSize: 4, // Connection pool size
  },
});

Python

from bunqueue import Queue

remote_queue = Queue(
    "tasks",
    host="192.168.1.100",
    port=6789,
    token="secret-token",  # If AUTH_TOKENS is set on the server
)

# External SDKs take job options on each add.
remote_queue.add("send", {"user_id": 42}, attempts=3, remove_on_complete=True)

PHP

use Bunqueue\Queue;

$remoteQueue = new Queue('tasks', [
    'host' => '192.168.1.100',
    'port' => 6789,
    'token' => 'secret-token', // If AUTH_TOKENS is set on the server
]);

$remoteQueue->add('send', ['userId' => 42], [
    'attempts' => 3,
    'removeOnComplete' => true,
]);

Go

remoteQueue := bunqueue.NewQueue("tasks", bunqueue.Options{
    Host:  "192.168.1.100",
    Port:  6789,
    Token: "secret-token", // If AUTH_TOKENS is set on the server
})
defer remoteQueue.Close()

remoteQueue.Add("send", map[string]any{"userId": 42}, bunqueue.JobOptions{
    "attempts": 3, "removeOnComplete": true,
})

Rust

use bunqueue_client::{ConnectionOptions, JobOptions, Queue, Value};

let remote_queue = Queue::new("tasks", ConnectionOptions {
    host: "192.168.1.100".into(),
    port: 6789,
    token: Some("secret-token".into()), // If AUTH_TOKENS is set on the server
    ..Default::default()
});

remote_queue.add("send", Value::Nil, JobOptions {
    attempts: Some(3),
    remove_on_complete: Some(true),
    ..Default::default()
})?;

Elixir

remote_queue =
  Bunqueue.queue("tasks",
    host: "192.168.1.100",
    port: 6789,
    token: "secret-token" # If AUTH_TOKENS is set on the server
  )

{:ok, _job} =
  Bunqueue.Queue.add(remote_queue, "send", %{user_id: 42},
    attempts: 3,
    removeOnComplete: true
  )

The two TypeScript packages share the same Queue API, including defaultJobOptions, prefixKey, auto-batching, and nested connection.poolSize. Use embedded: false on Node.js and Deno; embedded storage requires Bun. In the other SDKs, pass connection options to the constructor (see SDKs) and job options per add call.

Where to go next

Guide	What it covers
Adding Jobs	add, addBulk, priorities, delays, durability
Deduplication and Idempotent Job Adds	Idempotent adds, dedup keys, custom job ids
Querying Jobs	Fetch jobs, states, counts and results
Queue Control and Maintenance	Pause, drain, obliterate, clean and repair
Progress, Job Logs and Dependencies	Progress, per-job logs and dependencies
Queue Rate Limiting and Global Concurrency	Rate limits and global concurrency caps
Job Groups	Per-group priority/FIFO, fairness and capacity
Job Schedulers from the Queue	Named repeatable schedules from the queue
DLQ Operations from the Queue Object	Failed-job operations from the Queue object
Workers, Stats and Metrics from the Queue	Registered workers, stats and metrics windows
Namespaces, Auto-Batching and Store-and-Forward	Namespaces, auto-batching, store-and-forward
JobOptions Reference	Every JobOptions field, with defaults