Struct rocket::response::stream::EventStream

pub struct EventStream<S> { /* private fields */ }

A potentially infinite stream of Server-Sent Events (SSE).

An EventStream can be constructed from any Stream of items of type Event. The stream can be constructed directly via EventStream::from() or through generator syntax via EventStream!.

Responder

EventStream is a (potentially infinite) responder. The response Content-Type is set to EventStream. The body is unsized, and values are sent as soon as they are yielded by the internal iterator.

Heartbeat

A heartbeat comment is injected into the internal stream and sent at a fixed interval. The comment is discarded by clients and serves only to keep the connection alive; it does not interfere with application data. The interval defaults to 30 seconds but can be adjusted with EventStream::heartbeat().

Examples

Use EventStream! to yield an infinite series of “ping” SSE messages to the client, one per second:

use rocket::response::stream::{Event, EventStream};;
use rocket::tokio::time::{self, Duration};

#[get("/events")]
fn stream() -> EventStream![] {
    EventStream! {
        let mut interval = time::interval(Duration::from_secs(1));
        loop {
            yield Event::data("ping");
            interval.tick().await;
        }
    }
}

Yield 9 events: 3 triplets of retry, data, and comment events:

use rocket::response::stream::{Event, EventStream};
use rocket::tokio::time::Duration;

#[get("/events")]
fn events() -> EventStream![] {
    EventStream! {
        for i in 0..3 {
            yield Event::retry(Duration::from_secs(10));
            yield Event::data(format!("{}", i)).id("cat").event("bar");
            yield Event::comment("silly boy");
        }
    }
}

The syntax of EventStream! as an expression is identical to that of stream!. For how to gracefully terminate an otherwise infinite stream, see graceful shutdown.

Borrowing

If an EventStream contains a borrow, the extended type syntax EventStream![Event + '_] must be used:

use rocket::State;
use rocket::response::stream::{Event, EventStream};

#[get("/events")]
fn events(ctxt: &State<bool>) -> EventStream![Event + '_] {
    EventStream! {
        // By using `ctxt` in the stream, the borrow is moved into it. Thus,
        // the stream object contains a borrow, prompting the '_ annotation.
        if *ctxt.inner() {
            yield Event::data("hi");
        }
    }
}

See stream#borrowing for further details on borrowing in streams.

Pitfalls

Server-Sent Events are a rather simple mechanism, though there are some pitfalls to be aware of.

• Buffering

  Protocol restrictions complicate implementing an API that does not buffer. As such, if you are sending lots of data, consider sending the data via multiple data fields (with events to signal start and end). Alternatively, send one event which instructs the client to fetch the data from another endpoint which in-turn streams the data.

• Raw SSE requires UTF-8 data

  Only UTF-8 data can be sent via SSE. If you need to send arbitrary bytes, consider encoding it, for instance, as JSON using Event::json(). Alternatively, as described before, use SSE as a notifier which alerts the client to fetch the data from elsewhere.

• Raw SSE is Lossy

  Data sent via SSE cannot contain new lines \n or carriage returns \r due to interference with the line protocol.

  The protocol allows expressing new lines as multiple messages, however, and Rocket automatically transforms a message of foo\nbar into two messages, foo and bar, so that they are reconstructed (automatically) as foo\nbar on the client-side. For messages that only contain new lines \n, the conversion is lossless.

  However, the protocol has no mechanism for expressing carriage returns and thus it is not possible to send unencoded carriage returns via SSE. Rocket handles carriage returns like it handles new lines: it splits the data into multiple messages. Thus, a sequence of \r\n becomes \n at the client side. A single \r that is not part of an \r\n sequence also becomes \n at the client side. As a result, the message foo\r\nbar\rbaz is read as foo\nbar\nbaz at the client-side.

  To send messages losslessly, they must be encoded first, for instance, by using Event::json().

Implementations

impl<S: Stream<Item = Event>> EventStream<S>

pub fn heartbeat<H: Into<Option<Duration>>>(self, heartbeat: H) -> Self

Sets a “ping” interval for this EventStream to avoid connection timeouts when no data is being transferred. The default interval is 30 seconds.

The ping is implemented by sending an empty comment to the client every interval seconds.

Example

use rocket::response::stream::{Event, EventStream};
use rocket::tokio::time::Duration;

#[get("/events")]
fn events() -> EventStream![] {
    // Remove the default heartbeat.
    EventStream::from(event_stream).heartbeat(None);

    // Set the heartbeat interval to 15 seconds.
    EventStream::from(event_stream).heartbeat(Duration::from_secs(15));

    // Do the same but for a generated `EventStream`:
    let stream = EventStream! {
        yield Event::data("hello");
    };

    stream.heartbeat(Duration::from_secs(15))
}

Trait Implementations

impl<S: Stream<Item = Event>> From<S> for EventStream<S>

fn from(stream: S) -> Self

Creates an EventStream from a [Stream] of Events.

Use EventStream::from() to construct an EventStream from an already existing stream. Otherwise, prefer to use EventStream!.

Example

use rocket::response::stream::{Event, EventStream};
use rocket::futures::stream;

let raw = stream::iter(vec![Event::data("a"), Event::data("b")]);
let stream = EventStream::from(raw);

impl<'r, S: Stream<Item = Event> + Send + 'r> Responder<'r, 'r> for EventStream<S>

fn respond_to(self, _: &'r Request<'_>) -> Result<'r>

Returns Ok if a Response could be generated successfully. Otherwise, returns an Err with a failing Status.