1// Async (non-Effect) Worker that produces to a queue binding AND consumes
2// the same queue, recording delivered bodies in instance memory so the test
3// can poll them back over HTTP. Producer + consumer in one worker is the
4// supported local shape: the broker lives in the consuming instance.
5interface Env {
6 QUEUE: {
7 send(body: unknown, options?: { contentType?: string }): Promise<void>;
8 sendBatch(
9 messages: Iterable<{ body: unknown; contentType?: string }>,
10 ): Promise<void>;
11 };
12}
13
14interface MessageLike {
15 body: unknown;
16}
17
18const received: string[] = [];
19const batches: number[] = [];
20
21export default {
22 fetch: async (request: Request, env: Env) => {
23 const url = new URL(request.url);
24 if (url.pathname === "/plain") {
25 return Response.json({ ok: true });
26 }
27 if (url.pathname === "/send") {
28 const text = url.searchParams.get("text") ?? "hello";
29 await env.QUEUE.send(text, { contentType: "text" });
30 return Response.json({ sent: text });
31 }
32 if (url.pathname === "/sendbatch") {
33 const text = url.searchParams.get("text") ?? "hello";
34 await env.QUEUE.sendBatch([
35 { body: `${text}-a`, contentType: "text" },
36 { body: { text, index: 2 }, contentType: "json" },
37 ]);
38 return Response.json({ sent: 2 });
39 }
40 if (url.pathname === "/received") {
41 return Response.json({ received });
42 }
43 if (url.pathname === "/batches") {
44 return Response.json({ batches });
45 }
46 return new Response("not found", { status: 404 });
47 },
48 queue: async (batch: { messages: MessageLike[] }) => {
49 batches.push(batch.messages.length);
50 for (const message of batch.messages) {
51 received.push(String(message.body));
52 }
53 },
54};