Sending stream responses (experimental)

[pi0]

Moving from (#19)

Coming with h3@1.7.0 and new event.handled flag it is possible to take over h3 response mechanism and send streams. For Workers, we can check res._body property existence. Direct fetch in unenv (src) passes the body as is.

This way we can implement sendStream with the minimum possible changes. Once tested enough and working for common presets, I plan to add this as a built-in (nitro/h3) utility.

Example usage: https://github.com/unjs/nitro-deploys/blob/main/routes/stream.ts

Working deployments:

• Cloudflare workers: https://nitro-deployment.pi0.workers.dev/stream
• Netlify functions: https://nitro-deployment-edge.netlify.app/stream
• Vercel edge: https://nitro-app-edge.vercel.app/stream

Screen.Recording.2023-06-21.at.02.56.44.mov

[Hebilicious]

This is approach is working great for me, see here.

@pi0 Could we replace the sendStream in h3 with this new updated version instead of adding it to nitro ?

[jlucaso1]

I'm experiencing problems detecting when the request is closed to clean up processing.

  event.node.req.once("close", () => {
    cleanupFunction() // <-- this is never called
  })

@pi0 can you update the example of streams to handle this? I don't know if this is a bug.

[dosstx]

So how does one set this up for Nuxt ?

[Hebilicious]

I'm experiencing problems detecting when the request is closed to clean up processing.

  event.node.req.once("close", () => {
    cleanupFunction() // <-- this is never called
  })

@pi0 can you update the example of streams to handle this? I don't know if this is a bug.

For node.js, you could pass a cleanup function to your sendStream handler, and run it as part of the close() callback of WritableStream.

function sendStream(event: H3Event, stream: ReadableStream, cleanup: any) {
  // Mark to prevent h3 handling response
  event._handled = true;

  // Workers (unenv)
  (event.node.res as unknown as { _data: BodyInit })._data = stream;

  // Node.js
  if (event.node.res.socket) {
    stream.pipeTo(
      new WritableStream({
        write(chunk) {
          event.node.res.write(chunk);
        },
        close() {
         cleanup()
          event.node.res.end();
        },
      })
    );
  }
}

[jlucaso1]

@Hebilicious
I attempted this approach; however, when I forcefully close the request, the cleanup() function does not get invoked as expected.

[Hebilicious]

@jlucaso1 Try abort()

[pi0]

Update: Web Stream Support directly landing in H3 (unjs/h3#432).

We still might need to tweak some presets for full compatibility and experiment worker support.

[dosstx]

Can we use this yet in Nuxt or not yet? Confused.

[pi0]

@dosstx h3 changes are still not released and being tested. But very soon you would be able to try it via nuxt or nitro edge channels.

[dosstx]

@pi0 OK thank you very much. Sorry to waste your time looking at these kinds of questions. Looking forward to the release!

[Hebilicious]

@pi0 OK thank you very much. Sorry to waste your time looking at these kinds of questions. Looking forward to the release!

You can already use it with a few workarounds: vercel/ai#295
As pi said, soon you will be able to use it without the workarounds.

[Giancarlo-Ma]

how can i intercept the streaming content to save it to db on cf worker

I have tried this:
`
const reader = stream.getReader();
let { readable, writable } = new TransformStream();
// Workers (unenv)
// @ts-expect-error _data will be there.
event.node.res._data = readable;
const writer = writable.getWriter();
// pipe stream to writable and log the chunk content use reader
while (true) {
const { done, value } = await reader.read();

if (done) {
  console.log("Stream complete");
  await writer.close();

  break;
}
const textDecoder = new TextDecoder("utf-8");
console.log(textDecoder.decode(value));
await writer.write(value);

}
`
but only get one whole pice after a while, not real streaming in front end

[pi0]

streaming support should be more stable now. please report any issues if you encounter so we can improve 🙏🏼