use core::fmt::Write as _;
use core::ops::ControlFlow;
use core::str::FromStr;
use serde_json::Value;
use thiserror::Error;
use crate::domain::{EventId, EventName, EventNameError};
const ID_FIELD: &str = "id";
const DATA_FIELD: &str = "data";
const EVENT_FIELD: &str = "event";
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SseEvent {
name: Option<EventName>,
id: Option<EventId>,
data: String,
}
impl SseEvent {
#[must_use]
pub const fn new(id: EventId, data: String) -> Self {
Self {
name: None,
id: Some(id),
data,
}
}
#[must_use]
pub const fn without_id(data: String) -> Self {
Self {
name: None,
id: None,
data,
}
}
#[must_use]
pub const fn named(name: EventName, data: String) -> Self {
Self {
name: Some(name),
id: None,
data,
}
}
#[must_use]
pub const fn named_with_id(name: EventName, id: EventId, data: String) -> Self {
Self {
name: Some(name),
id: Some(id),
data,
}
}
#[must_use]
pub const fn name(&self) -> Option<&EventName> {
self.name.as_ref()
}
#[must_use]
pub fn is_named(&self, event: &EventName) -> bool {
self.name.as_ref().is_some_and(|name| name.is(event))
}
pub fn json(id: EventId, value: &Value) -> Result<Self, SseError> {
let data = serde_json::to_string(value).map_err(|_| SseError::Unserialisable)?;
Ok(Self::new(id, data))
}
#[must_use]
pub const fn id(&self) -> Option<EventId> {
self.id
}
#[must_use]
pub fn data(&self) -> &str {
&self.data
}
#[must_use]
pub fn into_data(self) -> String {
self.data
}
#[must_use]
pub fn encode(&self) -> String {
let mut out = String::with_capacity(self.data.len() + 32);
if let Some(name) = &self.name {
let _ = writeln!(out, "{EVENT_FIELD}: {name}");
}
if let Some(id) = self.id {
let _ = writeln!(out, "{ID_FIELD}: {id}");
}
for_each_line(&self.data, |line| {
out.push_str(DATA_FIELD);
out.push_str(": ");
out.push_str(line);
out.push('\n');
ControlFlow::Continue(())
})
.ok();
out.push('\n');
out
}
pub fn decode(block: &str) -> Result<Self, SseError> {
Self::dispatch(block)?.ok_or(SseError::NoData)
}
pub fn dispatch(block: &str) -> Result<Option<Self>, SseError> {
let mut name: Option<EventName> = None;
let mut id: Option<EventId> = None;
let mut data: Vec<&str> = Vec::new();
for_each_line(block, |line| {
if line.starts_with(':') {
return ControlFlow::Continue(());
}
let (field, value) = match line.split_once(':') {
Some((field, value)) => (
field,
value.strip_prefix(' ').unwrap_or(value),
),
None => (line, ""),
};
match field {
ID_FIELD => match parse_event_id(value) {
Ok(parsed) => id = Some(parsed),
Err(error) => return ControlFlow::Break(error),
},
EVENT_FIELD => match EventName::from_str(value) {
Ok(parsed) => name = Some(parsed),
Err(error) => {
return ControlFlow::Break(SseError::MalformedName(error));
}
},
DATA_FIELD => data.push(value),
_ => {}
}
ControlFlow::Continue(())
})?;
if data.is_empty() {
return Ok(None);
}
Ok(Some(Self {
name,
id,
data: data.join("\n"),
}))
}
}
fn parse_event_id(value: &str) -> Result<EventId, SseError> {
u64::from_str(value).map_or(Err(SseError::MalformedId), |id| Ok(EventId::new(id)))
}
fn for_each_line<'i>(
input: &'i str,
mut visit: impl FnMut(&'i str) -> ControlFlow<SseError>,
) -> Result<(), SseError> {
let mut start = 0;
let mut index = 0;
while let Some(byte) = input.as_bytes().get(index) {
match *byte {
b'\n' => {
if let Some(line) = input.get(start..index)
&& let ControlFlow::Break(error) = visit(line)
{
return Err(error);
}
index += 1;
start = index;
}
b'\r' => {
if let Some(line) = input.get(start..index)
&& let ControlFlow::Break(error) = visit(line)
{
return Err(error);
}
index += 1;
if input.as_bytes().get(index) == Some(&b'\n') {
index += 1;
}
start = index;
}
_ => index += 1,
}
}
if let Some(last) = input.get(start..)
&& let ControlFlow::Break(error) = visit(last)
{
return Err(error);
}
Ok(())
}
#[derive(Debug, Clone, PartialEq, Eq, Error)]
pub enum SseError {
#[error("this block carries no data field, so it dispatches no event")]
NoData,
#[error("the id field is not a decimal event id")]
MalformedId,
#[error("the event field is not a usable event name: {0}")]
MalformedName(EventNameError),
#[error("the payload cannot be serialised as JSON")]
Unserialisable,
}
#[cfg(test)]
mod tests {
use serde_json::{Value, json};
use super::{SseError, SseEvent, for_each_line};
use crate::domain::{EventId, EventName, EventNameError, names};
use core::str::FromStr;
type TestError = Box<dyn std::error::Error>;
fn split(input: &str) -> Vec<&str> {
let mut lines = Vec::new();
let _ = for_each_line(input, |line| {
lines.push(line);
core::ops::ControlFlow::Continue(())
});
lines
}
#[test]
fn a_single_line_payload_is_one_data_line() {
let frame = SseEvent::new(EventId::new(7), r#"{"jsonrpc":"2.0"}"#.to_owned());
assert_eq!(frame.encode(), "id: 7\ndata: {\"jsonrpc\":\"2.0\"}\n\n");
}
#[test]
fn a_newline_in_the_payload_becomes_another_data_line() {
let frame = SseEvent::new(EventId::new(1), "one\ntwo\nthree".to_owned());
let encoded = frame.encode();
assert_eq!(
encoded, "id: 1\ndata: one\ndata: two\ndata: three\n\n",
"a payload newline must be carried by repeating the field name"
);
assert_eq!(encoded.matches("data: ").count(), 3);
assert!(
!encoded.contains("one\ntwo"),
"a raw newline reached the wire inside a data line"
);
}
#[test]
fn encode_then_decode_round_trips() -> Result<(), TestError> {
let payloads = [
String::new(),
"a".to_owned(),
"\n".to_owned(),
"\n\n".to_owned(),
"line one\nline two".to_owned(),
r#"{"a":"b\"c\\d"}"#.to_owned(),
"café — ünïcode".to_owned(),
"\u{7f}".to_owned(),
"trailing\n".to_owned(),
];
for payload in payloads {
let frame = SseEvent::new(EventId::new(3), payload.clone());
let decoded = SseEvent::decode(&frame.encode())?;
assert_eq!(decoded.data(), payload, "round trip changed {payload:?}");
assert_eq!(decoded.id(), Some(EventId::new(3)));
}
Ok(())
}
#[test]
fn the_data_line_count_is_the_newline_count_plus_one() -> Result<(), TestError> {
for newlines in 0..5_usize {
let payload = "x\n".repeat(newlines);
let encoded = SseEvent::without_id(payload.clone()).encode();
assert_eq!(
encoded.matches("data: ").count(),
newlines + 1,
"a payload of {newlines} newlines was mis-framed: {encoded:?}"
);
assert_eq!(SseEvent::decode(&encoded)?.data(), payload);
}
Ok(())
}
#[test]
fn a_carriage_return_is_a_line_terminator() -> Result<(), TestError> {
assert_eq!(split("a\rb"), vec!["a", "b"]);
assert_eq!(split("a\r\nb"), vec!["a", "b"]);
assert_eq!(split("a\nb"), vec!["a", "b"]);
assert_eq!(split("a\n"), vec!["a", ""]);
assert_eq!(split("a\r\n"), vec!["a", ""]);
assert_eq!(split("a\r"), vec!["a", ""]);
assert_eq!(split("a\n\n"), vec!["a", "", ""]);
assert_eq!(split("a"), vec!["a"]);
assert_eq!(split(""), vec![""]);
let frame = SseEvent::without_id("one\rtwo".to_owned());
assert_eq!(frame.encode(), "data: one\ndata: two\n\n");
assert_eq!(SseEvent::decode(&frame.encode())?.data(), "one\ntwo");
Ok(())
}
#[test]
fn crlf_encoded_frames_dispatch_exactly_one_event() -> Result<(), TestError> {
let frame = SseEvent::new(EventId::new(4), "one\ntwo".to_owned());
let crlf = frame.encode().replace('\n', "\r\n");
let decoded = SseEvent::decode(&crlf)?;
assert_eq!(decoded.data(), "one\ntwo");
assert_eq!(decoded.id(), Some(EventId::new(4)));
Ok(())
}
#[test]
fn a_frame_with_no_id_decodes_with_no_id() -> Result<(), TestError> {
let frame = SseEvent::without_id("{\"a\":1}".to_owned());
let encoded = frame.encode();
assert_eq!(encoded, "data: {\"a\":1}\n\n");
let decoded = SseEvent::decode(&encoded)?;
assert_eq!(decoded.id(), None);
assert_eq!(decoded.data(), "{\"a\":1}");
Ok(())
}
#[test]
fn a_comment_line_dispatches_nothing() -> Result<(), TestError> {
for block in [":", ": keep-alive", "", "\n\n", ": one\n: two\n\n"] {
assert_eq!(
SseEvent::dispatch(block)?,
None,
"a comment-only block dispatched an event"
);
assert_eq!(SseEvent::decode(block), Err(SseError::NoData));
}
let frame = SseEvent::decode("id: 2\n: a comment\ndata: payload")?;
assert_eq!(frame.id(), Some(EventId::new(2)));
assert_eq!(frame.data(), "payload");
Ok(())
}
#[test]
fn a_corrupt_id_is_an_error_and_not_a_zero() -> Result<(), TestError> {
for block in [
"id: abc\ndata: x",
"id: \ndata: x",
"id: 12x\ndata: x",
"id: 1.0\ndata: x",
"id: 99999999999999999999\ndata: x",
] {
assert_eq!(
SseEvent::decode(block),
Err(SseError::MalformedId),
"{block:?} was accepted"
);
}
assert_eq!(
SseEvent::decode("id: 0\ndata: x")?.id(),
Some(EventId::new(0))
);
Ok(())
}
#[test]
fn a_json_payload_is_parsed_as_json_by_a_reader() -> Result<(), TestError> {
let value = json!({"text": "before\u{7f}after", "n": 1});
let frame = SseEvent::json(EventId::new(9), &value)?;
let encoded = frame.encode();
assert!(
!encoded.contains("\\u{"),
"the payload carries Rust debug escaping: {encoded:?}"
);
let decoded = SseEvent::decode(&encoded)?;
let reparsed: Value = serde_json::from_str(decoded.data())?;
assert_eq!(reparsed, value);
Ok(())
}
#[test]
fn unknown_fields_are_ignored() -> Result<(), TestError> {
let frame =
SseEvent::decode("event: message\nretry: 5000\nid: 5\nwhatever: x\ndata: payload")?;
assert_eq!(frame.id(), Some(EventId::new(5)));
assert_eq!(frame.data(), "payload");
Ok(())
}
#[test]
fn exactly_one_space_after_the_colon_is_framing() -> Result<(), TestError> {
assert_eq!(SseEvent::decode("data: two")?.data(), " two");
assert_eq!(SseEvent::decode("data:two")?.data(), "two");
assert_eq!(
SseEvent::dispatch("data")?,
Some(SseEvent::without_id(String::new()))
);
Ok(())
}
#[test]
fn concatenated_frames_read_back_as_the_same_events() -> Result<(), TestError> {
let frames = [
SseEvent::new(EventId::new(1), "one".to_owned()),
SseEvent::new(EventId::new(2), "two\nlines".to_owned()),
SseEvent::without_id("three".to_owned()),
];
let stream: String = frames.iter().map(SseEvent::encode).collect();
let mut decoded = Vec::new();
for block in stream.trim_end_matches('\n').split("\n\n") {
if let Some(event) = SseEvent::dispatch(block)? {
decoded.push(event);
}
}
assert_eq!(decoded, frames.to_vec());
Ok(())
}
#[test]
fn a_named_frame_round_trips_its_name() -> Result<(), TestError> {
let frame = SseEvent::named(names::ENDPOINT, "/messages?sessionId=abc".to_string());
let decoded = SseEvent::decode(&frame.encode())?;
if !decoded.is_named(&names::ENDPOINT) {
return Err("the name did not survive the round trip".into());
}
if decoded.data() != "/messages?sessionId=abc" {
return Err("the payload changed across the round trip".into());
}
if decoded.is_named(&names::MESSAGE) {
return Err("an endpoint frame was mistaken for a message".into());
}
Ok(())
}
#[test]
fn an_unnamed_frame_stays_unnamed() -> Result<(), TestError> {
let frame = SseEvent::new(EventId::new(7), "{}".to_string());
let decoded = SseEvent::decode(&frame.encode())?;
if decoded.name().is_some() {
return Err("an unnamed frame came back named".into());
}
if decoded.is_named(&names::MESSAGE) {
return Err("an unnamed frame matched a name".into());
}
Ok(())
}
#[test]
fn the_two_legacy_frames_are_told_apart_by_name() -> Result<(), TestError> {
let wire = concat!(
"event: endpoint\ndata: /messages?sessionId=abc\n\n",
"event: message\ndata: {\"jsonrpc\":\"2.0\"}\n\n",
);
let mut frames = Vec::new();
for block in wire.split("\n\n").filter(|b| !b.trim().is_empty()) {
frames.push(SseEvent::decode(block)?);
}
let [endpoint, message] = frames.as_slice() else {
return Err(format!("expected two frames, found {}", frames.len()).into());
};
if !endpoint.is_named(&names::ENDPOINT) {
return Err("the first frame is not an endpoint frame".into());
}
if endpoint.data() != "/messages?sessionId=abc" {
return Err("the endpoint frame lost its URI".into());
}
if !message.is_named(&names::MESSAGE) {
return Err("the second frame is not a message frame".into());
}
Ok(())
}
#[test]
fn a_terminator_after_a_name_is_a_separator_and_not_part_of_it() -> Result<(), TestError> {
let block = "event: message\ndata: {\"jsonrpc\":\"2.0\"}\n\n";
let frame = SseEvent::decode(block)?;
if !frame.is_named(&names::MESSAGE) {
return Err("the name was not read as `message`".into());
}
if frame.data() != "{\"jsonrpc\":\"2.0\"}" {
return Err(format!("the payload was read as {:?}", frame.data()).into());
}
Ok(())
}
#[test]
fn an_empty_event_field_is_refused() -> Result<(), TestError> {
let block = "event:\ndata: x\n\n";
if SseEvent::decode(block) != Err(SseError::MalformedName(EventNameError::Empty)) {
return Err("an empty event field was not refused".into());
}
Ok(())
}
#[test]
fn the_encoder_writes_the_name_on_its_own_line() -> Result<(), TestError> {
let encoded = SseEvent::named(names::MESSAGE, "{}".to_string()).encode();
if !encoded.starts_with("event: message\ndata: {}\n\n") {
return Err(format!("encoded as {encoded:?}").into());
}
Ok(())
}
#[test]
fn a_peer_supplied_name_matches_a_constant_one() -> Result<(), TestError> {
let parsed = EventName::from_str("message")?;
let frame = SseEvent::named(parsed, "{}".to_string());
if !frame.is_named(&names::MESSAGE) {
return Err("a parsed name did not match the constant".into());
}
Ok(())
}
}