elysiajs / stream Goto Github PK

According to MDN, when sending multiple lines of data via SSE, each line should be preceded by data: , and the current implementation is clearly doing it wrong.

If you try to send multiple lines of data like:

app.get('/live', () => {
    return new Stream(async (stream) => {
        stream.send("first line\nsecond line");
        stream.close();
    });
})

You will get:

id: <some id>
data: first line
second line  // Wrong here

I am having this issue, since I've migrated from Node to Bun, writing streams by hand or using this plugin.

The problem consists of streams accumulating chunks and displaying all at once OR chunks simply being lost.

Here is an example:
PS: If I simply return the Stream object like it's said in the docs, Elysia returns a "jsonfied" version of new Stream(...), so... I had to tweak it a little.

const stream = new Stream(async (st) => {
  let count = 1;
  const interval = setInterval(() => {
    st.send({ count });
    count++;
    if(count > 50) {
      st.close();
      clearInterval(interval);
    }
  }, 10);
});

return new Response(stream.value, {
  headers: {
    'Cache-Control': 'no-cache',
    'Connection': 'keep-alive',
    'Content-Type': 'text/event-stream',
    'Transfer-Encoding': 'chunked',
  },
  status: 200,
});

The code above works with 250ms timeouts or higher, but won't with faster frequencies.

Since the project I am working leverages gen AI, this timeout won't deliver an acceptable UX.

Assuming the snippet is within a handler, what could be going wrong?

When using elysia's streaming plugin with the CORS plugin, the streaming breaks. Below you can find a minimal reproducable example.

import { Elysia } from "elysia";
import cors from "@elysiajs/cors";
import Stream from "@elysiajs/stream";
import OpenAI from "openai";

const openai = new OpenAI({
  apiKey: Bun.env.OPENAI_API_KEY,
});

const app = new Elysia()
  .use(cors())
  .get(
    '/ai',
    ({ query: { prompt } }) =>
      new Stream(
        openai.chat.completions.create({
          model: 'gpt-3.5-turbo',
          stream: true,
          messages: [{
            role: 'user',
            content: prompt
          }]
        })
      )
  )
  .listen(PORT, () => {
    console.log(`🦊 Elysia is running at on port ${PORT}`)
  });

Doing this results in a content-type header of application/json along with a body of {}

This is running in a bun workspace

../../node_modules/@elysiajs/stream/src/index.ts(56,25): error TS2552: Cannot find name 'ReadableStreamController'. Did you mean 'ReadableStreamDirectController'?

bun version 1.1.13
@elyisa/stream 1.0.2
elysia 1.0.23

The Homepage and Repository links here points to the wrong github repo https://www.npmjs.com/package/@elysiajs/stream

Unfortunately, adding an onabort-handler for my SSE-endpoints has only been possible with a workaround which you can see in this minimum reproducible example.

Creating an interval which prevents the request from getting garbage collected makes the onabort work properly. Stream.close() also doesn't call onabort (which makes sense but isn't really practical if you just want to handle connection loss, no matter if the client aborted it or the server), it would be nice if the Stream-API could get extended to allow developers to handle disconnects in an intuitive way.

import { Elysia } from "elysia";
import { Stream } from "@elysiajs/stream";

// [1] => Uncommenting this makes onabort work properly (page close/client-side connection loss calls onabort)
// [2] => stream.close() doesn't trigger onabort as well
// Running the code like this leads to "connection closed" never getting logged even if it should happen
new Elysia()
  .get("/", ({ request }) => {
    request.signal.onabort = () => {
      console.log("connection closed");
      // [1]
      // clearInterval(_);
    };
    // [1]
    // const _ = setInterval(() => request.signal.aborted, Math.pow(2, 31) - 1);
    return new Stream(async (stream) => {
      stream.send("test");
      // [2]
      // stream.close();
    });
  })
  .listen(3000);

It would be great to have the event field (https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#event) added to the SSE stream, so that clients can subscribe to specific named events.

When the client closes the connection, the server maybe has to do something like clearing an interval.

app.get('/live', () => {
	return new Stream(async (stream) => {
		setInterval(() => {
			stream.send('hello')
		}, 1000)
	})
})

How am i supposed to achive that with the current version? I´d expect to be able to return a callback that will be called when the connection is closed to handle such stuff.

.pnpm/@elysiajs[email protected][email protected]/node_modules/@elysiajs/stream/dist/index.js:1
import { nanoid } from 'nanoid';
^^^^^^

SyntaxError: Cannot use import statement outside a module
Node.js v18.14.2

While Bun supports both es6 module and commonjs, node is stuck on using only one or the other. Adding type: module in package.json would likely fix the issue.

Hi, thanks for the great Stream plugin! Can you please add an example of how to use this together with the @elysiajs/cors plugin? It seems that when adding the CORS plugin, the SSE response is converted to json?