Read SSE Events with Reqwest

use futures_util::StreamExt;
use reqwest::Client;
use std::time::Duration;

#[derive(Debug)]
struct SseEvent {
    event: String,
    data: String,
}

#[derive(Default)]
struct SseParser {
    event: String,
    data: Vec<String>,
}

impl SseParser {
    fn push_line(&mut self, line: &str) -> Option<SseEvent> {
        if line.is_empty() {
            if self.data.is_empty() {
                self.event.clear();
                return None;
            }
            let name = std::mem::take(&mut self.event);
            let data = std::mem::take(&mut self.data);
            return Some(SseEvent {
                event: if name.is_empty() {
                    "message".to_string()
                } else {
                    name
                },
                data: data.join("\n"),
            });
        }
        if line.starts_with(':') {
            return None;
        }
        let (field, value) = match line.split_once(':') {
            Some((field, value)) => (field, value.strip_prefix(' ').unwrap_or(value)),
            None => (line, ""),
        };
        match field {
            "event" => self.event = value.to_string(),
            "data" => self.data.push(value.to_string()),
            _ => {}
        }
        None
    }

    fn finish(&mut self) -> Option<SseEvent> {
        self.push_line("")
    }
}

async fn read_events(url: &str) -> Result<Vec<SseEvent>, reqwest::Error> {
    let client = Client::builder()
        .timeout(Duration::from_secs(30))
        .build()?;
    let response = client
        .get(url)
        .header("Accept", "text/event-stream")
        .send()
        .await?
        .error_for_status()?;

    let mut stream = response.bytes_stream();
    let mut buffer = String::new();
    let mut parser = SseParser::default();
    let mut events = Vec::new();

    while let Some(chunk) = stream.next().await {
        let chunk = chunk?;
        buffer.push_str(&String::from_utf8_lossy(&chunk));
        while let Some(index) = buffer.find('\n') {
            let line = buffer[..index].trim_end_matches('\r').to_string();
            buffer.drain(..=index);
            if let Some(event) = parser.push_line(&line) {
                events.push(event);
            }
        }
    }
    if !buffer.is_empty() {
        if let Some(event) = parser.push_line(&buffer) {
            events.push(event);
        }
    }
    if let Some(event) = parser.finish() {
        events.push(event);
    }
    Ok(events)
}

#[tokio::main]
async fn main() {
    match read_events("https://stream.example.com/events").await {
        Ok(events) => {
            for event in events {
                println!("{}: {}", event.event, event.data);
            }
        }
        Err(error) => eprintln!("stream failed: {error}"),
    }
}