ai-app/client-core/src/sse.rs

//! Server-sent-events framing, ported from `app/.../Sse.kt`: `data:` and
//! `event:` lines accumulate until a blank line ends the frame, comments
//! start with `:`, and a frame is either named with no payload or a payload
//! with no name.
//!
//! Pure and line-at-a-time, unlike the Kotlin original which also owned the
//! socket: `server/routes.rs`'s SSE bodies are one event per line, so a
//! caller here feeds lines from wherever they came from (a real connection,
//! a test fixture) and gets frames back with no I/O of its own -- which is
//! what lets this be tested with no server, per RUST.md's "pure logic
//! first" for this crate.

/// One SSE frame: its name (`None` for an ordinary data frame) and its payload.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Frame {
    pub name: Option<String>,
    pub data: String,
}

/// Accumulates lines into [`Frame`]s. One instance per connection --
/// `feed_line` is called for every line the transport reads (with line
/// endings already stripped), and answers a frame when a blank line closes
/// one.
#[derive(Debug, Default)]
pub struct SseReader {
    data: String,
    name: Option<String>,
}

impl SseReader {
    pub fn new() -> Self {
        Self::default()
    }

    /// Feeds one line (no trailing `\n`). Answers the frame this line
    /// completed, if any.
    pub fn feed_line(&mut self, line: &str) -> Option<Frame> {
        if line.is_empty() {
            if self.name.is_some() || !self.data.is_empty() {
                let frame = Frame {
                    name: self.name.take(),
                    data: std::mem::take(&mut self.data),
                };
                return Some(frame);
            }
            return None;
        }
        if let Some(rest) = line.strip_prefix("data:") {
            self.data.push_str(rest.trim());
        } else if let Some(rest) = line.strip_prefix("event:") {
            self.name = Some(rest.trim().to_string());
        }
        // `id:`, comments -- nothing to do.
        None
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn frames(lines: &[&str]) -> Vec<Frame> {
        let mut reader = SseReader::new();
        lines.iter().filter_map(|l| reader.feed_line(l)).collect()
    }

    #[test]
    fn a_data_only_frame_has_no_name() {
        assert_eq!(
            frames(&["data:hello", ""]),
            vec![Frame {
                name: None,
                data: "hello".to_string()
            }]
        );
    }

    #[test]
    fn a_named_frame_with_no_payload_still_completes() {
        assert_eq!(
            frames(&["event:reset", ""]),
            vec![Frame {
                name: Some("reset".to_string()),
                data: String::new()
            }]
        );
    }

    #[test]
    fn a_blank_line_with_nothing_pending_yields_no_frame() {
        assert_eq!(frames(&[""]), vec![]);
    }

    #[test]
    fn a_comment_and_an_id_line_are_ignored() {
        assert_eq!(
            frames(&[":keepalive", "id:5", "data:hi", ""]),
            vec![Frame {
                name: None,
                data: "hi".to_string()
            }]
        );
    }

    #[test]
    fn two_frames_in_a_row_are_both_reported() {
        assert_eq!(
            frames(&["data:one", "", "data:two", ""]),
            vec![
                Frame {
                    name: None,
                    data: "one".to_string()
                },
                Frame {
                    name: None,
                    data: "two".to_string()
                },
            ]
        );
    }
}