INTEGRITY Cloudflare Docs

← Queues / queues / configuration

JavaScript APIs

Cloudflare Queues is integrated with Cloudflare Workers. To send and receive messages, you must use a Worker.

A Worker that can send messages to a Queue is a producer Worker, while a Worker that can receive messages from a Queue is a consumer Worker. It is possible for the same Worker to be a producer and consumer, if desired.

In the future, we expect to support other APIs, such as HTTP endpoints to send or receive messages. To report bugs or request features, go to the Cloudflare Community Forums ↗. To give feedback, go to the #queues ↗ Discord channel.

Producer

These APIs allow a producer Worker to send messages to a Queue.

An example of writing a single message to a Queue:

index.js
export default {
	async fetch(req, env, ctx) {
		await env.MY_QUEUE.send({
			url: req.url,
			method: req.method,
			headers: Object.fromEntries(req.headers),
		});
		return new Response("Sent!");
	},
};
index.ts
interface Env {
  readonly MY_QUEUE: Queue;
}

export default {
  async fetch(req, env, ctx): Promise<Response> {
    await env.MY_QUEUE.send({
      url: req.url,
      method: req.method,
      headers: Object.fromEntries(req.headers),
    });
    return new Response("Sent!");
  },
} satisfies ExportedHandler<Env>;
from workers import Response, WorkerEntrypoint

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        await self.env.MY_QUEUE.send({
            "url": request.url,
            "method": request.method,
            "headers": dict(request.headers),
        })
        return Response("Sent!")

The Queues API also supports writing multiple messages at once:

index.js
const sendResultsToQueue = async (results, env) => {
	const batch = results.map((value) => ({
		body: value,
	}));
	await env.MY_QUEUE.sendBatch(batch);
};
index.ts
const sendResultsToQueue = async (results: Array<unknown>, env: Env) => {
	const batch: MessageSendRequest[] = results.map((value) => ({
		body: value,
	}));
	await env.MY_QUEUE.sendBatch(batch);
};
async def send_results_to_queue(results, env):
    batch = [
        {"body": value}
        for value in results
    ]
    await env.MY_QUEUE.sendBatch(batch)

Queue

A binding that allows a producer to send messages to a Queue.

interface Queue<Body = unknown> {
  send(body: Body, options?: QueueSendOptions): Promise<QueueSendResult>;
  sendBatch(messages: Iterable<MessageSendRequest<Body>>, options?: QueueSendBatchOptions): Promise<QueueSendResult>;
  metrics(): Promise<QueueMetrics>;
}

MessageSendRequest

A wrapper type used for sending message batches.

interface MessageSendRequest<Body = unknown> {
  body: Body;
  contentType?: QueueContentType;
  delaySeconds?: number;
}

QueueSendOptions

Optional configuration that applies when sending a message to a queue.

QueueSendBatchOptions

Optional configuration that applies when sending a batch of messages to a queue.

QueuesContentType

A union type containing valid message content types.

// Default: json
type QueuesContentType = "text" | "bytes" | "json" | "v8";

If you specify an invalid content type, or if your specified content type does not match the message content's type, the send operation will fail with an error.

QueueSendResult

The result of a successful send operation.

interface QueueSendResult {
	metadata: {
		metrics: QueueMetrics;
	};
}

QueueMetrics

Realtime metrics for a queue.

interface QueueMetrics {
	backlogCount: number;
	backlogBytes: number;
	oldestMessageTimestamp: number;
}

Consumer

These APIs allow a consumer Worker to consume messages from a Queue.

To define a consumer Worker, add a queue() function to the default export of the Worker. This will allow it to receive messages from the Queue.

By default, all messages in the batch will be acknowledged as soon as all of the following conditions are met:

  1. The queue() function has returned.
  2. If the queue() function returned a promise, the promise has resolved.
  3. Any promises passed to waitUntil() have resolved.

If the queue() function throws, or the promise returned by it or any of the promises passed to waitUntil() were rejected, then the entire batch will be considered a failure and will be retried according to the consumer's retry settings.

index.js
export default {
	async queue(batch, env, ctx) {
		for (const message of batch.messages) {
			console.log("Received", message.body);
		}
	},
};
index.ts
interface Env {
  // Add your bindings here
}

export default {
  async queue(batch, env, ctx): Promise<void> {
    for (const message of batch.messages) {
      console.log("Received", message.body);
    }
  },
} satisfies ExportedHandler<Env>;
from workers import WorkerEntrypoint

class Default(WorkerEntrypoint):
    async def queue(self, batch):
        for message in batch.messages:
            print("Received", message)

The env and ctx fields are as documented in the Workers documentation.

TypeScript message types

You can type queue messages with Queue<T> on the producer and ExportedHandler<Env, T> on the consumer.

type MyMessage = {
  id: string;
};

interface Env {
  MY_QUEUE: Queue<MyMessage>;
}

export default {
  async queue(batch) {
    for (const message of batch.messages) {
      console.log(message.body.id);
    }
  },
} satisfies ExportedHandler<Env, MyMessage>;

For primitive messages, use Queue<number> or satisfies ExportedHandler<Env, number>. If you do not specify a type, message.body is unknown.

Or alternatively, a queue consumer can be written using the (deprecated) service worker syntax:

addEventListener('queue', (event) => {
	event.waitUntil(handleMessages(event));
});

In service worker syntax, event provides the same fields and methods as MessageBatch, as defined below, in addition to waitUntil() ↗.

MessageBatch

A batch of messages that are sent to a consumer Worker.

interface MessageBatch<Body = unknown> {
  readonly queue: string;
  readonly messages: readonly Message<Body>[];
  ackAll(): void;
  retryAll(options?: QueueRetryOptions): void;
}

Message

A message that is sent to a consumer Worker.

interface Message<Body = unknown> {
  readonly id: string;
  readonly timestamp: Date;
  readonly body: Body;
	readonly attempts: number;
  ack(): void;
  retry(options?: QueueRetryOptions): void;
}

QueueRetryOptions

Optional configuration when marking a message or a batch of messages for retry.

interface QueueRetryOptions {
  delaySeconds?: number;
}