class WorkersQueue
implements IQueue
Since 0.2.0

A background job queue backed by Cloudflare Queues.

Two of the three committed methods map onto the platform directly. addRecurring does not and throws: Cloudflare has no recurring queue message, and the platform's answer — Cron Triggers — is a different mechanism with its own handler export, reached through WorkersCron.

Consuming requires the application to export the handler createQueueHandler builds:

Examples

Example 1

const app = createApplication({
  plugins: [RuntimePlugin({ env }), CloudflarePlugin({ env, queue: { binding: 'JOBS' } })],
});
await app.start();

const queue = app.services.get<IQueue>(CAPABILITIES.QUEUE);
queue.process<{ to: string }>('send-email', async (job) => {
  await mailer.send(job.data);
}, { concurrency: 5 });

export default { fetch: app.fetch, queue: createQueueHandler(app) };

Constructors

WorkersQueue(
producer: IQueueProducer,
)
Parameters
producer: IQueueProducer

The Queues producer binding

Unique id source; pass IRuntimeServices

optional
options: WorkersQueueOptions

Logger and delay cap

Methods

add<T>(
name: string,
data: T,
options?: AddJobOptions
): Promise<string>

Enqueues a job.

The returned id is generated here rather than by the platform, because producer.send() resolves to void. It travels inside the message envelope, so the id this returns is the id the processor sees as job.id.

addRecurring<T>(
name: string,
_data: T,
options: RecurringOptions
): Promise<void>

Not supported on Cloudflare Queues.

dispatch(batch: IQueueMessageBatch): Promise<void>

Dispatches one delivered batch into the registered processors.

Every message is acked or retried exactly once:

  • a body that is not a readable envelope → retry(), because a message the consumer cannot route is a configuration problem and acking would discard it permanently and silently;
  • a name with no registered processor → retry(), same reason;
  • attempts past the envelope's maxAttempts → ack() without running the processor, since the caller asked for the job to stop;
  • a processor that throws → retry(), leaving the queue's own max_retries and dead-letter configuration to decide what happens next;
  • a processor that resolves → ack().
process<T>(
name: string,
processor: JobProcessor<T>,
options?: ProcessOptions
): void

Registers a processor for a job name.

Registering twice under one name replaces the earlier processor, matching QueueService.process in queue-plugin so the committed port behaves the same way on every backend.

Usage

import { WorkersQueue } from "cloudflare-plugin/src/index.ts";