Support for server-sent events

[fitzgen]

Feature Request

Support for server-sent events in Tide apps.

Detailed Description

Support for working creating server-sent event streams, easily, with Tide.

Some more general server-sent events info:

• Standard: https://html.spec.whatwg.org/multipage/#server-sent-events
  • Grammar: https://html.spec.whatwg.org/multipage/server-sent-events.html#parsing-an-event-stream
• Using Server-Sent Events: https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events

Context

I would like to use server-sent events. I suppose other folks might want to as well.

Possible Implementation

Ideally, I would be able to have some method on Response or some sort of type that implements IntoResponse that sets the Content-Type: text/event-stream header for me, and takes a Stream<Item = ServerSentEvent> where ServerSentEvent is a struct with an optional event type string and a data string.

Maybe it would automatically send heartbeats too, to keep the connection alive?

[yoshuawuyts]

Reference from a Node.js implementation I did a while back:

Refs

• https://github.com/choojs/bankai/blob/7afacfc2cbf3913035b8c744b0195c6b7bfe3159/http.js#L132-L142
• https://github.com/choojs/bankai/blob/7afacfc2cbf3913035b8c744b0195c6b7bfe3159/http.js#L170

Headers

      'Content-Type': 'text/event-stream',
      'X-Accel-Buffering': 'no',
      'Cache-Control': 'no-cache'

heartbeat

  var interval = setInterval(function () {
      if (res.finished) return // prevent writes after stream has closed
      res.write(`id:${id++}\ndata:{ "type:": "heartbeat" }\n\n`)
}, 4000)

data

In particular making sure to increment the id on every tick is useful because Chrome's debug tooling picks up on this and does a better job at displaying individual events in the network tab.

var msg = JSON.stringify({ type: 'scripts' })
res.write(`id:${id++}\ndata:${msg}\n\n`)

Screenshot

What it looks like to run this:

[pickfire]

I believe people can just swap ServerSentEvent with the data they desired from Stream<Item = ServerSentEvent>.

In some cases, it may be better to return JSON over plain text, I wonder if Stream<Item = Json<Data>> helps?

By the way, I believe that this should be marked as design where it may be similar to #67.

[yoshuawuyts]

@pickfire the output stream should probably take an AsyncRead and produce an AsyncRead. It can then decorate the incoming messages with the right SSE framing, and also produce regular heartbeat messages.

In general we should probably move the http-service interface to use AsyncRead for both requests and responses rather than the existing pre-allocated streams that we're exposing.

[yoshuawuyts]

API

I've sketched out an API we could use for Tide:

use async_std::task;

fn main() -> Result<(), std::io::Error> {
    task::block_on(async {
        let mut app = tide::new();
        app.at("/sse").get(tide::sse()); // shorthand function, we might also want to provide a builder
        app.at("/", async |req| {
            let mut sse = req.sse();
            println!("next message id: {}", sse.id());      // u64
            println!("sse status is: {}", sse.status());    // enum { Connecting, Open, Closed }
            println!("sse is open?: {}", sse.is_open());    // bool
            sse.send("message", b"hello world").await;      // send(&self, event: &str, data: &[u8])
            "hello world" // return some response
        });
        app.listen("127.0.0.1:8080").await?;
        Ok(())
    })
}

A single connection would be established per peer (when they make an EventSource connection to /sse in the example), and from there we'd keep it open with a keepalive until the client drops us.

The connection would be stored inside the app instance, and accessible as req.sse() for other connections from the same host. This has some security implications though (see next section).

We would probably also want to provide extra metadata on the sse object on Request. Info such as whether the connection is open, message id, etc. might be useful.

Security

We can't base the host purely on incoming TCP request since we could be behind a proxy. Instead we should be aware of x-forwarded-for and forwarded headers, and mention this as part of the security model.

References

This model is inspired by Phoenix's channels, which have a similar model. The main difference being that we don't take care of the multiplexing aspects, but instead leave it up to the application author to decide how they want to use the connection. Phoenix provides a matching client library to handle the "channel", and I think we probably shouldn't do that.

• https://developer.mozilla.org/en-US/docs/Web/API/EventSource
• https://html.spec.whatwg.org/multipage/server-sent-events.html#the-eventsource-interface
• https://docs.rs/warp/0.2.0/warp/filters/sse/index.html

---

cc/ @goto-bus-stop I'm curious what you think of this API

[pickfire]

        app.at("/sse").get(tide::sse()); // shorthand function, we might also want to provide a builder

Looking at that, seemed like it should only expected to have one server-sent events per server. I believe SSE is usually one per server but could there be multiple endpoints for SSE?

How about?

        app.at("/", async |req| {
            // assuming connection does not close in the block (not sure what happens if it does)
            // enum { Connecting, Open(SSEConnection), Closed }
            if let Open(mut sse) = req.sse() {
                println!("next message id: {}", sse.id()); // u64
                sse.send("message", b"hello world").await; // send(&self, event: &str, data: &[u8])
            }
            "hello world" // return some response
        });

[yoshuawuyts]

@pickfire the /sse endpoint would be for clients to connect on; there could be multiple endpoints for this, but I'm quite sure clients should only have a single SSE connection per server in most cases.

I like your idea of wrapping the SSE object inside an option; that might perhaps be the best way to go about it .

[pickfire]

Should the send return a result? Because there is no saying that SSE connection will be dropped while inside the block.

        app.at("/", async |req| {
            // assuming connection does not close in the block (not sure what happens if it does)
            // enum { Connecting, Open(SSEConnection), Closed }
            if let Open(mut sse) = req.sse() {
                println!("next message id: {}", sse.id()); // u64
                // <- sse connection dropped here
                sse.send("message", b"hello world").await?; // result?
            }
            "hello world" // return some response
        });

I wonder if we should even have that block in the first place, no saying when will the connection will be dropped.

By the way, Happy Chinese New Year! ^^

[yoshuawuyts]

@pickfire Happy new year to you too! I'm thinking we should also add a try_send method that returns immediately if the connection isn't open. Related issue: async-rs/async-std#585.

I've been thinking about what you said earlier by the way, and actually I think it might be susceptible to race conditions. What happens if we need to send a message over SSE, but this message might be sent after the connection has been established.

In fact I think we should support several different scenarios; and probably returning an Option<SSE> from req.sse() is not as flexible as we should want it to be. Instead I think the original design might be a bit better suited here:

Examples

// Wait until we can send the message.
app.at("/", async |req| {
    req.sse().send("message", b"hello world").await;
    Response::new(200)
});

// Wait until we can send the message, but timeout after 5 secs.
app.at("/", async |req| {
    req.sse()
        .send("message", b"hello world")
        .timeout(Duration::from_secs(5))
        .await;
    Response::new(200)
});

// Error if we can't send the message immediately. This includes
// a full queue.
app.at("/", async |req| {
    req.sse().try_send("message", b"hello world")?.await;
    Response::new(200)
});

// Only send a message if we're already connected.
app.at("/", async |req| {
    let sse = req.sse();
    if sse.is_connected() {
        req.sse().send("message", b"hello world").await;
    }
    Response::new(200)
});

[pickfire]

The other examples seems fair enough but looking at the fourth example.

// Only send a message if we're already connected.
app.at("/", async |req| {
    let sse = req.sse();
    if sse.is_connected() {
        req.sse().send("message", b"hello world").await;
    }
    Response::new(200)
});

It may be possible that sse connection maybe dropped just right after the check on is_connected?

[yoshuawuyts]

@pickfire accurate; it could be a temporary hiccup and the client could reconnect. Or perhaps it's an actual timeout and we should timeout the whole request. We can't distinguish between these cases. Either way, waiting seems like the right behavior.

[pickfire]

Or is it possible that we could return a response first and then wait to send the SSE?

[yoshuawuyts]

@pickfire That's a possibility, though I see there might be difficulty in finding the right peer -- so it would need to take some context from Request, and use it for Response. That might make for a difficult mode.

I spent some time this week writing about the APIs here by the way: https://blog.yoshuawuyts.com/tide-channels/.

[goto-bus-stop]

That's an interesting API direction—not something I had thought of! I'll write down some of my thoughts based mostly on an actual realtime app I have that's currently using WebSockets, but that could just as well use EventSource. I'm not sure that this is a typical app so some of the features I'd like may not be that important in general.

e; i wrote most of this a couple days ago so i hadn't seen the post yet 🙃 it looks like that addresses a lot of the same things, which is prob a good sign!

I think the most common case is having one SSE endpoint, but it can make sense to have multiple for different types of resources. Or something like a /thread/:id/posts-stream on a forum software, where the number of endpoints is not known.

One thing is kinda unclear to me in the proposed API (e; from the OP, but addressed in the blog post). If user A and B are connected to the SSE endpoint, and user B sends a request to /, what happens? Is this event broadcasted to both, is it only sent to A, or only sent to B?

There's a couple scenarios where you might need to send something over an SSE connection. Sometimes, you need to broadcast something to all clients, or to a subset, or to just one. Sometimes this may be directly in response to an HTTP request (then req.sse() is useful), but it could also be in response to any other event, like getting a new message from a message queue. Some of that could be achieved by allowing SSE connection instances to subscribe to a "topic", and then you can publish to that topic anywhere else in the app (whether in response to an HTTP request or anything else). An app could choose to subscribe to like a "globalChat" and a "$myUserId" topic to address the diffirent granularities. I think something like that would probably be implemented with a bunch of channels in Rust, and it could be mostly independent of the actual HTTP/SSE implementation. There might be a crate out there that does it already...

I think @fitzgen's proposed API is a good starting point regardless of what a higher level API would look like. If/when tide settles on a higher level API, having an escape hatch for the edge cases would still be valuable.

I think a higher level API could/should also attempt to make Last-Event-ID "just work", which would involve keeping some messages queued after a connection is lost. I don't know if this fits in very well with Tide because you might want to store your messages elsewhere (I store them in Redis in my app to be sorta resilient to server crashes and restarts.) Without giving much thought to implementation, I think abstraction layers could look something like:

• IntoResponse for a stream of SSE messages
• "Connection" abstraction that can handle reconnects with Last-Event-ID
• Some pub/sub thing (and I think this could be where a tide::sse() comes in)