stream-error-handling.mjs

const { controller, readable } = await new Promise((resolve) => {
  let readable = new ReadableStream({
    start: (controller) => setTimeout(() => resolve({ controller, readable })),
  });
});
const encoder = new TextEncoder();
const decoder = new TextDecoder();

const transformer = new TransformStream({
  transform(chunk, transformController) {
    if (chunk.type === "text") {
      transformController.enqueue(encoder.encode(chunk.text));
    }

    if (chunk.type === "error") {
      transformController.error(chunk.text);
    }
  },
});

const transformStream = readable.pipeThrough(transformer);
const response = new Response(transformStream);
const reader = response.body.getReader();

function read() {
  return new Promise(async (resolve) => {
    while (true) {
      const { value, done } = await reader.read().catch((error) => {
        console.log("Captured error", { error });
        return { done: true };
      });

      if (done) break;
      console.log("Received", { chunk: decoder.decode(value) });
    }
    resolve();
  });
}

const promise = read();
controller.enqueue({ type: "text", text: "hello" });
controller.enqueue({ type: "error", text: "Custom error message" });

await promise;
console.log("Graceful end");