io_http/sse/
frame.rs

1//! I/O-free coroutine decoding a W3C [Server-Sent Events] stream.
2//! Line-oriented and infallible: the parser never terminates and the
3//! outer driver stops when the body stream closes.
4//!
5//! [Server-Sent Events]: https://html.spec.whatwg.org/multipage/server-sent-events.html
6
7use core::{convert::Infallible, mem, str};
8
9use alloc::{
10    string::{String, ToString},
11    vec::Vec,
12};
13
14use log::trace;
15use memchr::memchr;
16
17use crate::coroutine::*;
18
19/// One dispatched Server-Sent Event. `event` is `None` when the
20/// `event:` field was absent; `data` has the trailing newline stripped.
21#[derive(Clone, Debug, Default, PartialEq, Eq)]
22pub struct SseFrame {
23    /// The event type, when the stream named one.
24    pub event: Option<String>,
25    /// The event payload, data lines joined by newlines.
26    pub data: String,
27    /// The last event id in effect when the event was dispatched.
28    pub id: Option<String>,
29    /// The reconnection delay in milliseconds, when the stream set one.
30    pub retry: Option<u64>,
31}
32
33/// Per-step yield emitted by [`SseFrameParser`].
34#[derive(Debug)]
35pub enum SseFrameParserYield {
36    /// One dispatched event.
37    Frame(SseFrame),
38    /// The parser wants more body bytes handed back on the next
39    /// resume.
40    WantsBytes,
41}
42
43/// I/O-free Server-Sent Events frame parser.
44#[derive(Debug, Default)]
45pub struct SseFrameParser {
46    buf: Vec<u8>,
47    bom_stripped: bool,
48    event: Option<String>,
49    data: String,
50    last_event_id: Option<String>,
51    retry: Option<u64>,
52}
53
54impl SseFrameParser {
55    /// Last-event-id seen so far; persists across dispatched frames so a
56    /// reconnecting caller can resume via the `Last-Event-ID` header.
57    pub fn last_event_id(&self) -> Option<&str> {
58        self.last_event_id.as_deref()
59    }
60}
61
62impl HttpCoroutine for SseFrameParser {
63    type Yield = SseFrameParserYield;
64    type Return = Infallible;
65
66    fn resume(&mut self, arg: Option<&[u8]>) -> HttpCoroutineState<Self::Yield, Self::Return> {
67        if let Some(data) = arg {
68            self.buf.extend_from_slice(data);
69        }
70
71        if !self.bom_stripped && self.buf.len() >= 3 {
72            if self.buf.starts_with(&[0xEF, 0xBB, 0xBF]) {
73                self.buf.drain(..3);
74            }
75            self.bom_stripped = true;
76        }
77
78        loop {
79            let Some((line, consumed)) = next_line(&self.buf) else {
80                return HttpCoroutineState::Yielded(SseFrameParserYield::WantsBytes);
81            };
82
83            let line_bytes = self.buf[..line].to_vec();
84            self.buf.drain(..consumed);
85
86            if line_bytes.is_empty() {
87                if self.data.is_empty() && self.event.is_none() {
88                    continue;
89                }
90
91                if self.data.ends_with('\n') {
92                    self.data.pop();
93                }
94
95                let frame = SseFrame {
96                    event: self.event.take(),
97                    data: mem::take(&mut self.data),
98                    id: self.last_event_id.clone(),
99                    retry: self.retry.take(),
100                };
101                return HttpCoroutineState::Yielded(SseFrameParserYield::Frame(frame));
102            }
103
104            if line_bytes.first() == Some(&b':') {
105                continue;
106            }
107
108            let (name, value) = split_field(&line_bytes);
109            let Ok(name) = str::from_utf8(name) else {
110                trace!("ignore field with non-utf8 name");
111                continue;
112            };
113            let Ok(value) = str::from_utf8(value) else {
114                trace!("ignore field with non-utf8 value");
115                continue;
116            };
117
118            match name {
119                "event" => self.event = Some(value.to_string()),
120                "data" => {
121                    self.data.push_str(value);
122                    self.data.push('\n');
123                }
124                "id" => {
125                    if !value.contains('\0') {
126                        self.last_event_id = Some(value.to_string());
127                    }
128                }
129                "retry" => {
130                    if let Ok(n) = value.parse::<u64>() {
131                        self.retry = Some(n);
132                    }
133                }
134                _ => trace!("ignore unknown field `{name}`"),
135            }
136        }
137    }
138}
139
140/// Returns (line_end_excl_terminator, total_consumed) or None when the
141/// buffer doesn't yet contain a complete line. Terminator may be \r\n,
142/// \n, or bare \r; a trailing \r is treated as incomplete pending \n.
143fn next_line(buf: &[u8]) -> Option<(usize, usize)> {
144    let cr = memchr(b'\r', buf);
145    let lf = memchr(b'\n', buf);
146
147    match (cr, lf) {
148        (Some(cr), Some(lf)) if cr + 1 == lf => Some((cr, lf + 1)),
149        (Some(cr), Some(lf)) if cr < lf => {
150            if cr + 1 == buf.len() {
151                None
152            } else {
153                Some((cr, cr + 1))
154            }
155        }
156        (Some(cr), None) => {
157            if cr + 1 == buf.len() {
158                None
159            } else {
160                Some((cr, cr + 1))
161            }
162        }
163        (_, Some(lf)) => Some((lf, lf + 1)),
164        (None, None) => None,
165    }
166}
167
168/// Splits a non-empty SSE line on the first `:`; a single leading SP
169/// in the value is stripped per spec.
170fn split_field(line: &[u8]) -> (&[u8], &[u8]) {
171    match memchr(b':', line) {
172        None => (line, &[]),
173        Some(colon) => {
174            let name = &line[..colon];
175            let mut value = &line[colon + 1..];
176            if value.first() == Some(&b' ') {
177                value = &value[1..];
178            }
179            (name, value)
180        }
181    }
182}
183
184#[cfg(test)]
185mod tests {
186    use alloc::vec;
187
188    use crate::sse::frame::*;
189
190    fn collect(stream: &[u8]) -> Vec<SseFrame> {
191        let mut parser = SseFrameParser::default();
192        let mut arg: Option<&[u8]> = Some(stream);
193        let mut frames = Vec::new();
194
195        loop {
196            match parser.resume(arg.take()) {
197                HttpCoroutineState::Yielded(SseFrameParserYield::Frame(frame)) => {
198                    frames.push(frame)
199                }
200                HttpCoroutineState::Yielded(SseFrameParserYield::WantsBytes) => break,
201                HttpCoroutineState::Complete(never) => match never {},
202            }
203        }
204
205        frames
206    }
207
208    #[test]
209    fn single_data_event() {
210        let frames = collect(b"data: hello\n\n");
211        assert_eq!(
212            frames,
213            vec![SseFrame {
214                event: None,
215                data: "hello".into(),
216                id: None,
217                retry: None,
218            }]
219        );
220    }
221
222    #[test]
223    fn multi_line_data_joined_by_newline() {
224        let frames = collect(b"data: hello\ndata: world\n\n");
225        assert_eq!(frames[0].data, "hello\nworld");
226    }
227
228    #[test]
229    fn event_and_id_fields() {
230        let frames = collect(b"event: state\ndata: x\nid: 42\n\n");
231        assert_eq!(frames[0].event.as_deref(), Some("state"));
232        assert_eq!(frames[0].data, "x");
233        assert_eq!(frames[0].id.as_deref(), Some("42"));
234    }
235
236    #[test]
237    fn retry_parsed_when_integer() {
238        let frames = collect(b"retry: 5000\ndata: x\n\n");
239        assert_eq!(frames[0].retry, Some(5000));
240    }
241
242    #[test]
243    fn retry_ignored_when_non_integer() {
244        let frames = collect(b"retry: hello\ndata: x\n\n");
245        assert_eq!(frames[0].retry, None);
246    }
247
248    #[test]
249    fn comment_lines_ignored() {
250        let frames = collect(b": keep-alive\ndata: x\n\n");
251        assert_eq!(frames[0].data, "x");
252    }
253
254    #[test]
255    fn empty_event_no_dispatch() {
256        let frames = collect(b"\n\n\n");
257        assert!(frames.is_empty());
258    }
259
260    #[test]
261    fn id_persists_across_events() {
262        let mut parser = SseFrameParser::default();
263        let mut arg: Option<&[u8]> = Some(b"id: 1\ndata: a\n\ndata: b\n\n");
264        let mut frames = Vec::new();
265
266        loop {
267            match parser.resume(arg.take()) {
268                HttpCoroutineState::Yielded(SseFrameParserYield::Frame(frame)) => {
269                    frames.push(frame)
270                }
271                HttpCoroutineState::Yielded(SseFrameParserYield::WantsBytes) => break,
272                HttpCoroutineState::Complete(never) => match never {},
273            }
274        }
275
276        assert_eq!(frames[0].id.as_deref(), Some("1"));
277        assert_eq!(frames[1].id.as_deref(), Some("1"));
278        assert_eq!(parser.last_event_id(), Some("1"));
279    }
280
281    #[test]
282    fn id_with_null_is_ignored() {
283        let mut parser = SseFrameParser::default();
284        let stream = b"id: bad\0\ndata: x\n\n";
285        let arg: Option<&[u8]> = Some(stream);
286
287        match parser.resume(arg) {
288            HttpCoroutineState::Yielded(SseFrameParserYield::Frame(_)) => {}
289            HttpCoroutineState::Yielded(SseFrameParserYield::WantsBytes) => {
290                unreachable!("wants bytes");
291            }
292            HttpCoroutineState::Complete(never) => match never {},
293        }
294
295        assert_eq!(parser.last_event_id(), None);
296    }
297
298    #[test]
299    fn crlf_line_separator() {
300        let frames = collect(b"data: hello\r\n\r\n");
301        assert_eq!(frames[0].data, "hello");
302    }
303
304    #[test]
305    fn bare_cr_line_separator() {
306        let frames = collect(b"data: hello\r\rTAIL");
307        assert_eq!(frames[0].data, "hello");
308    }
309
310    #[test]
311    fn bom_stripped_at_stream_start() {
312        let frames = collect(b"\xEF\xBB\xBFdata: hello\n\n");
313        assert_eq!(frames[0].data, "hello");
314    }
315
316    #[test]
317    fn field_value_leading_space_stripped() {
318        let frames = collect(b"data:  hello\n\n");
319        assert_eq!(frames[0].data, " hello");
320    }
321
322    #[test]
323    fn field_no_value() {
324        let frames = collect(b"data\n\n");
325        assert_eq!(frames[0].data, "");
326    }
327
328    #[test]
329    fn incomplete_then_resumed() {
330        let mut parser = SseFrameParser::default();
331        let mut arg: Option<&[u8]> = Some(b"data: hel");
332        let mut frames = Vec::new();
333
334        loop {
335            match parser.resume(arg.take()) {
336                HttpCoroutineState::Yielded(SseFrameParserYield::Frame(frame)) => {
337                    frames.push(frame);
338                    break;
339                }
340                HttpCoroutineState::Yielded(SseFrameParserYield::WantsBytes) => {
341                    if arg.is_none() {
342                        arg = Some(b"lo\n\n");
343                    } else {
344                        break;
345                    }
346                }
347                HttpCoroutineState::Complete(never) => match never {},
348            }
349        }
350
351        assert_eq!(frames[0].data, "hello");
352    }
353
354    #[test]
355    fn unknown_field_ignored() {
356        let frames = collect(b"foobar: x\ndata: y\n\n");
357        assert_eq!(frames[0].data, "y");
358    }
359
360    #[test]
361    fn event_resets_after_dispatch() {
362        let frames = collect(b"event: a\ndata: x\n\ndata: y\n\n");
363        assert_eq!(frames[0].event.as_deref(), Some("a"));
364        assert_eq!(frames[1].event, None);
365    }
366}
io_http/sse/frame.rs

io_http/sse/
frame.rs
