sim_lib_net_core/
sse.rs

1//! Server-Sent Events record decoding.
2
3use crate::line::LineDecoder;
4
5/// A decoded Server-Sent Events record.
6#[derive(Clone, Debug, PartialEq, Eq)]
7pub struct SseEvent {
8    /// The `event:` field value, if any was present in the record.
9    pub event: Option<String>,
10    /// The `data:` payload. Multiple `data:` lines are joined with `\n`.
11    pub data: String,
12}
13
14/// Incremental Server-Sent Events decoder.
15///
16/// Accumulates `event:` and `data:` field lines and emits an [`SseEvent`] when a
17/// blank line terminates the record (standard SSE dispatch). Multiple `data:`
18/// lines within one record are joined with `\n`. Comment lines (those starting
19/// with `:`) and unknown fields are ignored. Field values have a single leading
20/// space after the colon stripped, per the SSE specification.
21///
22/// `push` does the newline framing internally via [`LineDecoder`]; callers that
23/// already have whole lines can use [`push_line`](SseDecoder::push_line).
24#[derive(Debug, Default)]
25pub struct SseDecoder {
26    lines: LineDecoder,
27    event: Option<String>,
28    data: Vec<String>,
29    have_record: bool,
30}
31
32impl SseDecoder {
33    /// Create an empty decoder.
34    pub fn new() -> Self {
35        Self::default()
36    }
37
38    /// Feed raw bytes; return any records completed by blank lines.
39    pub fn push(&mut self, bytes: &[u8]) -> Vec<SseEvent> {
40        let mut events = Vec::new();
41        for line in self.lines.push(bytes) {
42            // SSE payloads are UTF-8; lossy decoding keeps the framer robust and
43            // pushes any invalid-byte handling onto the caller's data parse.
44            let text = String::from_utf8_lossy(&line);
45            if let Some(event) = self.push_line(&text) {
46                events.push(event);
47            }
48        }
49        events
50    }
51
52    /// Feed a single already-framed line (without its terminating newline).
53    /// Returns an [`SseEvent`] when this line is the blank line that dispatches
54    /// the accumulated record.
55    pub fn push_line(&mut self, line: &str) -> Option<SseEvent> {
56        if line.is_empty() {
57            return self.dispatch();
58        }
59        if line.starts_with(':') {
60            // Comment line: ignored, but it still marks an active record so a
61            // following blank line dispatches an empty event only if fields set.
62            return None;
63        }
64        let (field, value) = match line.split_once(':') {
65            Some((field, rest)) => (field, rest.strip_prefix(' ').unwrap_or(rest)),
66            None => (line, ""),
67        };
68        match field {
69            "event" => {
70                self.event = Some(value.to_owned());
71                self.have_record = true;
72            }
73            "data" => {
74                self.data.push(value.to_owned());
75                self.have_record = true;
76            }
77            _ => {}
78        }
79        None
80    }
81
82    fn dispatch(&mut self) -> Option<SseEvent> {
83        if !self.have_record {
84            return None;
85        }
86        let event = self.event.take();
87        let data = std::mem::take(&mut self.data).join("\n");
88        self.have_record = false;
89        Some(SseEvent { event, data })
90    }
91
92    /// Flush any buffered partial line and any accumulated-but-undispatched
93    /// record (a stream that ended without a final blank line).
94    pub fn flush(&mut self) -> Option<SseEvent> {
95        if let Some(line) = self.lines.flush() {
96            let text = String::from_utf8_lossy(&line);
97            if let Some(event) = self.push_line(&text) {
98                return Some(event);
99            }
100        }
101        self.dispatch()
102    }
103}
sim_lib_net_core/sse.rs

sim_lib_net_core/
sse.rs
