Struct rocket::response::stream::EventStream[][src]

pub struct EventStream<S> { /* fields omitted */ }
Expand description

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.

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>[src]

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

Sets a “ping” interval for this EventStream to avoid connection timeouts when no data is being transferred. The default interval for a newly created EventStream is None, which disables this functionality.

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>[src]

fn from(stream: S) -> Self[src]

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>[src]

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

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

Auto Trait Implementations

impl<S> RefUnwindSafe for EventStream<S> where
    S: RefUnwindSafe

impl<S> Send for EventStream<S> where
    S: Send

impl<S> Sync for EventStream<S> where
    S: Sync

impl<S> Unpin for EventStream<S> where
    S: Unpin

impl<S> UnwindSafe for EventStream<S> where
    S: UnwindSafe

Blanket Implementations

impl<T> Any for T where
    T: 'static + ?Sized
[src]

pub fn type_id(&self) -> TypeId[src]

Gets the TypeId of self. Read more

impl<T> Borrow<T> for T where
    T: ?Sized
[src]

pub fn borrow(&self) -> &T[src]

Immutably borrows from an owned value. Read more

impl<T> BorrowMut<T> for T where
    T: ?Sized
[src]

pub fn borrow_mut(&mut self) -> &mut T[src]

Mutably borrows from an owned value. Read more

impl<T> From<T> for T[src]

pub fn from(t: T) -> T[src]

Performs the conversion.

impl<T> Instrument for T[src]

fn instrument(self, span: Span) -> Instrumented<Self>[src]

Instruments this type with the provided Span, returning an Instrumented wrapper. Read more

fn in_current_span(self) -> Instrumented<Self>[src]

Instruments this type with the current Span, returning an Instrumented wrapper. Read more

impl<T, U> Into<U> for T where
    U: From<T>, 
[src]

pub fn into(self) -> U[src]

Performs the conversion.

impl<T> IntoCollection<T> for T

pub fn into_collection<A>(self) -> SmallVec<A> where
    A: Array<Item = T>, 

Converts self into a collection.

pub fn mapped<U, F, A>(self, f: F) -> SmallVec<A> where
    F: FnMut(T) -> U,
    A: Array<Item = U>, 

impl<T> Same<T> for T

type Output = T

Should always be Self

impl<T, U> TryFrom<U> for T where
    U: Into<T>, 
[src]

type Error = Infallible

The type returned in the event of a conversion error.

pub fn try_from(value: U) -> Result<T, <T as TryFrom<U>>::Error>[src]

Performs the conversion.

impl<T, U> TryInto<U> for T where
    U: TryFrom<T>, 
[src]

type Error = <U as TryFrom<T>>::Error

The type returned in the event of a conversion error.

pub fn try_into(self) -> Result<U, <U as TryFrom<T>>::Error>[src]

Performs the conversion.

impl<V, T> VZip<V> for T where
    V: MultiLane<T>, 

pub fn vzip(self) -> V