Setu-TS
    Preparing search index...

    Class WorkersQueue

    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:

    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) };

    0.2.0

    Implements

    Index
    constructor
    add addRecurring dispatch process

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

      Type Parameters

      • T

        The payload type

      Parameters

      • name: string

        Job name; a processor must be registered under it to consume

      • data: T

        Job payload, serialized as JSON by the platform

      • Optionaloptions: AddJobOptions

        delayMs (converted to whole seconds, rounded up) and maxAttempts (enforced at dispatch, since Cloudflare's max_retries is queue-wide configuration rather than per message)

      Returns Promise<string>

      The job id

      When delayMs exceeds the configured cap

    • 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().

      Parameters

      Returns Promise<void>

      Resolves once every message has been acked or retried

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

      Type Parameters

      • T

        The payload type

      Parameters

      • name: string

        Job name

      • processor: JobProcessor<T>

        Invoked per delivered job

      • Optionaloptions: ProcessOptions

        concurrency bounds how many of one batch's messages for this name run at a time

      Returns void

    Settings
    Member Visibility

    On This Page
    Constructors
    Methods