feltdb_client/
events.rs

1use serde::de::DeserializeOwned;
2
3use crate::error::{Error, Result};
4
5/// One server-sent event.
6#[derive(Debug, Clone, PartialEq, Eq)]
7pub struct ServerEvent {
8    /// The opaque cursor after this event. Pass it back to resume from here.
9    pub id: Option<String>,
10    /// The event name; `state.committed` for a committed change.
11    pub event: String,
12    /// The event payload, a JSON document for `state.committed`.
13    pub data: String,
14}
15
16impl ServerEvent {
17    /// Decode the payload.
18    pub fn json<T: DeserializeOwned>(&self) -> Result<T> {
19        serde_json::from_str(&self.data).map_err(|source| Error::Decode {
20            status: 200,
21            source,
22        })
23    }
24}
25
26/// A live event stream: a delivery adapter over committed history, not the history itself.
27///
28/// It does not reconnect on its own. When it ends or fails, open a new one from
29/// [`EventStream::last_event_id`] and nothing committed in between is skipped. For a complete,
30/// bounded read of history use the application's event replay instead.
31pub struct EventStream {
32    response: reqwest::Response,
33    buffer: Vec<u8>,
34    last_event_id: Option<String>,
35    finished: bool,
36}
37
38impl EventStream {
39    pub(crate) fn new(response: reqwest::Response, resumed_from: Option<String>) -> Self {
40        Self {
41            response,
42            buffer: Vec::new(),
43            last_event_id: resumed_from,
44            finished: false,
45        }
46    }
47
48    /// The cursor of the last event delivered (or the one this stream resumed from).
49    pub fn last_event_id(&self) -> Option<&str> {
50        self.last_event_id.as_deref()
51    }
52
53    /// The next event; `None` when the server closed the stream.
54    ///
55    /// An `authority.error` event from the server ends the stream with [`Error::Stream`].
56    pub async fn next(&mut self) -> Option<Result<ServerEvent>> {
57        loop {
58            if self.finished {
59                return None;
60            }
61            while let Some(frame) = take_frame(&mut self.buffer) {
62                let Some(event) = parse_frame(&frame) else {
63                    continue;
64                };
65                if event.event == "authority.error" {
66                    self.finished = true;
67                    return Some(Err(Error::Stream(event.data)));
68                }
69                if event.id.is_some() {
70                    self.last_event_id.clone_from(&event.id);
71                }
72                return Some(Ok(event));
73            }
74            match self.response.chunk().await {
75                // Bytes, not text: a chunk can end inside a multi-byte character, and only a
76                // complete frame is decoded. Carriage returns are dropped so CRLF framing parses
77                // as LF; JSON payloads escape theirs.
78                Ok(Some(bytes)) => self
79                    .buffer
80                    .extend(bytes.iter().copied().filter(|byte| *byte != b'\r')),
81                Ok(None) => {
82                    self.finished = true;
83                    return None;
84                }
85                Err(source) => {
86                    self.finished = true;
87                    return Some(Err(Error::transport(source, false)));
88                }
89            }
90        }
91    }
92}
93
94fn take_frame(buffer: &mut Vec<u8>) -> Option<String> {
95    let end = buffer.windows(2).position(|pair| pair == b"\n\n")?;
96    let frame = String::from_utf8_lossy(&buffer[..end]).into_owned();
97    buffer.drain(..end + 2);
98    Some(frame)
99}
100
101/// Parse one SSE frame. Comments (keep-alives) and frames without data are skipped.
102fn parse_frame(frame: &str) -> Option<ServerEvent> {
103    let mut id = None;
104    let mut event = None;
105    let mut data: Vec<&str> = Vec::new();
106    for line in frame.split('\n') {
107        if line.starts_with(':') {
108            continue;
109        }
110        let (field, value) = line.split_once(':').unwrap_or((line, ""));
111        let value = value.strip_prefix(' ').unwrap_or(value);
112        match field {
113            "id" => id = Some(value.to_owned()),
114            "event" => event = Some(value.to_owned()),
115            "data" => data.push(value),
116            _ => {}
117        }
118    }
119    if data.is_empty() {
120        return None;
121    }
122    Some(ServerEvent {
123        id,
124        event: event.unwrap_or_else(|| "message".to_owned()),
125        data: data.join("\n"),
126    })
127}
128
129#[cfg(test)]
130mod tests {
131    use super::*;
132
133    #[test]
134    fn frames_are_split_and_parsed() {
135        let mut buffer =
136            b": keep-alive\n\nid: c-2\nevent: state.committed\ndata: {\"a\":\ndata: 1}\n\nid: partial"
137                .to_vec();
138        assert_eq!(parse_frame(&take_frame(&mut buffer).unwrap()), None);
139        let event = parse_frame(&take_frame(&mut buffer).unwrap()).unwrap();
140        assert_eq!(event.id.as_deref(), Some("c-2"));
141        assert_eq!(event.event, "state.committed");
142        assert_eq!(event.data, "{\"a\":\n1}");
143        assert_eq!(take_frame(&mut buffer), None);
144        assert_eq!(buffer, b"id: partial");
145    }
146}
feltdb_client/events.rs

feltdb_client/
events.rs
