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

io_http/sse/
frame.rs
