class WorkersBroker
implements IMessageBroker
Since 0.2.0

A message broker backed by Cloudflare Queues.

Consuming requires the application to export the handler createMessagingHandler builds, and the queue to be declared as a consumer in wrangler.toml:

Examples

Example 1

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

const broker = app.services.get<IMessageBroker>(CAPABILITIES.MESSAGING);
await broker.subscribe<{ id: string }>('user.created', async (user) => {
  await mailer.send(user.id);
});

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

Constructors

WorkersBroker(
producer: IQueueProducer,
runtime: BrokerRuntime,
)

Builds a broker over one Queues producer binding.

Parameters
producer: IQueueProducer

The Queues producer binding

runtime: BrokerRuntime

Id, clock, and timer source; pass IRuntimeServices

optional
options: WorkersBrokerOptions

Logger thunk and the optional reply-inbox binding

Methods

connect(): Promise<void>

No-op: a producer binding is ready as soon as the Worker has its env, and there is no connection to open.

disconnect(): Promise<void>

Closes the reply inbox, rejects every in-flight request, and drops every subscription.

dispatch(batch: IQueueMessageBatch): Promise<void>

Dispatches one delivered batch into the registered subscribers and responders.

Every message is acked or retried exactly once:

  • a body that is not a readable envelope → retry(), because a foreign producer or a version skew is a configuration problem and acking would discard the message permanently and silently;
  • a publish whose topic has no subscriber → ack(), because publishing to a topic nobody listens on is ordinary pub/sub, and retrying would burn the queue's retry budget and dead-letter every fire-and-forget message;
  • a publish whose handler throws → retry(), leaving the queue's own max_retries and dead-letter configuration to decide what happens next;
  • a request whose topic has no responder, or whose responder throws → ack() after sending the caller a failed reply, because the caller is waiting and a redelivery would re-run a handler that already ran.
publish<T>(
topic: string,
message: T
): Promise<void>

Publishes a message to a topic.

request<TReq, TRes>(
topic: string,
message: TReq,
options?: RequestOptions
): Promise<TRes>

Sends a request and awaits its single correlated reply.

respond<TReq, TRes>(
topic: string,
handler: RequestHandler<TReq, TRes>,
options?: SubscribeOptions
): Promise<ISubscription>

Registers a responder for a request topic.

subscribe<T>(
topic: string,
handler: MessageHandler<T>,
options?: SubscribeOptions
): Promise<ISubscription>

Subscribes to a topic.

Registration only: delivery happens when the Worker's queue export dispatches a batch into WorkersBroker.dispatch.

Usage

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