Stop AsyncIterator/ReadableStream when streaming serialization stops?

[yume-chan]

I'm using Cloudflare Workers, where API can return a ReadableStream (via new Response(new ReadableStream(...), { status: 200 }). But I want to return more complex objects containing ReadableStreams, so I used seroval to serialize them:

import { toCrossJSONStream } from "seroval";
import { ReadableStreamPlugin } from "seroval-plugins/web";

export function serializeStream(object: unknown) {
  let cancel!: () => void;
  return new ReadableStream<string>({
    start(controller) {
      cancel = toCrossJSONStream(object, {
        plugins: [ReadableStreamPlugin],
        onParse(node, initial) {
          const line = (initial ? "" : "\n") + JSON.stringify(node);
          // console.log("[server-function]", "onParse", line);
          controller.enqueue(line);
        },
        onDone() {
          // console.log("[server-function]", "onDone");
          controller.close();
        },
      });
    },
    cancel() {
      cancel();
    },
  }).pipeThrough(new TextEncoderStream());
}

However, one thing I lost is, when client disconnects before receiving the full response, the cancel method above will be called, but it doesn't actually cancel the ReadableStream being serialized:

import { toCrossJSONStream } from "seroval";
import { ReadableStreamPlugin } from "seroval-plugins/web";

const cancel = toCrossJSONStream(
  new ReadableStream({
    async start(controller) {
      while (true) {
        await new Promise((resolve) => {
          setTimeout(resolve, 1000);
        });
        console.log("ReadableStream enqueue");
        controller.enqueue(0);
      }
    },
    cancel() {
      console.log("ReadableStream cancel");
    },
  }),
  {
    plugins: [ReadableStreamPlugin],
    onParse(node, initial) {
      console.log("onParse", JSON.stringify(node), initial);
    },
  },
);

setTimeout(() => {
  console.log("call cancel");
  cancel();
}, 5000);

After 5 seconds, it stops printing onParse, but there is no "ReadableStream cancel", and "ReadableStream enqueue" continues indefinitely.

[lxsmnsyc]

fter 5 seconds, it stops printing onParse, but there is no "ReadableStream cancel", and "ReadableStream enqueue" continues indefinitely.

This is because Seroval doesn't cancel ReadableStream instances when it is cancelled, it's up to the user to cancel them. It's a bit of a two-way street, and I don't really want Seroval to be the one to manage this.

Take note that I will still put this into consideration, but don't really get your hopes up on this.

[yume-chan]

it's up to the user to cancel them

But ReadableStreamPlugin calls ReadableStream.getReader which locks the stream, so only Seroval can cancel it. If Seroval want users to cancel the ReadableStream, it would also need to release the reader when toCrossJSONStream stops.

[lxsmnsyc]

@yume-chan yeah this is where the two-way street happens. Either the user passes a ReadableStream that seroval will manage (getReader is safe, so no locking issue, but no cancellation allowed) or the user passes a tee stream

Still, this is still up for consideration

[yume-chan]

I don't understand what you mean. Even if I tee the stream, pass one to Seroval, then cancel the other, the original stream still won't be cancelled because it needs both branches to be cancelled.

const stream = new ReadableStream({
    async start(controller) {
        while (true) {
            await new Promise((resolve) => {
                setTimeout(resolve, 1000);
            });
            console.log("ReadableStream enqueue");
            controller.enqueue(0);
        }
    },
    cancel() {
        console.log("ReadableStream cancel");
    },
})

const [stream1, stream2] = stream.tee();

stream1.pipeTo(new WritableStream())

setTimeout(() => {
    console.log("call cancel");
    stream2.cancel()
}, 5000);