trillium_client/
sse.rs

1//! Client-side [Server-Sent Events][spec].
2//!
3//! [`Conn::into_sse`] executes a request and interprets the response body as an
4//! `text/event-stream`, yielding a [`Stream`] of [`Event`]s. Unlike
5//! [`into_websocket`][Conn::into_websocket], this is not a protocol upgrade — an event stream is
6//! an ordinary response whose body is read incrementally and parsed line-by-line. It works
7//! identically over HTTP/1.x, HTTP/2, and HTTP/3.
8//!
9//! This is a single-response stream: it ends when the connection closes. It does **not**
10//! implement the [`EventSource`][es] automatic-reconnection behavior (re-issuing the request with
11//! `Last-Event-ID` and honoring server `retry:` timing), which only makes sense for idempotent
12//! event feeds. To retry a dropped request, drive the whole request through a retrying
13//! [`ClientHandler`][crate::ClientHandler].
14//!
15//! [spec]: https://html.spec.whatwg.org/multipage/server-sent-events.html
16//! [es]: https://developer.mozilla.org/en-US/docs/Web/API/EventSource
17
18use crate::Conn;
19use futures_lite::{AsyncRead, stream::Stream};
20use std::{
21    collections::VecDeque,
22    error::Error,
23    fmt::{self, Debug, Display, Formatter},
24    ops::{Deref, DerefMut},
25    pin::Pin,
26    task::{Context, Poll, ready},
27    time::Duration,
28};
29use trillium_http::{KnownHeaderName, Status};
30
31const READ_BUF_LEN: usize = 8 * 1024;
32
33impl Conn {
34    /// Execute this request and interpret the response body as a [Server-Sent Events][spec]
35    /// stream.
36    ///
37    /// This is an *execution* method: it sends the request (setting `Accept: text/event-stream`
38    /// if not already present), then validates that the response has a success status and a
39    /// `text/event-stream` content-type before handing back an [`EventStream`]. Calling it on a
40    /// conn that has already been awaited returns [`SseErrorKind::AlreadyExecuted`] — build the
41    /// conn, then call this; don't await it yourself first.
42    ///
43    /// On any failure the returned [`SseError`] still carries the [`Conn`], so the caller can
44    /// inspect the response (status, headers, error body) or convert it back with
45    /// [`From`]/[`Into`].
46    ///
47    /// [spec]: https://html.spec.whatwg.org/multipage/server-sent-events.html
48    pub async fn into_sse(mut self) -> Result<EventStream, SseError> {
49        if self.status().is_some() {
50            return Err(SseError::new(self, SseErrorKind::AlreadyExecuted));
51        }
52
53        self.request_headers_mut()
54            .try_insert(KnownHeaderName::Accept, "text/event-stream");
55
56        if let Err(e) = (&mut self).await {
57            return Err(SseError::new(self, e.into()));
58        }
59
60        let status = self.status().expect("Response did not include status");
61        if !status.is_success() {
62            return Err(SseError::new(self, SseErrorKind::Status(status)));
63        }
64
65        if !is_event_stream(
66            self.response_headers()
67                .get_str(KnownHeaderName::ContentType),
68        ) {
69            let content_type = self
70                .response_headers()
71                .get_str(KnownHeaderName::ContentType)
72                .map(String::from);
73            return Err(SseError::new(
74                self,
75                SseErrorKind::UnexpectedContentType(content_type),
76            ));
77        }
78
79        Ok(EventStream::new(self))
80    }
81}
82
83/// True if `content_type` names the `text/event-stream` media type, ignoring any parameters
84/// (e.g. `; charset=utf-8`) and ASCII case.
85fn is_event_stream(content_type: Option<&str>) -> bool {
86    content_type.is_some_and(|ct| {
87        ct.split(';')
88            .next()
89            .is_some_and(|media_type| media_type.trim().eq_ignore_ascii_case("text/event-stream"))
90    })
91}
92
93/// A single server-sent event.
94///
95/// Field accessors follow the [SSE specification][spec]: [`event_type`](Event::event_type) is
96/// `None` for the default `message` type, [`data`](Event::data) has had its lines joined with
97/// `\n` and the trailing newline removed, and [`id`](Event::id) reflects the most recent `id:`
98/// field seen on the stream (it persists across events, matching `EventSource.lastEventId`).
99///
100/// [spec]: https://html.spec.whatwg.org/multipage/server-sent-events.html#event-stream-interpretation
101#[derive(Debug, Clone, Eq, PartialEq)]
102pub struct Event {
103    data: String,
104    event_type: Option<String>,
105    id: Option<String>,
106    retry: Option<Duration>,
107}
108
109impl Event {
110    /// The event payload, with multiple `data:` lines joined by `\n`.
111    #[must_use]
112    pub fn data(&self) -> &str {
113        &self.data
114    }
115
116    /// The event type from the `event:` field, or `None` for the default `message` type.
117    #[must_use]
118    pub fn event_type(&self) -> Option<&str> {
119        self.event_type.as_deref()
120    }
121
122    /// The last event id seen on the stream up to and including this event.
123    #[must_use]
124    pub fn id(&self) -> Option<&str> {
125        self.id.as_deref()
126    }
127
128    /// The server-requested reconnection time from a `retry:` field, if one preceded this event.
129    ///
130    /// This is a connection-level directive; because [`EventStream`] does not reconnect, it is
131    /// surfaced purely informationally for callers that implement their own reconnection.
132    #[must_use]
133    pub fn retry(&self) -> Option<Duration> {
134        self.retry
135    }
136}
137
138/// A [`Stream`] of [`Event`]s decoded from a `text/event-stream` response body.
139///
140/// Created by [`Conn::into_sse`]. The stream yields `Result<Event, trillium_http::Error>`; an
141/// error item is an IO failure reading the underlying transport, after which the stream ends.
142/// The stream ends with `None` when the connection closes; an incomplete event at end-of-stream
143/// (no terminating blank line) is discarded per the specification.
144#[derive(Debug)]
145pub struct EventStream {
146    conn: Conn,
147    decoder: Decoder,
148    pending: VecDeque<Event>,
149    read_buf: Box<[u8]>,
150    done: bool,
151}
152
153impl EventStream {
154    fn new(conn: Conn) -> Self {
155        Self {
156            conn,
157            decoder: Decoder::default(),
158            pending: VecDeque::new(),
159            read_buf: vec![0; READ_BUF_LEN].into_boxed_slice(),
160            done: false,
161        }
162    }
163
164    /// The executed [`Conn`] this stream was created from, for response metadata — status,
165    /// response headers, peer address.
166    pub fn conn(&self) -> &Conn {
167        &self.conn
168    }
169}
170
171impl Stream for EventStream {
172    type Item = trillium_http::Result<Event>;
173
174    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
175        let this = self.get_mut();
176        loop {
177            if let Some(event) = this.pending.pop_front() {
178                return Poll::Ready(Some(Ok(event)));
179            }
180            if this.done {
181                return Poll::Ready(None);
182            }
183
184            let mut response_body = this.conn.response_body();
185            match ready!(Pin::new(&mut response_body).poll_read(cx, &mut this.read_buf)) {
186                // EOF: a trailing event without its blank line is discarded per spec.
187                Ok(0) => {
188                    this.done = true;
189                    return Poll::Ready(None);
190                }
191                Ok(n) => this.decoder.push(&this.read_buf[..n], &mut this.pending),
192                Err(e) => {
193                    this.done = true;
194                    return Poll::Ready(Some(Err(e.into())));
195                }
196            }
197        }
198    }
199}
200
201/// Incremental, allocation-reusing parser for the SSE wire format.
202///
203/// Bytes are fed in arbitrary chunks via [`push`](Decoder::push); completed [`Event`]s are
204/// appended to the caller's queue. Line terminators (CR, LF, CRLF) are handled across chunk
205/// boundaries via `last_char_was_cr`.
206#[derive(Debug, Default)]
207struct Decoder {
208    line: Vec<u8>,
209    last_char_was_cr: bool,
210    data: String,
211    event_type: Option<String>,
212    id: Option<String>,
213    retry: Option<Duration>,
214    has_data: bool,
215}
216
217impl Decoder {
218    fn push(&mut self, bytes: &[u8], out: &mut VecDeque<Event>) {
219        for &byte in bytes {
220            match byte {
221                b'\r' => {
222                    self.line_done(out);
223                    self.last_char_was_cr = true;
224                }
225                b'\n' if self.last_char_was_cr => self.last_char_was_cr = false,
226                b'\n' => self.line_done(out),
227                _ => {
228                    self.last_char_was_cr = false;
229                    self.line.push(byte);
230                }
231            }
232        }
233    }
234
235    fn line_done(&mut self, out: &mut VecDeque<Event>) {
236        if self.line.is_empty() {
237            self.dispatch(out);
238        } else {
239            let mut line = std::mem::take(&mut self.line);
240            self.process_field(&line);
241            line.clear();
242            self.line = line;
243        }
244    }
245
246    fn process_field(&mut self, line: &[u8]) {
247        let (field, value) = match memchr::memchr(b':', line) {
248            Some(0) => return, // leading colon: comment
249            Some(colon) => {
250                let value = &line[colon + 1..];
251                let value = value.strip_prefix(b" ").unwrap_or(value);
252                (&line[..colon], value)
253            }
254            None => (line, &b""[..]),
255        };
256
257        match field {
258            b"event" => self.event_type = Some(String::from_utf8_lossy(value).into_owned()),
259
260            b"data" => {
261                self.data.push_str(&String::from_utf8_lossy(value));
262                self.data.push('\n');
263                self.has_data = true;
264            }
265
266            b"id" => {
267                if !value.contains(&0) {
268                    self.id = Some(String::from_utf8_lossy(value).into_owned());
269                }
270            }
271
272            b"retry" => {
273                if !value.is_empty()
274                    && value.iter().all(u8::is_ascii_digit)
275                    && let Ok(ms) = std::str::from_utf8(value).unwrap_or_default().parse()
276                {
277                    self.retry = Some(Duration::from_millis(ms));
278                }
279            }
280
281            _ => {}
282        }
283    }
284
285    fn dispatch(&mut self, out: &mut VecDeque<Event>) {
286        if !self.has_data {
287            // No data accumulated: reset the data and event-type buffers without dispatching,
288            // but leave `id` (last-event-id) and any pending `retry` intact, per spec.
289            self.data.clear();
290            self.event_type = None;
291            return;
292        }
293
294        if self.data.ends_with('\n') {
295            self.data.pop();
296        }
297
298        out.push_back(Event {
299            data: std::mem::take(&mut self.data),
300            event_type: self.event_type.take().filter(|s| !s.is_empty()),
301            id: self.id.clone(),
302            retry: self.retry.take(),
303        });
304        self.has_data = false;
305    }
306}
307
308/// The kind of error that occurred attempting to open an [`EventStream`].
309#[derive(thiserror::Error, Debug)]
310#[non_exhaustive]
311pub enum SseErrorKind {
312    /// An HTTP error attempting to make the request.
313    #[error(transparent)]
314    Http(#[from] trillium_http::Error),
315
316    /// The response status was not a success (2xx).
317    #[error("Unexpected response status {0} for SSE request")]
318    Status(Status),
319
320    /// The response content-type was not `text/event-stream`.
321    #[error("Unexpected content-type for SSE request: {0:?}")]
322    UnexpectedContentType(Option<String>),
323
324    /// [`Conn::into_sse`] was called on a [`Conn`] that had already been executed (its status is
325    /// already set). The request *is* the execution; build the conn and await `into_sse`
326    /// directly without awaiting first.
327    #[error(
328        "Conn::into_sse called after execution — build the conn and await into_sse instead of \
329         awaiting the conn separately"
330    )]
331    AlreadyExecuted,
332
333    /// The response had no body to read as an event stream.
334    #[error("SSE response had no body")]
335    NoBody,
336}
337
338/// An attempt to open an [`EventStream`] via [`Conn::into_sse`] failed.
339///
340/// This dereferences to the [`Conn`] and converts back into it with [`From`]/[`Into`], so the
341/// caller can inspect the response that caused the failure.
342#[derive(Debug)]
343pub struct SseError {
344    /// The kind of error that occurred.
345    pub kind: SseErrorKind,
346    conn: Box<Conn>,
347}
348
349impl SseError {
350    fn new(conn: Conn, kind: SseErrorKind) -> Self {
351        Self {
352            kind,
353            conn: Box::new(conn),
354        }
355    }
356}
357
358impl From<SseError> for Conn {
359    fn from(value: SseError) -> Self {
360        *value.conn
361    }
362}
363
364impl Deref for SseError {
365    type Target = Conn;
366
367    fn deref(&self) -> &Self::Target {
368        &self.conn
369    }
370}
371
372impl DerefMut for SseError {
373    fn deref_mut(&mut self) -> &mut Self::Target {
374        &mut self.conn
375    }
376}
377
378impl Error for SseError {}
379
380impl Display for SseError {
381    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
382        Display::fmt(&self.kind, f)
383    }
384}
385
386#[cfg(test)]
387mod tests {
388    use super::*;
389
390    /// Feed `input` to a fresh decoder in one chunk, then again one byte at a time, asserting
391    /// both produce the same events. Splitting per byte exercises the cross-chunk line-terminator
392    /// and field-accumulation state.
393    fn decode(input: &[u8]) -> Vec<Event> {
394        let mut whole = Decoder::default();
395        let mut whole_out = VecDeque::new();
396        whole.push(input, &mut whole_out);
397
398        let mut split = Decoder::default();
399        let mut split_out = VecDeque::new();
400        for byte in input {
401            split.push(&[*byte], &mut split_out);
402        }
403
404        assert_eq!(whole_out, split_out, "chunked decode diverged from whole");
405        whole_out.into()
406    }
407
408    #[test]
409    fn fields_comments_and_terminators() {
410        let events =
411            decode(b": this is a comment\nevent: greeting\ndata: hello\nid: 42\nretry: 3000\n\n");
412        assert_eq!(events.len(), 1);
413        let event = &events[0];
414        assert_eq!(event.data(), "hello");
415        assert_eq!(event.event_type(), Some("greeting"));
416        assert_eq!(event.id(), Some("42"));
417        assert_eq!(event.retry(), Some(Duration::from_millis(3000)));
418    }
419
420    #[test]
421    fn multiline_data_joins_with_newline() {
422        let events = decode(b"data: one\ndata: two\ndata:three\n\n");
423        // Only the single space after the first colon is stripped; "data:three" has none.
424        assert_eq!(events[0].data(), "one\ntwo\nthree");
425    }
426
427    #[test]
428    fn crlf_and_cr_terminators() {
429        let crlf = decode(b"data: a\r\n\r\n");
430        assert_eq!(crlf[0].data(), "a");
431        let cr = decode(b"data: b\r\r");
432        assert_eq!(cr[0].data(), "b");
433    }
434
435    #[test]
436    fn empty_data_line_dispatches_empty_event() {
437        // A bare `data` field (no value) still counts as data and dispatches.
438        let events = decode(b"data\n\n");
439        assert_eq!(events.len(), 1);
440        assert_eq!(events[0].data(), "");
441    }
442
443    #[test]
444    fn blank_lines_without_data_dispatch_nothing() {
445        assert!(decode(b"\n\n\n").is_empty());
446        assert!(decode(b": just a comment\n\n").is_empty());
447    }
448
449    #[test]
450    fn incomplete_trailing_event_is_discarded() {
451        // No terminating blank line: the event is never dispatched.
452        assert!(decode(b"data: pending\n").is_empty());
453    }
454
455    #[test]
456    fn id_persists_across_events_retry_does_not() {
457        let events = decode(b"id: 1\nretry: 500\ndata: a\n\ndata: b\n\n");
458        assert_eq!(events[0].id(), Some("1"));
459        assert_eq!(events[0].retry(), Some(Duration::from_millis(500)));
460        // `id` is the last-event-id and carries forward; `retry` is consumed by the first event.
461        assert_eq!(events[1].id(), Some("1"));
462        assert_eq!(events[1].retry(), None);
463    }
464
465    #[test]
466    fn invalid_retry_is_ignored() {
467        let events = decode(b"retry: not-a-number\ndata: a\n\n");
468        assert_eq!(events[0].retry(), None);
469    }
470}
trillium_client/sse.rs

trillium_client/
sse.rs
