Hey @sor4chi thanks for taking this up! I'd added a few more convenience methods to my own StreamingApi implementation, which you might be interested in including:

// Lets you easily emit data in JSON lines format
log(obj: any) {
  return this.writeln(JSON.stringify(obj)).flush();
}

// Lets you send a ReadableStream directly, e.g. from a subrequest
async pipe(body: ReadableStream) {
  this.writer.releaseLock();
  await body.pipeTo(this.writable, { preventClose: true });
  this.writer = this.writable.getWriter();
}

My only thought is whether this should be called c.textStream or something, since everything is based around strings so far? But I'll leave that up to @yusukebe to decide. Maybe this implementation can be made generic to allow both strings and binary data? Afaik, it's just the TextEncoder and the "content-type": "text/plain" header that makes it string-y. Thoughts?

Hi @sor4chi!

Awesome! This will be set to be a highlight feature for our next minor update, v3.7.0.

I haven't looked closely the implementation yet, but my primary concerns are:

• Naming: As @geelen's mentioned, c.textStream() might be better if it supports only "content-type": "text/plain".
• Focus: If there's potential to make this feature more general – like creating a stream that supports binary data – it would be great. In that case, we could name it c.stream().

I'll be reviewing your implementation more closely later. Either way, this is an important feature, so take our time making it.

Hello @geelen, @yusukebe.
It took me a while, but I changed the interface based on your suggestion.
How do you like it?

@sor4chi

Great job! There are two points I'd like us to consider:

streamText()

While c.stream() can handle all cases, it's not necessity for c.textStream(). But, it could be useful, because when we often use text/plain for debugging. I think naming it c.streamText() might be more consistent, given the prefix matches c.stream().

And if you create a c.streamText(), you can do it without redundant:

  streamText = (
    cb: (stream: StreamingApi) => Promise<void>,
    arg?: StatusCode | ResponseInit,
    headers?: HeaderRecord
  ): Response => {
    headers = headers ?? {}
    headers['content-type'] = 'text/plain; charset=UTF-8'
    headers['x-content-type-options'] = 'nosniff'
    return this.stream(cb, arg ?? 200, headers)
  }

Note: Including x-content-type-options: nosniff would be beneficial when accessing via a web browser.

stream.wait()

Though it is also not necessary, stream.wait() could be helpful:

export class StreamingApi {
  //...
  wait(ms: number) {
    return new Promise((resolve) => setTimeout(resolve, ms))
  }
}

Sample

Combining all features, the code looks much cleaner:

app.get('/', (c) => {
  return c.streamText(async (stream) => {
    while (true) {
      const message = `It is ${new Date().toISOString()}`
      stream.writeln(message)
      await stream.wait(1000)
    }
  })
})

Cool!

@yusukebe
Thank you for pointing this out.
It certainly seems like a good idea to align stream/streamText and prefix from a completion perspective.
As for the wait method, it would certainly be useful.

Is it like leaving it up to the user whether to add transfer-encoding: chunked or not?
Also, should I put back charset=UTF-8?

@sor4chi

Is it like leaving it up to the user whether to add transfer-encoding: chunked or not?

Yes, we have to keep it because I think it should be chunked for a streaming response.

should I put back charset=UTF-8?

I've suggested that because of the compatibility for c.text():

I'm planning making 'text/plain; charset=UTF-8' as a variable to reduce the bundle size, but I'll do that later.

Ah, we can refer to the Vercel's AI sdk: https://sdk.vercel.ai/docs

Back-pressure and cancellation management (detailed in Vercel AI SDK here).

A minimalist code I'm using now with Hono and Server Sent Event message format

const completionStream = await openai.chat.completions.create(....)
const completionStreamIterator = completionStream[Symbol.asyncIterator]();

const eventStream = new ReadableStream({
  // Client asks for more data
  async pull(controller) {
      const { done, value } = await completionStreamIterator.next();
      
      // Nothing more from OpenAI
      if (done) {
       // Close client connection, we could send a "finish" SSE custom event before closing
        controller.close()
      }
      controller.enqueue(`data: ${value.choices[0]?.delta?.content || ""}\n\n`);
  },
  // Client aborts the connection
  async cancel(reason) {
      // Cancel OpenAI upstream
      await completionStreamIterator.return?.(reason);
  }
}).pipeThrough(new TextEncoderStream());
 
// Respond with the stream and SSE required headers
return new Response(eventStream, {
  headers: {
    ...corsHeaders,
    "Content-Type": "text/event-stream; charset=utf-8",
    "Cache-Control": "no-store",
    Connection: "keep-alive",
  },
});
  

@yusukebe When testing my openai proxy (not using hono c.stream) with CloudFlare functions I get more than 100ms CPU time according to the completion length form OpenAI (so way above the 50ms CPU time limit).

What CPU time you get with the c.stream based solution?

About SSE and cancellation with bun there is this issue:
Bun does not support http2 right now and the cancellation management is a bit tricky:
oven-sh/bun#2443
oven-sh/bun#3390

Similarly, it appears that AWS Lambda's response streaming only supports HTTP1.1 as well.

You can find more information at

• https://docs.aws.amazon.com/lambda/latest/dg/configuration-response-streaming.html
• https://docs.aws.amazon.com/lambda/latest/dg/runtimes-custom.html#runtimes-custom-response-streaming.

It's a challenging question to decide how generic to make the system, but I think it would be good to have options in the roadmap.

There is no issues with HTTP1.1, only the limit of 6 parallel SSE connections per domain. The code stays the same.
The only JS runtime that requires a specific request cancellation management is bun for the moment:

async pull(controller) {
   // with bun we have to check if the request is cancelled here
   if (req.signal.aborted) {
          controller.close();
          // Cancel OpenAI upstream
          await completionStreamIterator.return?.(reason);
          return;
    }
    // ...
 }

I see...if HTTP 1.1 support is only for AWS lambda. There is no need to support it in this commit. There are other complicated differences, so I will absorb them with adaptor.

@arnaudbzn

When testing my openai proxy (not using hono c.stream) with CloudFlare functions I get more than 100ms CPU time according to the completion length form OpenAI (so way above the 50ms CPU time limit).

What CPU time you get with the c.stream based solution?

I think c.stream() doesn't resolve the problem, but when I tried a ChatGPT streaming gateway on Cloudflare Pages, the CPU time is not so long. It works well.

Hi @arnaudbzn @watany-dev

Thank you for your comments.

c.stream() is not dependent of HTTP version. With HTTP/1.1 on Wrangler, Node.js, Deno, and Bun, it will be chunked responses. In the case of Cloudflare, With HTTP/2 or H3, the transfer-encoding header with chunked set will be removed and the response will be kept as streaming. We should not care about protocol versions in Hono implementation.

About SSE. SSE is not supported in c.stream(). If you want to use SSE, we don't know if we will adopt it, but you can create a helper to make it happen.

So what I'm saying is that c.stream() should be a simple function to output HTTP streaming responses. If you want to do something else, you can create your own ReadableStream.

@sor4chi

So let's build on the current implementation!

@yusukebe
I'm done!

Hey!

Now, I'll merge it into the next branch which will be released as a next minor version v3.7.0.

Awesome work, @sor4chi ! Thanks!