用 Reqwest 读取 SSE 事件

功能说明

服务器发送事件以文本行传输，空行表示一帧结束。data 可以出现多行，最终用换行拼接；event 指定类型，缺省为 message；以冒号开头的是注释，常作为心跳。网络数据块与逻辑行没有对应关系，一帧可能被拆到多个块，一个块也可能包含多帧，因此必须在字节流上维护缓冲，只处理完整换行。

本例用 reqwest 发起请求并检查状态码，再通过 bytes_stream 增量读取。SseParser 保存当前事件的类型和数据行，每遇到空行就产出一帧。流结束时把剩余缓冲和未闭合的帧刷新出来。

关键代码

push_line 对空行做两件事：没有数据时清空事件名并返回空，防止上一个事件名泄漏到下一帧；有数据时取走事件名和数据，事件名为空则回退为 message。字段分割只做一次，冒号后的一个前导空格按规范去掉。read_events 把字节块以有损方式转成字符串，找到换行后截取整行并移除回车，再交给解析器。缓冲区使用 drain 删除已处理部分，避免无限增长。

使用注意

该函数在流结束时返回全部事件，真实 SSE 连接通常长期不结束，生产代码应改为边解析边回调或通过通道发送，并设置空闲超时与心跳检测。示例未发送 Last-Event-ID，断线重连可能重复事件，消费端需幂等。HTTP 错误通过 error_for_status 转为错误，认证令牌应从安全配置注入而不是硬编码。缓冲区没有大小上限，异常服务端发送超长行时可能耗尽内存。TLS 校验默认开启，不要为调试关闭。

验证方式

把 URL 换成返回事件流的本地服务，运行后应逐条打印事件名与数据。可在 data 中放多行内容，确认输出用换行拼接；插入以冒号开头的注释行，应被忽略。让服务端把一帧拆成多个数据块发送，解析结果仍应完整，验证跨块缓冲正确。

代码示例

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}"),
    }
}