guide · queue

The producer side.

A Queue is where work goes in. Same object whether it writes straight to a SQLite file in your process or talks to a shared server over TCP, so the code you write 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'); // connects to localhost:6789

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

import { Queue } from 'bunqueue-client';

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

// Remote server: connection options are top-level
const remoteQueue = new Queue('tasks', {
  host: '192.168.1.100',
  port: 6789,
  token: 'secret-token',  // If AUTH_TOKENS is set on the server
  poolSize: 4,            // Producer-side connection pool
});

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
  )

embedded: true and defaultJobOptions exist in the Bun bunqueue package only; the poolSize connection pool is TypeScript-only. In the other SDKs, pass connection options to the constructor (see SDKs) and job options per add call.

Where to go next

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 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