using ReadableStream as Response will not send each chunk one by one but in batches

[BennyAlex]

I return a ReadableStream in a bun.serve server, but in the browser I always see multiple chunks at once instead of getting one by one.

Is this normal? Can I change this behavior?

[cirospaciari]

Let check some alternatives:

using server = Bun.serve({
  port: 0,
  async fetch(req) {
    return new Response(
      new ReadableStream({
        async start(controller) {
          controller.enqueue("Hello, world!");
        },

        async pull(controller) {
          for (let i = 0; i < 10; i++) {
            await Bun.sleep(1000);

            controller.enqueue("Hello, world!");
          }
          controller.close();
        },
      }),
    );
  },
});

const response = await fetch(server.url);
const reader = response.body.getReader();
const decoder = new TextDecoder();
while (true) {
  const { done, value } = await reader.read();
  if (done) {
    break;
  }
  console.log("chunk:", decoder.decode(value));
}

In this case we will flush on every await call so you will see:

chunk: Hello, world!
chunk: Hello, world!
chunk: Hello, world!
chunk: Hello, world!
chunk: Hello, world!
chunk: Hello, world!
chunk: Hello, world!
chunk: Hello, world!
chunk: Hello, world!

Removing the await would give you something like:

chunk: Hello, world!Hello, world!Hello, world!Hello, world!Hello, world!Hello, world!Hello, world!Hello, world!Hello, world!Hello, world!

The broswer may wait for the entire document to load before showing up something, but fetch will get chunk by chunk.

If you wanna tell the browser to always update you chunk by chunk you can use { headers: { "content-type": "text/event-stream" } }

[BennyAlex]

@cirospaciari
Thanks for helping.

When I send a stringified json I get multiple chunks at once sometimes:

const server = Bun.serve({
  port: 3000,
  async fetch(req) {
    const ip = server.requestIP(req);
    console.log("Request from IP:", ip);

    if (req.method === "OPTIONS") {
      return new Response("OK", {
        headers: {
          "Access-Control-Allow-Origin": "*",
          "Access-Control-Allow-Methods": "GET, POST, PUT, DELETE, OPTIONS",
          "Access-Control-Allow-Headers": "*",
        },
      });
    }

    return new Response(
      new ReadableStream({
        async start(controller) {
          console.log("Stream started");
          controller.enqueue(JSON.stringify({ text: "START" }));
        },

        async pull(controller) {
          for (let i = 0; i < 10; i++) {
            await Bun.sleep(400);
            console.log("Stream chunk", i);
            controller.enqueue(JSON.stringify({ text: "Hello, world!" + i }));
          }
          controller.close();
        },
      }),
      {
        headers: {
          "Access-Control-Allow-Origin": "*",
          "Access-Control-Allow-Methods": "GET, POST, PUT, DELETE, OPTIONS",
          "Access-Control-Allow-Headers": "*",
        },
      }
    );
  },
});

console.log("Server started at http://localhost:" + server.port);

Is there a working alternative to sending ReadableStream

[BennyAlex]

I also added the header but when I remove the await sleep or set it to a low value like 50 the server will send multiple chunks

[BennyAlex]

In node it worked fine

[cirospaciari]

Is never a guarantee that everything will be flush in a individual way (with any runtime), this may be buffered/corked, the socket may not be readable/writable at the time, you may receive only part of the message or even byte by byte is a possibility and a lot more factors in, on HTTP if you wanna do messaging you have some options like SSE protocol or WebSockets are the common way (you can also customize your own protocol etc) SSE is very simple and you can use EventSource in browser, and WebSocket is more complex but Bun gives you the websocket interface and the browser also have a WebSocket object to make it easier.

About SSE I recommend you read about it in:
https://html.spec.whatwg.org/multipage/server-sent-events.html
Bun also have a timeout for idle sockets of 10s, you can set it higher or disable using 0, or use server.timeout(req, seconds) do change in a per-request bases.

Here is a naive (and incomplete) example of SSE:

import type { Server } from "bun";

async function sendSSECustom(controller: ReadableStreamDirectController, eventName: string, data: string) {
  await controller.write(`event: ${eventName}\ndata:${JSON.stringify(data)}\n\n`);
  await controller.flush();
}
async function sendSSEMessage(controller: ReadableStreamDirectController, data: string) {
  await controller.write(`data:${JSON.stringify(data)}\n\n`);
  await controller.flush();
}
function sse(req: Request, server: Server): Response {
  const signal = req.signal;
  server.timeout(req, 120); // 2 minutes of idle time
  return new Response(
    new ReadableStream({
      type: "direct",
      async pull(controller: ReadableStreamDirectController) {
        while (!signal.aborted) {
          await sendSSECustom(controller, "bun", "Hello, World!");
          await sendSSEMessage(controller, "Hello, World!");
          await controller.flush();
          await Bun.sleep(1000);
        }
        controller.close();
      },
    }),
    { status: 200, headers: { "Content-Type": "text/event-stream" } },
  );
}
Bun.serve({
  port: 3000,
  fetch(req, server) {
    if (new URL(req.url).pathname === "/stream") {
      return sse(req, server);
    }
    return new Response("Hello, World!");
  },
});

You may will wanna use something like lastEventId to resume reconnections (by sending id:), and send retry: to control retry intervals.

const lastEventId = req.headers.get("last-event-id")