a3s_boot/http/
sse.rs

1use crate::{BootError, Result};
2use futures_core::Stream;
3use serde::Serialize;
4use std::pin::Pin;
5
6pub type SseStream = Pin<Box<dyn Stream<Item = Result<SseEvent>> + Send + 'static>>;
7
8#[derive(Debug, Clone, Default, PartialEq, Eq)]
9pub struct SseEvent {
10    id: Option<String>,
11    event: Option<String>,
12    retry: Option<u64>,
13    comment: Option<String>,
14    data: Option<String>,
15}
16
17impl SseEvent {
18    pub fn new(data: impl Into<String>) -> Self {
19        Self::default().with_data(data)
20    }
21
22    pub fn json<T>(data: &T) -> Result<Self>
23    where
24        T: Serialize,
25    {
26        let data =
27            serde_json::to_string(data).map_err(|err| BootError::Internal(err.to_string()))?;
28        Ok(Self::new(data))
29    }
30
31    pub fn comment(comment: impl Into<String>) -> Self {
32        Self::default().with_comment(comment)
33    }
34
35    pub fn stream<I>(events: I) -> SseStream
36    where
37        I: IntoIterator<Item = SseEvent>,
38        I::IntoIter: Send + 'static,
39    {
40        Box::pin(futures_util::stream::iter(events.into_iter().map(Ok)))
41    }
42
43    pub fn with_id(mut self, id: impl Into<String>) -> Self {
44        self.id = Some(id.into());
45        self
46    }
47
48    pub fn with_event(mut self, event: impl Into<String>) -> Self {
49        self.event = Some(event.into());
50        self
51    }
52
53    pub fn with_retry(mut self, retry: u64) -> Self {
54        self.retry = Some(retry);
55        self
56    }
57
58    pub fn with_comment(mut self, comment: impl Into<String>) -> Self {
59        self.comment = Some(comment.into());
60        self
61    }
62
63    pub fn with_data(mut self, data: impl Into<String>) -> Self {
64        self.data = Some(data.into());
65        self
66    }
67
68    pub fn encode(&self) -> Vec<u8> {
69        let mut encoded = String::new();
70
71        if let Some(comment) = &self.comment {
72            push_comment(&mut encoded, comment);
73        }
74        if let Some(id) = &self.id {
75            push_field(&mut encoded, "id", id);
76        }
77        if let Some(event) = &self.event {
78            push_field(&mut encoded, "event", event);
79        }
80        if let Some(retry) = self.retry {
81            encoded.push_str("retry: ");
82            encoded.push_str(&retry.to_string());
83            encoded.push('\n');
84        }
85        if let Some(data) = &self.data {
86            push_field(&mut encoded, "data", data);
87        }
88
89        encoded.push('\n');
90        encoded.into_bytes()
91    }
92}
93
94fn push_comment(encoded: &mut String, value: &str) {
95    for line in sse_lines(value) {
96        encoded.push_str(": ");
97        encoded.push_str(&line);
98        encoded.push('\n');
99    }
100}
101
102fn push_field(encoded: &mut String, name: &str, value: &str) {
103    for line in sse_lines(value) {
104        encoded.push_str(name);
105        encoded.push_str(": ");
106        encoded.push_str(&line);
107        encoded.push('\n');
108    }
109}
110
111fn sse_lines(value: &str) -> Vec<String> {
112    let normalized = value.replace("\r\n", "\n").replace('\r', "\n");
113    if normalized.is_empty() {
114        return vec![String::new()];
115    }
116
117    normalized.split('\n').map(str::to_string).collect()
118}
a3s_boot/http/sse.rs

a3s_boot/http/
sse.rs
