feat: add utilities for server sent events

[joshmossas]

🔗 Linked issue

#477

❓ Type of change

• 📖 Documentation (updates to the documentation, readme, or JSdoc annotations)
• 🐞 Bug fix (a non-breaking change that fixes an issue)
• 👌 Enhancement (improving an existing functionality like performance)
• ✨ New feature (a non-breaking change that adds functionality)
• 🧹 Chore (updates to the build process or auxiliary tools and libraries)
• ⚠️ Breaking change (fix or feature that would cause existing functionality to change)

📚 Description

This PR adds utilities for server sent events. This enables developers to easily create endpoints that return an event stream as outlined here

The new utilities can be used like so:

eventHandler((event) => {
  const eventStream = createEventStream(event);
  
  // start sending the stream to the client
  eventStream.start();

  // send a message every second
  const interval = setInterval(() => {
    await eventStream.push({data: 'hello world'});
  }, 1000)
  
  // cleanup when a client disconnects
  eventStream.on('disconnect', () => {
    clearInterval(interval);
  })
})

Other Changes
Unit tests have been added to tests these new features
The README.md has been updated to include the createEventStream utility in its utility list.

📝 Checklist

• I have linked an issue or discussion.
• I have updated the documentation accordingly.

[joshmossas]

One thing that I'm kind of iffy on is whether it's better to explicitly start sending the stream to the client with eventStream.start() or whether we should just automatically start sending the stream to the client when createEventStream() is called.

Right now I think being more explicit is better, but I'm open to differing opinions on this.

[arily]

just my personal thoughts.
I think start() only makes sense if

• have pause() / stop()
• flush() if you will buffer any unsent events

and these control flows should inside Apps not transports.

how about provide a utility controlling state?

const bSSE = buffered(createEventStream(event))

setInterval(() => bSSE.push({}), 100)

let started = true
setInterval(() => {
  (started = !started) ? (bSSE.start(), bSSE.flush()) : bSSE.stop()
}, 1000)

// transmit every other second

[joshmossas]

@arily thanks for the input. It's made me reevaluate how this API should be structured.

Right now start isn't a control flow thing. It's being used to send the stream to the client. I only put it in there because I think people might not want the stream to be immediately sent as the response. Technically, you can still access the stream with eventStream.stream so people could call sendStream(event, eventStream.stream) instead of start and essentially get the same result.

However this all has made me think that naming the method start might be wrong. Maybe something like eventStream.init() would be better since it doesn't have any overlap with control flow concepts such as pause and stop?

Although this has made me realize that it may be even better to make a separate utility for returning the eventStream to the client. For example something like sendEventStream(event, eventStream). This would also be in-keeping with some of the other H3 utilities. (ex: sendStream, sendWebResponse, sendNoContent, sendError, and send) so it has that added benefit.

We could also add logic allowing devs to simply return an eventStream inside a handler and H3 would just take care of it like it does with streams and web responses.

So basically the adjusted API would look like this:

// using the send utilty
eventHandler((event) => {
  const eventStream = createEventStream(event);
  setInterval(() => {
    eventStream.push("hello world");
  })
  sendEventStream(event, eventStream);
}

// using a simple return
eventHandler((event) => {
  const eventStream = createEventStream(event)

  setInterval(() => {
    eventStream.push("hello world")
  }, 1000)

  return eventStream;
})

[joshmossas]

I do like the idea of pause(), resume(), and flush() methods though. It could be worth implementing them.

[arily]

Ah sorry my wording might not be accurate, by "control flow" I mean control the start/ stop of the flow of transmitting event.

[joshmossas]

Yes I think I was following you. It was my own usage of start() as a method name that was confusing.

We need to return the eventStream to the client (which is what start() did before). Then we also need to be able to stop() and start() writing data to the stream itself, which I believe is what you were referring to.

By making a separate sendEventStream() utility I think that makes it clearer as to what is happening.

[joshmossas]

I went ahead and implemented the new API. It looks like this:

eventHandler((event) => {
  const eventStream = createEventStream(event)
  setInterval(async () => {
     await eventStream.push("Hello world")
  }, 1000)
  return eventStream
})

I think it feels more clear this way.

I also implemented the pause(), resume(), and flush(). The semantics of the exact naming can probably be changed but here's a video of the functionality in action:

Screencast.from.12-30-2023.12.42.06.AM.webm

[arily]

Very nice! Just another thought, I think for GC to properly recycle any streams we are done for, or to terminate the connection before client does, we may need a close() method too, just like other streams.
https://developer.mozilla.org/en-US/docs/Web/API/WritableStream/close
https://developer.mozilla.org/en-US/docs/Web/API/WritableStreamDefaultWriter/close

[joshmossas]

Yeah so I've already implemented an end() method which terminates the connection before the client does. I'm unsure of whether I need to manually close the stream in addition to this or not. I'll investigate further to confirm when I get the chance.

[joshmossas]

Added some changes to make cleanup easier and the event callbacks less confusing:

1. end() has been renamed to close()
   This method closes the current stream and if the stream has been sent to the client it also closes the connection by calling event.node.res.end(). This makes cleanup much easier as you only need to call one method.

2. The on() event names have been changed to make things more clear.

   Before

   eventStream.on("close", () => {});
   eventStream.on("end", () => {});

   After

   eventStream.on("close", () => {
     // the stream has been closed
   })
   eventStream.on("request:close", () => {
     // the request connection has been closed
   });

   request:close could also be renamed to connection:close if that's more clear, but it is listening to the event.node.req.on('close', () => {}) trigger, which is why I named it that way.

3. Add an autoclose parameter to the createEventStream() helper. When set to true the stream will automatically be closed when the connection is closed (by the server or the client).

   // without autoclose
   eventHandler((event) => {
     const eventStream = createEventStream(event);
     const interval = setInterval(async () => {
       await eventStream.push("hello world");
     }, 1000);
     eventStream.on("request:close", async () => {
       // both the interval and stream need to be cleaned up
       clearInterval(interval);
       await eventStream.close();
     });
     return eventStream;
   });
   
   // with autoclose
   eventHandler((event) => {
     const eventStream = createEventStream(event, true);
     const interval = setInterval(async () => {
       await eventStream.push("hello world");
     }, 1000)
     eventStream.on("close", () => {
       // only the interval needs to be cleaned up
       clearInterval(interval);
     });
     return eventStream;
   });

[joshmossas]

I've published the changes I've made here as a standalone npm package in case anyone wants to make use of the changes now
https://github.com/joshmossas/h3-sse

I'll be maintaining this repo until native SSE support comes to H3. (Either by this PR getting merged or some other change)

[codecov]

Codecov Report

Attention: Patch coverage is 61.16505% with 120 lines in your changes are missing coverage. Please review.

Project coverage is 77.62%. Comparing base (a58d7c9) to head (a5e5ef6).
Report is 18 commits behind head on main.

Files	Patch %	Lines
src/utils/sse/event-stream.ts	56.39%	75 Missing ⚠️
examples/server-sent-events.ts	0.00%	25 Missing and 1 partial ⚠️
src/utils/sse/types.ts	0.00%	17 Missing and 1 partial ⚠️
src/utils/sse/utils.ts	98.14%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #586      +/-   ##
==========================================
- Coverage   77.83%   77.62%   -0.21%     
==========================================
  Files          47       52       +5     
  Lines        4286     4836     +550     
  Branches      611      635      +24     
==========================================
+ Hits         3336     3754     +418     
- Misses        933     1063     +130     
- Partials       17       19       +2     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[pi0]

Hi dear @joshmossas thanks so much for putting all efforts on this!

I quickly checked implementation it is flawless and nice! I have made few refactors out of my instincts in order to move this forward as fast as possible.

• Utils are refactored to sse/ with just one createEventStream util exposed
• Usage is return eventStream.send() in the end. I know it is less intuitive. This allows to move faster without adding direct handlers to the core matcher for now. Eventually we might do this.
• I have enabled autoclose by default (haven't found a reason why not to) and it can be opted-out with { autoClose: false } option.
• Close hook is via onClosed and registers both handles. I found it easier to recommend people use one trustable hook for cleanup.
• Docs are autogenerated from jsdocs into advanced section, later I might add it to better place below WebSocket section.
• Util marked as @experimental mainly because platform support is not tested but I'm positive it is a really good implementation.

Thinking to merge and try on nightly but please consider my changes blind changes and feel free to suggest any amends or make subsequent PRs.

[joshmossas]

Awesome thank you so much @pi0 . I was honestly thinking of enabling autoclose by default as well, so I'm glad you thought the same.

I agree onClosed being the only hook for cleanup is much nicer. My only concern is that means it'll fire twice correct? Once for the writer closing and once for the request closing. I think that could prolly lead to some unexpected side-effects.

I suppose the solution is to make the internals for onClosed a bit smarter. I'll have to think about what exactly that would look like though cuz there's prolly some iffy edge cases.

[joshmossas]

Or maybe we just make it so onClosed doesn't register a callback on this._h3Event.node.req.on('close') at all.

Here me out. The default is now the writer will automatically close when the request closes. So if someone chooses to opt out of autoclose they have already opted to do more work themselves anyway. So maybe in those cases we just tell them they need to register a hook on event.node.req.on('close') if they want to respond to the request. This is what you normally have to do anyway so I don't think i's actually that big a deal.

That also means the onClose hook for EventStreams is only related to the stream closing, which is actually less confusing if you think about it.

Making autoclose default actually simplified the whole thought process around cleanup and hooks now that I think about it.