ferra_agent/
sse.rs

1//! Minimal SSE wire-format parser.
2//!
3//! Only handles the fields Ferra emits: `event:`, `id:`, `data:`. Comments
4//! (`:` lines) and unknown fields are ignored. `data:` accumulates across
5//! multiple lines (joined with `\n`). Events terminate on a blank line.
6
7use bytes::Bytes;
8
9#[derive(Debug, Default, Clone)]
10pub struct SseEvent {
11    pub event: Option<String>,
12    pub id: Option<String>,
13    pub data: String,
14}
15
16pub struct SseParser {
17    buf: Vec<u8>,
18    cur: SseEvent,
19}
20
21impl Default for SseParser {
22    fn default() -> Self {
23        Self::new()
24    }
25}
26
27impl SseParser {
28    pub fn new() -> Self {
29        Self {
30            buf: Vec::new(),
31            cur: SseEvent::default(),
32        }
33    }
34
35    /// Feed bytes from the network and emit any complete events that
36    /// resulted. The returned events are owned and the parser retains its
37    /// internal partial-line state.
38    pub fn feed(&mut self, chunk: &Bytes) -> Vec<SseEvent> {
39        self.buf.extend_from_slice(chunk);
40        let mut out = Vec::new();
41        loop {
42            // Find end of next line: \n (handles CRLF too via \r trim).
43            let Some(pos) = self.buf.iter().position(|b| *b == b'\n') else {
44                break;
45            };
46            let raw_line: Vec<u8> = self.buf.drain(..=pos).collect();
47            let line_with_lf = String::from_utf8_lossy(&raw_line);
48            let line = line_with_lf.trim_end_matches(['\n', '\r']);
49
50            if line.is_empty() {
51                // Event terminator.
52                if let Some(ev) = self.flush() {
53                    out.push(ev);
54                }
55                continue;
56            }
57            if line.starts_with(':') {
58                // Comment.
59                continue;
60            }
61            let (field, value) = match line.find(':') {
62                Some(i) => (&line[..i], line[i + 1..].strip_prefix(' ').unwrap_or(&line[i + 1..])),
63                None => (line, ""),
64            };
65            match field {
66                "event" => self.cur.event = Some(value.to_string()),
67                "id" => self.cur.id = Some(value.to_string()),
68                "data" => {
69                    if !self.cur.data.is_empty() {
70                        self.cur.data.push('\n');
71                    }
72                    self.cur.data.push_str(value);
73                }
74                _ => {} // ignore unknown fields (incl. retry:)
75            }
76        }
77        out
78    }
79
80    fn flush(&mut self) -> Option<SseEvent> {
81        if self.cur.event.is_none() && self.cur.id.is_none() && self.cur.data.is_empty() {
82            return None;
83        }
84        Some(std::mem::take(&mut self.cur))
85    }
86}
87
88#[cfg(test)]
89mod tests {
90    use super::*;
91    use bytes::Bytes;
92
93    fn feed_str(p: &mut SseParser, s: &str) -> Vec<SseEvent> {
94        p.feed(&Bytes::copy_from_slice(s.as_bytes()))
95    }
96
97    #[test]
98    fn parses_simple_event() {
99        let mut p = SseParser::new();
100        let evs = feed_str(&mut p, "event: kv_changed\nid: 7\ndata: hello\n\n");
101        assert_eq!(evs.len(), 1);
102        assert_eq!(evs[0].event.as_deref(), Some("kv_changed"));
103        assert_eq!(evs[0].id.as_deref(), Some("7"));
104        assert_eq!(evs[0].data, "hello");
105    }
106
107    #[test]
108    fn handles_split_chunks() {
109        let mut p = SseParser::new();
110        let mut evs = feed_str(&mut p, "event: kv_chan");
111        assert!(evs.is_empty());
112        evs = feed_str(&mut p, "ged\nid: 1\ndata: ");
113        assert!(evs.is_empty());
114        evs = feed_str(&mut p, "x\n\n");
115        assert_eq!(evs.len(), 1);
116        assert_eq!(evs[0].event.as_deref(), Some("kv_changed"));
117        assert_eq!(evs[0].data, "x");
118    }
119
120    #[test]
121    fn parses_two_events_in_one_chunk() {
122        let mut p = SseParser::new();
123        let evs = feed_str(
124            &mut p,
125            "event: heartbeat\ndata: {}\n\nevent: kv_changed\nid: 2\ndata: {\"k\":1}\n\n",
126        );
127        assert_eq!(evs.len(), 2);
128        assert_eq!(evs[0].event.as_deref(), Some("heartbeat"));
129        assert_eq!(evs[1].event.as_deref(), Some("kv_changed"));
130        assert_eq!(evs[1].data, "{\"k\":1}");
131    }
132
133    #[test]
134    fn ignores_comments_and_crlf() {
135        let mut p = SseParser::new();
136        let evs = feed_str(
137            &mut p,
138            ": this is a comment\r\nevent: heartbeat\r\ndata: {}\r\n\r\n",
139        );
140        assert_eq!(evs.len(), 1);
141        assert_eq!(evs[0].event.as_deref(), Some("heartbeat"));
142    }
143
144    #[test]
145    fn multiline_data() {
146        let mut p = SseParser::new();
147        let evs = feed_str(&mut p, "event: x\ndata: line1\ndata: line2\n\n");
148        assert_eq!(evs.len(), 1);
149        assert_eq!(evs[0].data, "line1\nline2");
150    }
151}
ferra_agent/sse.rs

ferra_agent/
sse.rs
