guide · queue-group

Many queues, one namespace.

QueueGroup prefixes a set of queues with a shared name, so “invoices” inside the “billing” group becomes “billing:invoices”. Handy for multi-tenant apps and keeping domains apart.

A QueueGroup is a thin organizer: it creates normal Queue and Worker instances whose names carry the group prefix, and it can pause, resume, or clear all of them at once.

Availability

QueueGroup ships in the Bun package (bunqueue/client). From the other SDKs, create the queues individually with the prefixed name (for example new Queue('billing:invoices')): the prefix is just part of the queue name on the server, so grouped and non-grouped clients interoperate on the same queues.

Quick Start

Bun

import { QueueGroup } from 'bunqueue/client';

const billing = new QueueGroup('billing');

// Queues are automatically prefixed
const invoices = billing.getQueue('invoices', { embedded: true });   // "billing:invoices"
const payments = billing.getQueue('payments', { embedded: true });   // "billing:payments"

await invoices.add('create', { amount: 100 });
await payments.add('process', { orderId: '123' });

// Workers use the same prefixed names
const invoiceWorker = billing.getWorker('invoices', async (job) => {
  console.log('Processing invoice:', job.data);
  return { processed: true };
}, { embedded: true });

Node.js / Deno

import { QueueGroup } from 'bunqueue-client';

const billing = new QueueGroup('billing');

// Queues are automatically prefixed
const invoices = billing.getQueue('invoices', { embedded: false });   // "billing:invoices"
const payments = billing.getQueue('payments', { embedded: false });   // "billing:payments"

await invoices.add('create', { amount: 100 });
await payments.add('process', { orderId: '123' });

// Workers use the same prefixed names
const invoiceWorker = billing.getWorker('invoices', async (job) => {
  console.log('Processing invoice:', job.data);
  return { processed: true };
}, { embedded: false });

Python

from bunqueue import Queue, Worker

invoices = Queue("billing:invoices")
payments = Queue("billing:payments")

invoices.add("create", {"amount": 100})
payments.add("process", {"order_id": "123"})

invoice_worker = Worker(
    "billing:invoices",
    lambda job: {"processed": True, "invoice": job.data},
)
invoice_worker.run()  # blocking loop

PHP

use Bunqueue\Queue;
use Bunqueue\Worker;

$invoices = new Queue('billing:invoices');
$payments = new Queue('billing:payments');

$invoices->add('create', ['amount' => 100]);
$payments->add('process', ['orderId' => '123']);

$invoiceWorker = new Worker('billing:invoices',
    fn (Bunqueue\Job $job) => ['processed' => true, 'invoice' => $job->data()]
);
$invoiceWorker->run(); // blocking loop

Go

invoices := bunqueue.NewQueue("billing:invoices", bunqueue.Options{})
payments := bunqueue.NewQueue("billing:payments", bunqueue.Options{})

invoices.Add("create", map[string]any{"amount": 100}, nil)
payments.Add("process", map[string]any{"orderId": "123"}, nil)

invoiceWorker := bunqueue.NewWorker("billing:invoices", processor,
    bunqueue.WorkerOptions{})
invoiceWorker.Run() // blocking loop

Rust

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

let options = ConnectionOptions::default();
let invoices = Queue::new("billing:invoices", options.clone());
let payments = Queue::new("billing:payments", options);

invoices.add("create", Value::from(100), JobOptions::default())?;
payments.add("process", Value::from("123"), JobOptions::default())?;

let invoice_worker = Worker::new(
    "billing:invoices", processor, WorkerOptions::default(),
);
invoice_worker.run()?; // blocking loop

Elixir

invoices = Bunqueue.queue("billing:invoices")
payments = Bunqueue.queue("billing:payments")

{:ok, _job} = Bunqueue.Queue.add(invoices, "create", %{amount: 100})
{:ok, _job} = Bunqueue.Queue.add(payments, "process", %{order_id: "123"})

invoice_worker =
  Bunqueue.Worker.new("billing:invoices", fn job ->
    {:ok, %{processed: true, invoice: job.data}}
  end)
Bunqueue.Worker.run(invoice_worker) # blocking loop

Both TypeScript packages expose QueueGroup; getQueue and getWorker accept the same options as Queue and Worker. Node.js and Deno use TCP. Use the Async group methods to act on registered remote queues. The other SDK examples create normal queues and workers with the same prefixed names.

Common Tasks

Operate on the whole group

Bun

billing.listQueues();      // ['invoices', 'payments'] (names without prefix)
billing.pauseAll();        // pause every queue in the group
billing.resumeAll();       // resume them
billing.drainAll();        // remove all waiting jobs
billing.obliterateAll();   // remove ALL data from every queue

// Awaitable forms are authoritative in embedded and TCP modes
await billing.pauseAllAsync();
await billing.resumeAllAsync();
const removed = await billing.drainAllAsync();
await billing.obliterateAllAsync();

Node.js / Deno

const names = await billing.listQueuesAsync(); // registered remote queue names
await billing.pauseAllAsync();
await billing.resumeAllAsync();
const removed = await billing.drainAllAsync();
await billing.obliterateAllAsync();

Python

queues = [invoices, payments]
for queue in queues:
    queue.pause()
for queue in queues:
    queue.resume()
removed = [queue.drain() for queue in queues]

PHP

$queues = [$invoices, $payments];
foreach ($queues as $queue) {
    $queue->pause();
}
foreach ($queues as $queue) {
    $queue->resume();
}
$removed = array_map(fn ($queue) => $queue->drain(), $queues);

Go

queues := []*bunqueue.Queue{invoices, payments}
for _, queue := range queues {
    if err := queue.Pause(); err != nil { return err }
}
for _, queue := range queues {
    if err := queue.Resume(); err != nil { return err }
}

Rust

let queues = [&invoices, &payments];
for queue in queues {
    queue.pause()?;
}
for queue in queues {
    queue.resume()?;
}

Elixir

queues = [invoices, payments]
Enum.each(queues, fn queue -> :ok = Bunqueue.Queue.pause(queue) end)
Enum.each(queues, fn queue -> :ok = Bunqueue.Queue.resume(queue) end)
{:ok, removed} =
  Enum.reduce_while(queues, {:ok, []}, fn queue, {:ok, counts} ->
    case Bunqueue.Queue.drain(queue) do
      {:ok, count} -> {:cont, {:ok, [count | counts]}}
      {:error, error} -> {:halt, {:error, error}}
    end
  end)

Synchronous and awaitable forms

listQueues() and the synchronous bulk operations use the in-process embedded manager. For TCP queues, use listQueuesAsync, pauseAllAsync, resumeAllAsync, drainAllAsync, and obliterateAllAsync; they operate on every queue created through the group and wait for completion.

Isolate tenants

Bun

const tenantA = new QueueGroup('tenant-a');
const tenantB = new QueueGroup('tenant-b');

const tasksA = tenantA.getQueue('tasks', { embedded: true });  // "tenant-a:tasks"
const tasksB = tenantB.getQueue('tasks', { embedded: true });  // "tenant-b:tasks"

Node.js / Deno

const tenantA = new QueueGroup('tenant-a');
const tenantB = new QueueGroup('tenant-b');

const tasksA = tenantA.getQueue('tasks', { embedded: false });  // "tenant-a:tasks"
const tasksB = tenantB.getQueue('tasks', { embedded: false });  // "tenant-b:tasks"

Python

tasks_a = Queue("tenant-a:tasks")
tasks_b = Queue("tenant-b:tasks")

PHP

$tasksA = new Queue('tenant-a:tasks');
$tasksB = new Queue('tenant-b:tasks');

Go

tasksA := bunqueue.NewQueue("tenant-a:tasks", bunqueue.Options{})
tasksB := bunqueue.NewQueue("tenant-b:tasks", bunqueue.Options{})

Rust

let tasks_a = Queue::new("tenant-a:tasks", ConnectionOptions::default());
let tasks_b = Queue::new("tenant-b:tasks", ConnectionOptions::default());

Elixir

tasks_a = Bunqueue.queue("tenant-a:tasks")
tasks_b = Bunqueue.queue("tenant-b:tasks")

Separate environments

Bun

const env = process.env.NODE_ENV || 'development';
const group = new QueueGroup(`${env}-tasks`);
const queue = group.getQueue('jobs', { embedded: true });
// "development-tasks:jobs" or "production-tasks:jobs"

Node.js / Deno

const env = process.env.NODE_ENV || 'development';
const group = new QueueGroup(`${env}-tasks`);
const queue = group.getQueue('jobs', { embedded: false });
// "development-tasks:jobs" or "production-tasks:jobs"

Python

env = os.getenv("APP_ENV", "development")
queue = Queue(f"{env}-tasks:jobs")

PHP

$env = getenv('APP_ENV') ?: 'development';
$queue = new Queue("{$env}-tasks:jobs");

Go

env := os.Getenv("APP_ENV")
if env == "" { env = "development" }
queue := bunqueue.NewQueue(env+"-tasks:jobs", bunqueue.Options{})

Rust

let env = std::env::var("APP_ENV").unwrap_or_else(|_| "development".into());
let queue = Queue::new(format!("{env}-tasks:jobs"), ConnectionOptions::default());

Elixir

env = System.get_env("APP_ENV", "development")
queue = Bunqueue.queue("#{env}-tasks:jobs")

Methods Reference

Method	Description
getQueue(name, opts?)	Get a queue within the group (embedded or TCP)
getWorker(name, processor, opts?)	Create a worker for a queue in the group (embedded or TCP)
listQueues()	List queue names in the group, without prefix (embedded only)
pauseAll()	Pause all queues in the group (embedded only)
resumeAll()	Resume all queues in the group (embedded only)
drainAll()	Remove waiting jobs from all queues (embedded only)
obliterateAll()	Remove all data from all queues (embedded only)
listQueuesAsync()	List tracked group queues in either runtime
pauseAllAsync() / resumeAllAsync()	Await group control in either runtime
drainAllAsync()	Drain all tracked queues and return the aggregate count
obliterateAllAsync()	Await removal of all tracked queue data

Related Guides

• Queue API - Options accepted by getQueue
• Worker API - Options accepted by getWorker