Can eventStream take both sync / async functions?

[kenn]

Right now, eventStream takes a function like so:

export async function action({ request }) {
  return eventStream(request.signal, (send) => {
    async function run() {
      await LongRunningTask.start(send)
    }
    run()
    return () => {}
  })
}

But it would be helpful if it could optionally take async functions, and automatically invoke it. Now the code looks much cleaner.

export async function action({ request }) {
  return eventStream(request.signal, async (send) => {
    await LongRunningTask.start(send)
    return () => {}
  })
}

[kenn]

Even better if you could also receive optional AbortController instance as the second argument.

export async function action({ request }) {
  return eventStream(request.signal, async (send, controller) => {
    await LongRunningTask.start(send)
    controller.abort()
    return () => {}
  })
}

which does this boilerplate for you:

const controller = new AbortController()
request.signal.addEventListener('abort', () => controller.abort())

This will be useful for a lot of LLM-based applications.

[sergiodxa]

Your callback receives a close function as second argument already, it's used to close the stream from your function when you're done.

[kenn]

Ah, good to know!

[sergiodxa]

I need the cleanup function returned by the callback you pass to eventStream so I can call it when the connection is closed.

If it was a promise, I would have to wait for the promise to resolve to get the cleanup function, at that point the connection may have already been closed by the browser.

---

Also in your example, this:

export async function action({ request }) {
  return eventStream(request.signal, (send) => {
    async function run() {
      await LongRunningTask.start(send)
    }
    run()
    return () => {}
  })
}

Is the same as doing:

export async function action({ request }) {
  return eventStream(request.signal, (send) => {
    LongRunningTask.start(send)
    return () => {}
  })
}

Without needing to add async/await, but remember your LongRunningTask should be abortable/cancellable in case the connection is closed, or should handle that using request.signal

[kenn]

Sorry, the original post wasn't correct. The point is that close() must be called after await LongRunningTask.start(send) has been completed.

export async function action({ request }) {
  return eventStream(request.signal, (send, close) => {
    async function run() {
      await LongRunningTask.start(send)
      close()
    }
    run()

    return () => {}
  })
}

But now I get that you need a cleanup function as a return value. How about third (optional, as simple use cases don't need them) parameter, if and only if an async function is passed to the second param? Then the return of the main function can be a void.

export async function action({ request }) {
  return eventStream(request.signal, async (send, close) => {
    await LongRunningTask.start(send)
    close()
  }, () => {
    console.log('Cleanup eventStream')
  })
}

In the above case, I'd just write the cleanup code directly in the main function right before close(), so the third param may not even be necessary in many cases.

Now, my typical use cases would look like this, super simple:

export async function action({ request }) {
  return eventStream(request.signal, async (send, close) => {
    await LongRunningTask.start(send)
    close()
  })
}

[sergiodxa]

Even if it was a third argument I would need to make it required, because I need you to define close function.

If you're trying to run a long running task, it's better to use a background job instead, then, you can use eventStream to subscribe to know when it's done and do something when it's done.

Imagine something like this:

export function loader({ request, params }: ActionFunctionArgs) {
  let id = params.id;
  return eventStream(request.signal, (send, close) => {
    let unsubscribe = jobQueue.subscribe(id, async () => {
      // here do something when your job is done, and probably call `close()`
    });
   return unsubscribe;
  });
}

[kenn]

Even if it was a third argument I would need to make it required, because I need you to define close function.

How it's different from calling close() at the end of the function?

If you're trying to run a long running task, it's better to use a background job instead

My use case is ChatGPT API — using SSE because we want to push every token (sub-word) as fast as they arrive. It typically ends in 5-10 seconds (sending 100-1,000 events typically), and then disconnects. Was originally using WebSockets, but rewriting with SSE for this reason.

[sergiodxa]

How it's different from calling close() at the end of the function?

The function you return is needed to ensure your code doesn't leave any subscription or event listener, or another code running in the background when the connection is closed between the browser and server.

Think of it like useEffect on the server, if you do something like attach an event listener you need to clean it, otherwise the event listener will running and trying to set states, in this case if you don't clean it it will keep running server-side and try to send data to the browser when it won't work.

My use case is ChatGPT API — using SSE because we want to push every token (sub-word) as fast as they arrive. It typically ends in 5-10 seconds (sending 100-1,000 events typically), and then disconnects. Was originally using WebSockets, but rewriting with SSE for this reason.

If every request means the browser sent one request to get data from ChatGPT and then it's closed and you send a new request again, then you could move the async function to outside the eventStream, and inside the event stream subscribe to the token arriving.

Using some pseudo-code:

export function loader({ request, params }: ActionFunctionArgs) {
  let url = new URL(request.url);
  let response = await ChatGPT.send(url.searchParams.get("query"));

  return eventStream(request.signal, (send, close) => {
    response.addEventListener("token", onToken);
    response.addEventListener("end", onEnd);

    return () => {
      response.removeEventListener("token", onToken);
      response.removeEventListener("end", onEnd);
    };

    function onToken(token: string) {
      send(token);
    }

    function onEnd() {
      close();
    }
  });
}

[kenn]

How it's different from calling close() at the end of the function?

The function you return is needed to ensure your code doesn't leave any subscription or event listener, or another code running in the background when the connection is closed between the browser and server.

I see, I'm imagining it's called in the finally block of try {} finally {} to ensure it gets run?

In my case, only need close() for cleanup purposes, so a typical pattern would be:

  return eventStream(request.signal, (send, close) => {
    async function run() {
      await Task.run(send)
      close()
    }
    run()
    return close
  })

but then I'm getting RangeError: Maximum call stack size exceeded in doing this. I guess close is getting called within close this way?

My use case is ChatGPT API — using SSE because we want to push every token (sub-word) as fast as they arrive. It typically ends in 5-10 seconds (sending 100-1,000 events typically), and then disconnects. Was originally using WebSockets, but rewriting with SSE for this reason.

If every request means the browser sent one request to get data from ChatGPT and then it's closed and you send a new request again, then you could move the async function to outside the eventStream, and inside the event stream subscribe to the token arriving.

ChatGPT SDK here: https://github.com/openai/openai-node#streaming-responses

Here's my code that's been successfully working so far:

export async function action({ request }) {

  // const params = ...

  return eventStream(request.signal, (send, close) => {
    async function run() {
      await ChatGPT.run(send, params)
      close()
    }
    run()
    return () => {}
  })
}

Several points to note:

1. It's usually action (POST) as SSE client send full message history as JSON in request.body. (ChatGPT API is also POST-based SSE, so I imagine a lot of people do the same.)
2. Your code crosses the boundary between the route code and the ChatGPT concerns, by passing response to the closure. Encapsulating ChatGPT concerns in one single object and passing send function is much simpler as an API surface.
3. There's no event listeners, we're expected to just await on the streamed response via for await (const chunk of stream) as shown in this code. In our case, the ChatGPT.run does something like this:

for await (const chunk of stream) {
  const delta = chunk.choices[0]?.delta?.content || ''
  send({ event: 'message', data: delta })
}

Am I missing anything here? Otherwise I'm pretty satisfied with the current implementation, but just wanted to let you know that there's an area of improvements for this kind of use cases.

[sergiodxa]

It's usually action (POST) as SSE client send full message history as JSON in request.body. (ChatGPT API is also POST-based SSE, so I imagine a lot of people do the same.)

AFAIK, SSE only send GET requests, and there's no way to change the method, how are you using an action?

---

Your code crosses the boundary between the route code and the ChatGPT concerns, by passing response to the closure. Encapsulating ChatGPT concerns in one single object and passing send function is much simpler as an API surface

Passing things from the loader/action to the callback of eventStream is not a bad thing, it's actually intended, the idea is that you could run any async code like authenticating the user in the loader/action function before you send the eventStream response and start the streaming the data. In this case you can start the stream from OpenAI and use it inside the eventStream response.

---

There's no event listeners, we're expected to just await on the streamed response via for await (const chunk of stream) as shown in this code. In our case, the ChatGPT.run does something like this

Reading OpenAI docs you could do this (note I haven't tried it) and get events you can subscribe to:

export async function action({ request, params }: ActionFunctionArgs) {
  let messages = await getMessagesFromRequestSomehow(request);

  // Using this method, we can get a stream that has events
  let stream = openai.beta.chat.completions.stream({
    model: "gpt-4",
    messages,
    stream: true,
  });

  return eventStream(request.signal, (send, close) => {
    // Here we add even listeners to the stream
    stream.on("chunk", onChunk);
    stream.on("finalChatCompletion", close);
    // And here we remove them when the stream is closed
    return () => {
      stream.off("chunk", onChunk);
      stream.off("finalChatCompletion", close);
    };
    // This is the function that will be called when a chunk is received
    function onChunk(chunk: OpenAI.Chat.Completions.ChatCompletionChunk) {
      // ignore chunks if the stream is closed
      if (stream.aborted || request.signal.aborted) return;
      // get the data to stream
      let data = chunk.choices[0]?.delta?.content;
      if (data) send({ data }); // send the data
    }
  });
}

[kenn]

AFAIK, SSE only send GET requests, and there's no way to change the method, how are you using an action?

It's just a fetch POST call. Decoding the content is fairly easy according to: https://web.dev/eventsource-basics/

POST-based SSE is non-standard but very common when you need to pass a large JSON text as input. It's also common in Python / FastAPI, you can do the same with EventSourceResponse from sse-starlette.

Here's the client code to consume on the browser side:

export class SSEClient {
  static async start(func) { // => callback function with useState's `setContent` for instance
    try {
      const response = await fetch('/api/chat', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify([
          { role: 'system', content: 'Always answer in a friendly manner.' },
          { role: 'user', content: 'How old are you?' },
          { role: 'assistant', content: 'I was last updated in April 2023!' },
          { role: 'user', content: 'How do you like your job?' },
        ]),
      })
      await this.startLoop(response, func)
    } catch (e) {
      // Sentry.captureException(e)
    } finally {
      // Cleanup
    }
  }

  static async startLoop(response, func) {
    if (!response.body) return // body could be null
    const reader = response.body.getReader()
    const decoder = new TextDecoder()
    let streamedText = ''

    while (true) {
      const { done, value } = await reader.read()
      if (done) break

      const chunks = decoder
        .decode(value)
        .split('\n')
        .map((line) => {
          if (!line.startsWith('data: ')) return null
          return line.split('data: ').join('')
        })
        .filter(Boolean)

      for (const chunk of chunks) {
        if (!chunk) continue
        streamedText += chunk
        func(streamedText)    // => callback function with useState's `setContent` for instance
      }
    }
  }
}

---

There's no event listeners, we're expected to just await on the streamed response via for await (const chunk of stream) as shown in this code. In our case, the ChatGPT.run does something like this

Reading OpenAI docs you could do this (note I haven't tried it) and get events you can subscribe to:

export async function action({ request, params }: ActionFunctionArgs) {
  let messages = await getMessagesFromRequestSomehow(request);

  // Using this method, we can get a stream that has events
  let stream = openai.beta.chat.completions.stream({
    model: "gpt-4",
    messages,
    stream: true,
  });

  return eventStream(request.signal, (send, close) => {
    // Here we add even listeners to the stream
    stream.on("chunk", onChunk);
    stream.on("finalChatCompletion", close);
    // And here we remove them when the stream is closed
    return () => {
      stream.off("chunk", onChunk);
      stream.off("finalChatCompletion", close);
    };
    // This is the function that will be called when a chunk is received
    function onChunk(chunk: OpenAI.Chat.Completions.ChatCompletionChunk) {
      // ignore chunks if the stream is closed
      if (stream.aborted || request.signal.aborted) return;
      // get the data to stream
      let data = chunk.choices[0]?.delta?.content;
      if (data) send({ data }); // send the data
    }
  });
}

and compare it with this code:

export async function action({ request }) {
  let messages = await getMessagesFromRequestSomehow(request)

  return eventStream(request.signal, (send, close) => {
    async function run() {
      await ChatGPT.run(send, messages)
      close()
    }
    run()
    return () => {}
  })
}

I prefer the short and sweet, unless I need to do something special with those event hooks. 😁

On top of that, ChatGPT.run gets called in other places outside of SSE use cases, so it should be kept reusable. (getting summarized text or translations for a file uploaded, etc.) and in that case, just omit the func and do this:

const answer = await ChatGPT.run(messages)
await Record.create(answer)

Notice func is only called when it exists.

export class ChatGPT {
  static async run(messages = [], func?) {
    const response = await openai.chat.completions.create({
      model: 'gpt-4-turbo-preview',
      messages,
      stream: true,
    })
    let answer = ''

    for await (const chunk of response) {
      const delta = chunk.choices[0]?.delta?.content || ''
      answer += delta
      process.stdout.write(chalk.yellow(delta))
      if (func) func({ event: 'message', data: delta })
    }
    process.stdout.write('\n')

    return answer
  }
}

Even though technically you don't need stream: true in this case, LLM responses take time (up to a few minutes), so I always use streaming for better debugging experience.

[sergiodxa]

I sent a PR to add an alternative API to eventStream to accept sync functions #336