siderite_core/sse/
mod.rs

1//! Server-sent events: [`Event`] encodes one event of the
2//! `text/event-stream` format, and [`Sse`] streams events as a response.
3//!
4//! - Data may hold any text: each line becomes its own `data:` line
5//!   (`\n`, `\r\n` and `\r` all end a line), so a client receives the
6//!   text back with `\n` line ends.
7//! - Event names and ids are single lines: a CR or LF in them (or a NUL in
8//!   an id, which clients ignore) is rejected with [`SseError`] instead of
9//!   splitting the event.
10//! - Responses carry `Content-Type: text/event-stream` and
11//!   `Cache-Control: no-store`, so no cache stores or replays a stream.
12//! - An `Err` from the event stream ends the response body abruptly; the
13//!   error itself is never sent (clients see the connection end and may
14//!   reconnect). Send an event first if the client should know why.
15//! - Events are pulled only as fast as the connection takes them, so a
16//!   slow reader never makes the stream buffer events (feed it from a
17//!   bounded channel, never an unbounded one). A client that disconnects
18//!   drops the stream, and with it the work behind it.
19//! - [`Sse::keep_alive`] sends a comment after an idle interval, so
20//!   proxies and clients do not close a quiet connection;
21//!   [`Sse::until`] ends the stream when a future resolves, such as
22//!   `Readiness::stopping()`, so graceful shutdown need not time out.
23
24use crate::response::{IntoResponse, Response, SUCCESS};
25use crate::responses::StreamingResponse;
26use futures_util::Stream;
27use http::HeaderValue;
28use http::header::CACHE_CONTROL;
29use lifecycle::{Lifecycle, Until};
30use serde::Serialize;
31use siderite_openapi::{Operation, SchemaObject, SchemaRegistry};
32use std::error::Error as StdError;
33use std::fmt::Write;
34use std::future::Future;
35use std::time::Duration;
36use thiserror::Error;
37
38/// The media type of an event stream.
39pub const EVENT_STREAM: &str = "text/event-stream";
40
41/// A field value that would break the event framing.
42#[derive(Debug, Clone, PartialEq, Eq, Error)]
43pub enum SseError {
44    /// An event name containing a line break.
45    #[error("SSE event names must not contain CR or LF")]
46    EventName,
47    /// An id containing a line break or NUL.
48    #[error("SSE ids must not contain CR, LF or NUL")]
49    Id,
50}
51
52/// One server-sent event.
53#[derive(Debug, Clone, Default, PartialEq, Eq)]
54pub struct Event {
55    comment: Option<String>,
56    event: Option<String>,
57    id: Option<String>,
58    retry: Option<Duration>,
59    data: Option<String>,
60}
61
62impl Event {
63    /// An event carrying `text` (any text; line breaks are kept).
64    pub fn data(text: impl Into<String>) -> Self {
65        Self {
66            data: Some(text.into()),
67            ..Self::default()
68        }
69    }
70
71    /// An event carrying `value` as JSON.
72    ///
73    /// # Errors
74    /// Serialization errors of `value`.
75    pub fn json<T: Serialize>(value: &T) -> Result<Self, serde_json::Error> {
76        Ok(Self::data(serde_json::to_string(value)?))
77    }
78
79    /// A comment only (`: text`), which clients ignore: useful to keep a
80    /// connection active.
81    pub fn comment(text: impl Into<String>) -> Self {
82        Self {
83            comment: Some(text.into()),
84            ..Self::default()
85        }
86    }
87
88    /// Name the event (the client's `addEventListener` type).
89    ///
90    /// # Errors
91    /// [`SseError::EventName`] when `name` contains CR or LF.
92    pub fn event(mut self, name: impl Into<String>) -> Result<Self, SseError> {
93        let name = name.into();
94        if name.contains(['\r', '\n']) {
95            return Err(SseError::EventName);
96        }
97        self.event = Some(name);
98        Ok(self)
99    }
100
101    /// Set the id clients send back as `Last-Event-ID` when reconnecting.
102    ///
103    /// # Errors
104    /// [`SseError::Id`] when `id` contains CR, LF or NUL.
105    pub fn id(mut self, id: impl Into<String>) -> Result<Self, SseError> {
106        let id = id.into();
107        if id.contains(['\r', '\n', '\0']) {
108            return Err(SseError::Id);
109        }
110        self.id = Some(id);
111        Ok(self)
112    }
113
114    /// Ask clients to wait `delay` before reconnecting (whole
115    /// milliseconds).
116    #[must_use]
117    pub fn retry(mut self, delay: Duration) -> Self {
118        self.retry = Some(delay);
119        self
120    }
121
122    /// The event in `text/event-stream` form, ending with a blank line.
123    pub fn encode(&self) -> String {
124        let mut out = String::new();
125        if let Some(comment) = &self.comment {
126            for line in lines(comment) {
127                let _ = writeln!(out, ": {line}");
128            }
129        }
130        if let Some(event) = &self.event {
131            let _ = writeln!(out, "event: {event}");
132        }
133        if let Some(id) = &self.id {
134            let _ = writeln!(out, "id: {id}");
135        }
136        if let Some(retry) = self.retry {
137            let _ = writeln!(out, "retry: {}", retry.as_millis());
138        }
139        if let Some(data) = &self.data {
140            for line in lines(data) {
141                let _ = writeln!(out, "data: {line}");
142            }
143        }
144        out.push('\n');
145        out
146    }
147}
148
149/// The lines of `text`, split at `\r\n`, `\r` or `\n`; an empty text is
150/// one empty line.
151fn lines(text: &str) -> Vec<&str> {
152    let mut out = Vec::new();
153    let mut rest = text;
154    loop {
155        match rest.find(['\r', '\n']) {
156            Some(at) => {
157                out.push(&rest[..at]);
158                let skip = if rest[at..].starts_with("\r\n") { 2 } else { 1 };
159                rest = &rest[at + skip..];
160            }
161            None => {
162                out.push(rest);
163                return out;
164            }
165        }
166    }
167}
168
169/// A `text/event-stream` response from a stream of events.
170pub struct Sse<S> {
171    stream: S,
172    keep_alive: Option<Duration>,
173    until: Option<Until>,
174}
175
176impl<S, E> Sse<S>
177where
178    S: Stream<Item = Result<Event, E>> + Send + 'static,
179    E: Into<Box<dyn StdError + Send + Sync>> + Send + 'static,
180{
181    /// Stream `events`; an `Err` ends the response (see the module docs).
182    pub fn new(events: S) -> Self {
183        Self {
184            stream: events,
185            keep_alive: None,
186            until: None,
187        }
188    }
189
190    /// Send a comment (`:`) whenever no event was sent for `interval`.
191    #[must_use]
192    pub fn keep_alive(mut self, interval: Duration) -> Self {
193        self.keep_alive = Some(interval);
194        self
195    }
196
197    /// End the stream cleanly when `signal` resolves, for example
198    /// `readiness.stopping()` so streams close as shutdown begins.
199    #[must_use]
200    pub fn until<F>(mut self, signal: F) -> Self
201    where
202        F: Future<Output = ()> + Send + 'static,
203    {
204        self.until = Some(Box::pin(signal));
205        self
206    }
207}
208
209impl<S, E> IntoResponse for Sse<S>
210where
211    S: Stream<Item = Result<Event, E>> + Send + 'static,
212    E: Into<Box<dyn StdError + Send + Sync>> + Send + 'static,
213{
214    fn into_response(self) -> Response {
215        let bytes = Lifecycle::new(self.stream, self.keep_alive, self.until);
216        let stream = StreamingResponse::new(bytes).content_type(EVENT_STREAM);
217        let mut response = stream.into_response();
218        let no_store = HeaderValue::from_static("no-store");
219        response.headers_mut().insert(CACHE_CONTROL, no_store);
220        response
221    }
222
223    fn describe(op: &mut Operation, _registry: &mut SchemaRegistry) {
224        let schema = SchemaObject::of_type("string");
225        let content = Some((EVENT_STREAM, schema));
226        op.add_response(SUCCESS, "Server-sent events", content);
227    }
228}
229
230mod lifecycle;
231
232#[cfg(test)]
233mod tests;
siderite_core/sse/mod.rs

siderite_core/sse/
mod.rs
