use bytes::{BufMut, Bytes, BytesMut};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use crate::{CoreError, TargetPath};
pub const PROTOCOL_VERSION: u16 = 1;
pub const DEFAULT_HOPS: u8 = 8;
const CONTROL_TARGET: &str = "*";
const MAX_HEADERS: usize = 64;
const MAX_HEAD_BYTES: usize = 64 * 1024;
pub const UNB_VERSION: &str = "unb-version";
pub const UNB_KIND: &str = "unb-kind";
pub const UNB_ID: &str = "unb-id";
pub const UNB_CORR: &str = "unb-corr";
pub const UNB_SEQ: &str = "unb-seq";
pub const UNB_HOPS: &str = "unb-hops";
pub const UNB_PATH: &str = "unb-path";
pub const UNB_CODE: &str = "unb-code";
pub const UNB_BODY: &str = "unb-body";
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Kind {
Hello,
Welcome,
Request,
Response,
Subscribe,
Event,
Channel,
Cancel,
Error,
Ping,
Pong,
Discover,
Identify,
IdentityAccepted,
RouteSnapshot,
RouteDelta,
RouteAck,
}
impl Kind {
pub fn token(self) -> &'static str {
match self {
Kind::Hello => "hello",
Kind::Welcome => "welcome",
Kind::Request => "request",
Kind::Response => "response",
Kind::Subscribe => "subscribe",
Kind::Event => "event",
Kind::Channel => "channel",
Kind::Cancel => "cancel",
Kind::Error => "error",
Kind::Ping => "ping",
Kind::Pong => "pong",
Kind::Discover => "discover",
Kind::Identify => "identify",
Kind::IdentityAccepted => "identity_accepted",
Kind::RouteSnapshot => "route_snapshot",
Kind::RouteDelta => "route_delta",
Kind::RouteAck => "route_ack",
}
}
pub fn from_token(token: &str) -> Option<Kind> {
Some(match token {
"hello" => Kind::Hello,
"welcome" => Kind::Welcome,
"request" => Kind::Request,
"response" => Kind::Response,
"subscribe" => Kind::Subscribe,
"event" => Kind::Event,
"channel" => Kind::Channel,
"cancel" => Kind::Cancel,
"error" => Kind::Error,
"ping" => Kind::Ping,
"pong" => Kind::Pong,
"discover" => Kind::Discover,
"identify" => Kind::Identify,
"identity_accepted" => Kind::IdentityAccepted,
"route_snapshot" => Kind::RouteSnapshot,
"route_delta" => Kind::RouteDelta,
"route_ack" => Kind::RouteAck,
_ => return None,
})
}
pub fn is_application_request(self) -> bool {
matches!(
self,
Kind::Request | Kind::Subscribe | Kind::Channel | Kind::Discover
)
}
pub fn is_application_response(self) -> bool {
matches!(self, Kind::Response | Kind::Event | Kind::Error)
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Envelope {
pub v: u16,
pub id: String,
#[serde(default, skip_serializing_if = "String::is_empty")]
pub target: String,
#[serde(default, skip_serializing_if = "String::is_empty")]
pub subject: String,
pub kind: Kind,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub corr: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub seq: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub hops: Option<u8>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub body_token: Option<String>,
#[serde(skip)]
pub payload: Bytes,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub path: Vec<String>,
#[serde(flatten)]
pub headers: serde_json::Map<String, Value>,
}
impl Envelope {
pub fn decode(frame: Bytes) -> Result<Envelope, CoreError> {
let (mut envelope, head_len) = Envelope::decode_head(&frame)?;
envelope.payload = frame.slice(head_len..);
Ok(envelope)
}
pub fn decode_head(frame: &[u8]) -> Result<(Envelope, usize), CoreError> {
if frame.len() > MAX_HEAD_BYTES && !Envelope::head_within_bounds(frame) {
return Err(CoreError::Malformed(
"header section exceeds the maximum head size".into(),
));
}
let mut envelope = Envelope {
v: PROTOCOL_VERSION,
id: String::new(),
target: String::new(),
subject: String::new(),
kind: Kind::Request,
corr: None,
seq: None,
hops: None,
body_token: None,
payload: Bytes::new(),
path: Vec::new(),
headers: serde_json::Map::new(),
};
let mut storage = [httparse::EMPTY_HEADER; MAX_HEADERS];
let is_status = frame.starts_with(b"HTTP/");
let (head_len, status, target) = if is_status {
let mut response = httparse::Response::new(&mut storage);
let head_len = match response.parse(frame).map_err(Envelope::parse_error)? {
httparse::Status::Complete(head_len) => head_len,
httparse::Status::Partial => {
return Err(CoreError::Malformed("frame head is incomplete".into()))
}
};
let code = response
.code
.ok_or_else(|| CoreError::Malformed("status line carries no status".into()))?;
(head_len, Some(code), None)
} else {
let mut request = httparse::Request::new(&mut storage);
let head_len = match request.parse(frame).map_err(Envelope::parse_error)? {
httparse::Status::Complete(head_len) => head_len,
httparse::Status::Partial => {
return Err(CoreError::Malformed("frame head is incomplete".into()))
}
};
if request.method != Some("POST") {
return Err(CoreError::Malformed(
"application and control frames use POST".into(),
));
}
let target = request
.path
.ok_or_else(|| CoreError::Malformed("request line carries no target".into()))?;
(head_len, None, Some(target.to_string()))
};
let az_kind = envelope.absorb_wire_headers(&storage, is_status)?;
if let Some(code) = status {
if az_kind.is_some() {
return Err(CoreError::Malformed(
"status-line frames carry no unb-kind".into(),
));
}
envelope.kind = match code {
200 if envelope.seq.is_some() => Kind::Event,
200 => Kind::Response,
400..=599 => Kind::Error,
other => {
return Err(CoreError::Malformed(format!(
"status {other} maps to no frame kind"
)));
}
};
} else {
let target = target.expect("request path present");
if target.contains(['?', '#']) {
return Err(CoreError::Malformed(
"targets must not contain uri query or fragment delimiters".into(),
));
}
if target == CONTROL_TARGET {
let control = az_kind
.ok_or_else(|| CoreError::Malformed("control frames carry unb-kind".into()))?;
if control.is_application_request() || control.is_application_response() {
return Err(CoreError::Malformed(format!(
"{control:?} is not a control kind"
)));
}
envelope.kind = control;
} else if target.starts_with('/') {
let kind = az_kind.unwrap_or(Kind::Request);
if !kind.is_application_request() {
return Err(CoreError::Malformed(format!(
"{kind:?} is not an application request kind"
)));
}
let target_path = if kind == Kind::Discover {
TargetPath::parse_discovery(&target)?
} else {
TargetPath::parse_application(&target)?
};
envelope.target = target_path.target().to_owned();
envelope.subject = target_path.subject().to_owned();
envelope.kind = kind;
} else {
return Err(CoreError::Malformed(
"request targets are origin-form and start with a slash".into(),
));
}
}
Ok((envelope, head_len))
}
pub fn encode(&self) -> Bytes {
use std::fmt::Write;
let mut head = String::with_capacity(64 + self.headers.len() * 32);
if self.kind.is_application_response() {
let (status, reason) = match self.kind {
Kind::Error => match self.error_code() {
Some(code) => (code.status().as_u16(), code.token()),
None => (500, "INTERNAL"),
},
_ => (200, "OK"),
};
write!(head, "HTTP/1.1 {status} {reason}\r\n")
.expect("writing to a string cannot fail");
} else if self.kind.is_application_request() {
let target_path = if self.kind == Kind::Discover {
TargetPath::discovery(self.target.clone())
} else {
TargetPath::application(self.target.clone(), self.subject.clone())
}
.expect("application envelopes carry a valid target path");
write!(head, "POST {target_path} HTTP/1.1\r\n")
.expect("writing to a string cannot fail");
} else {
write!(head, "POST {CONTROL_TARGET} HTTP/1.1\r\n")
.expect("writing to a string cannot fail");
}
let line = |head: &mut String, name: &str, value: &str| {
debug_assert!(
!value.contains(['\r', '\n']),
"header values are single-line strings"
);
head.push_str(name);
head.push_str(": ");
head.push_str(value);
head.push_str("\r\n");
};
line(&mut head, UNB_VERSION, &self.v.to_string());
if !self.id.is_empty() {
line(&mut head, UNB_ID, &self.id);
}
if let Some(corr) = &self.corr {
line(&mut head, UNB_CORR, corr);
}
if let Some(seq) = self.seq {
line(&mut head, UNB_SEQ, &seq.to_string());
}
if let Some(hops) = self.hops {
line(&mut head, UNB_HOPS, &hops.to_string());
}
if let Some(token) = &self.body_token {
line(&mut head, UNB_BODY, token);
}
if !self.path.is_empty() {
line(&mut head, UNB_PATH, &self.path.join(","));
}
if !self.kind.is_application_response() && self.kind != Kind::Request {
line(&mut head, UNB_KIND, self.kind.token());
}
if self.kind == Kind::Error {
if let Some(code) = self.payload_json()["code"].as_str() {
line(&mut head, UNB_CODE, code);
}
}
for (name, value) in &self.headers {
let value = value
.as_str()
.expect("application headers carry string values on the wire");
line(&mut head, name, value);
}
head.push_str("\r\n");
let mut frame = BytesMut::with_capacity(head.len() + self.payload.len());
frame.put_slice(head.as_bytes());
frame.put_slice(&self.payload);
frame.freeze()
}
fn head_within_bounds(frame: &[u8]) -> bool {
frame
.windows(4)
.take(MAX_HEAD_BYTES)
.position(|window| window == b"\r\n\r\n")
.is_some_and(|position| position + 4 <= MAX_HEAD_BYTES)
}
fn parse_error(error: httparse::Error) -> CoreError {
CoreError::Malformed(error.to_string())
}
fn absorb_wire_headers(
&mut self,
headers: &[httparse::Header],
is_status: bool,
) -> Result<Option<Kind>, CoreError> {
let mut az_kind = None;
let mut az_version_seen = false;
for header in headers {
if header.name.is_empty() {
continue;
}
let name = header.name.to_ascii_lowercase();
let value = std::str::from_utf8(header.value)
.map_err(|source| CoreError::Malformed(source.to_string()))?
.trim_start_matches(' ');
let duplicate = || CoreError::Malformed(format!("{name}: duplicate header"));
match name.as_str() {
"content-length" | "transfer-encoding" => {
return Err(CoreError::Malformed(format!(
"{name}: message-carrier frames carry no explicit body framing"
)));
}
UNB_VERSION => {
if az_version_seen {
return Err(duplicate());
}
az_version_seen = true;
self.v = value
.parse()
.map_err(|_| CoreError::Malformed(format!("{UNB_VERSION}: {value:?}")))?;
}
UNB_ID => {
if !self.id.is_empty() {
return Err(duplicate());
}
self.id = value.to_string();
}
UNB_CORR => {
if self.corr.is_some() {
return Err(duplicate());
}
self.corr = Some(value.to_string());
}
UNB_SEQ => {
if self.seq.is_some() {
return Err(duplicate());
}
self.seq = Some(
value
.parse()
.map_err(|_| CoreError::Malformed(format!("{UNB_SEQ}: {value:?}")))?,
);
}
UNB_HOPS => {
if self.hops.is_some() {
return Err(duplicate());
}
self.hops = Some(
value
.parse()
.map_err(|_| CoreError::Malformed(format!("{UNB_HOPS}: {value:?}")))?,
);
}
UNB_PATH => {
if !self.path.is_empty() {
return Err(duplicate());
}
self.path = value.split(',').map(str::to_string).collect();
}
UNB_BODY => {
if self.body_token.is_some() {
return Err(duplicate());
}
if value.is_empty() || value.len() > 256 {
return Err(CoreError::Malformed(format!(
"{UNB_BODY}: token length out of bounds"
)));
}
self.body_token = Some(value.to_string());
}
UNB_KIND => {
if az_kind.is_some() {
return Err(duplicate());
}
az_kind = Some(Kind::from_token(value).ok_or_else(|| {
CoreError::Malformed(format!("{UNB_KIND}: {value:?} is not a frame kind"))
})?);
}
UNB_CODE if is_status => {}
other if other.starts_with("unb-") => {
return Err(CoreError::Malformed(format!(
"{other}: unknown reserved header"
)));
}
_ => {
if self
.headers
.insert(name.clone(), Value::String(value.to_string()))
.is_some()
{
return Err(duplicate());
}
}
}
}
Ok(az_kind)
}
fn error_code(&self) -> Option<crate::ErrorCode> {
serde_json::from_value(self.payload_json()["code"].clone()).ok()
}
#[inline]
pub fn payload_json(&self) -> Value {
if self.payload.is_empty() {
return Value::Null;
}
serde_json::from_slice(&self.payload).unwrap_or(Value::Null)
}
#[inline]
pub fn encode_payload(value: &Value) -> Bytes {
if value.is_null() {
return Bytes::new();
}
serde_json::to_vec(value)
.expect("payload is plain json data")
.into()
}
#[inline]
pub fn parse_payload<T: serde::de::DeserializeOwned>(&self) -> Result<T, CoreError> {
serde_json::from_slice(&self.payload)
.map_err(|source| CoreError::Malformed(source.to_string()))
}
pub fn to_request(&self) -> Result<http::Request<Bytes>, CoreError> {
if !self.kind.is_application_request() {
return Err(CoreError::Malformed(format!(
"{:?} is not an application request kind",
self.kind
)));
}
let target_path = if self.kind == Kind::Discover {
TargetPath::discovery(self.target.clone())?
} else {
TargetPath::application(self.target.clone(), self.subject.clone())?
};
let mut request = http::Request::builder()
.method(http::Method::POST)
.uri(target_path.to_string())
.body(self.payload.clone())
.map_err(|source| CoreError::Malformed(source.to_string()))?;
if self.kind != Kind::Request {
request
.headers_mut()
.insert(UNB_KIND, http::HeaderValue::from_static(self.kind.token()));
}
self.project_reserved(request.headers_mut())?;
Self::project_custom(request.headers_mut(), &self.headers)?;
Ok(request)
}
pub fn to_local_request(&self) -> Result<http::Request<Bytes>, CoreError> {
if !matches!(self.kind, Kind::Request | Kind::Subscribe | Kind::Channel) {
return Err(CoreError::Malformed(format!(
"{:?} is not a destination-local handler request kind",
self.kind
)));
}
let target_path = TargetPath::application(self.target.clone(), self.subject.clone())?;
let mut request = http::Request::builder()
.method(http::Method::POST)
.uri(format!("/{}", target_path.subject().replace('.', "/")))
.body(self.payload.clone())
.map_err(|source| CoreError::Malformed(source.to_string()))?;
if self.kind != Kind::Request {
request
.headers_mut()
.insert(UNB_KIND, http::HeaderValue::from_static(self.kind.token()));
}
self.project_reserved(request.headers_mut())?;
Self::project_custom(request.headers_mut(), &self.headers)?;
Ok(request)
}
pub fn subject_of(uri: &http::Uri) -> String {
let path = uri.path().trim_start_matches('/');
if path.is_empty() {
uri.authority()
.map(|authority| authority.as_str().replace('/', "."))
.unwrap_or_default()
} else {
path.replace('/', ".")
}
}
pub fn ensure_headers_wire_safe(
headers: &serde_json::Map<String, Value>,
) -> Result<(), CoreError> {
Self::project_custom(&mut http::HeaderMap::new(), headers)
}
pub fn from_request(request: http::Request<Bytes>) -> Result<Envelope, CoreError> {
if request.method() != http::Method::POST {
return Err(CoreError::Malformed(format!(
"application requests use POST, not {}",
request.method()
)));
}
if request.uri().query().is_some() {
return Err(CoreError::Malformed(
"application request targets must not carry a query string".into(),
));
}
let kind = match request.headers().get(UNB_KIND) {
None => Kind::Request,
Some(value) => {
let value = value
.to_str()
.map_err(|source| CoreError::Malformed(format!("{UNB_KIND}: {source}")))?;
match Kind::from_token(value) {
Some(kind) if kind.is_application_request() => kind,
_ => {
return Err(CoreError::Malformed(format!(
"{UNB_KIND}: {value:?} is not an application request kind"
)))
}
}
}
};
let (parts, payload) = request.into_parts();
let target_path = if kind == Kind::Discover {
TargetPath::parse_discovery(parts.uri.path())?
} else {
TargetPath::parse_application(parts.uri.path())?
};
let mut envelope = Envelope {
v: PROTOCOL_VERSION,
id: String::new(),
target: target_path.target().to_owned(),
subject: target_path.subject().to_owned(),
kind,
corr: None,
seq: None,
hops: None,
body_token: None,
payload,
path: Vec::new(),
headers: serde_json::Map::new(),
};
envelope.absorb_headers(&parts.headers, false)?;
Ok(envelope)
}
pub fn to_response(&self) -> Result<http::Response<Bytes>, CoreError> {
let status = match self.kind {
Kind::Response | Kind::Event => http::StatusCode::OK,
Kind::Error => self
.error_code()
.ok_or_else(|| {
CoreError::Malformed("error frame payload carries no known code".into())
})?
.status(),
other => {
return Err(CoreError::Malformed(format!(
"{other:?} is not an application response kind"
)))
}
};
if !self.subject.is_empty() {
return Err(CoreError::Malformed(
"response frames carry no subject".into(),
));
}
if self.kind == Kind::Response && self.seq.is_some() {
return Err(CoreError::Malformed(
"a unary response carries no seq; seq marks stream events".into(),
));
}
if self.kind == Kind::Event && self.seq.is_none() {
return Err(CoreError::Malformed(
"an event frame requires seq to remain distinguishable".into(),
));
}
let mut response = http::Response::builder()
.status(status)
.body(self.payload.clone())
.map_err(|source| CoreError::Malformed(source.to_string()))?;
self.project_reserved(response.headers_mut())?;
if self.kind == Kind::Error {
let code = self.payload_json()["code"].clone();
if let Some(code) = code.as_str() {
response.headers_mut().insert(
UNB_CODE,
http::HeaderValue::from_str(code)
.map_err(|source| CoreError::Malformed(source.to_string()))?,
);
}
}
Self::project_custom(response.headers_mut(), &self.headers)?;
Ok(response)
}
pub fn from_response(response: http::Response<Bytes>) -> Result<Envelope, CoreError> {
let status = response.status();
let (parts, payload) = response.into_parts();
let kind = if status == http::StatusCode::OK {
if parts.headers.contains_key(UNB_SEQ) {
Kind::Event
} else {
Kind::Response
}
} else if status.is_client_error() || status.is_server_error() {
Kind::Error
} else {
return Err(CoreError::Malformed(format!(
"status {status} maps to no application response kind"
)));
};
let mut envelope = Envelope {
v: PROTOCOL_VERSION,
id: String::new(),
target: String::new(),
subject: String::new(),
kind,
corr: None,
seq: None,
hops: None,
body_token: None,
payload,
path: Vec::new(),
headers: serde_json::Map::new(),
};
envelope.absorb_headers(&parts.headers, true)?;
Ok(envelope)
}
fn project_reserved(&self, headers: &mut http::HeaderMap) -> Result<(), CoreError> {
let mut put = |name: &'static str, value: String| -> Result<(), CoreError> {
let value = http::HeaderValue::from_str(&value)
.map_err(|source| CoreError::Malformed(format!("{name}: {source}")))?;
headers.insert(name, value);
Ok(())
};
put(UNB_VERSION, self.v.to_string())?;
if !self.id.is_empty() {
put(UNB_ID, self.id.clone())?;
}
if let Some(corr) = &self.corr {
put(UNB_CORR, corr.clone())?;
}
if let Some(seq) = self.seq {
put(UNB_SEQ, seq.to_string())?;
}
if let Some(hops) = self.hops {
put(UNB_HOPS, hops.to_string())?;
}
if !self.path.is_empty() {
if self.path.iter().any(|hop| hop.contains(',')) {
return Err(CoreError::Malformed(
"path elements must not contain commas".into(),
));
}
put(UNB_PATH, self.path.join(","))?;
}
Ok(())
}
fn project_custom(
headers: &mut http::HeaderMap,
custom: &serde_json::Map<String, Value>,
) -> Result<(), CoreError> {
for (name, value) in custom {
let name = http::HeaderName::try_from(name.as_str())
.map_err(|source| CoreError::Malformed(source.to_string()))?;
if name.as_str().starts_with("unb-") {
return Err(CoreError::Malformed(format!(
"{name}: unb-* header names are reserved for framing metadata"
)));
}
let Value::String(value) = value else {
return Err(CoreError::Malformed(format!(
"{name}: application-frame header values must be strings"
)));
};
let value = http::HeaderValue::from_str(value)
.map_err(|source| CoreError::Malformed(format!("{name}: {source}")))?;
headers.insert(name, value);
}
Ok(())
}
fn absorb_headers(
&mut self,
headers: &http::HeaderMap,
response: bool,
) -> Result<(), CoreError> {
for name in headers.keys() {
let mut values = headers.get_all(name).iter();
let value = values.next().expect("keys() yields present names");
if values.next().is_some() {
return Err(CoreError::Malformed(format!(
"{name}: duplicate header values are not representable"
)));
}
let value = std::str::from_utf8(value.as_bytes())
.map_err(|source| CoreError::Malformed(format!("{name}: {source}")))?;
match name.as_str() {
UNB_VERSION => {
self.v = value
.parse()
.map_err(|_| CoreError::Malformed(format!("{name}: {value:?}")))?;
}
UNB_ID => self.id = value.to_string(),
UNB_CORR => self.corr = Some(value.to_string()),
UNB_SEQ => {
self.seq = Some(
value
.parse()
.map_err(|_| CoreError::Malformed(format!("{name}: {value:?}")))?,
);
}
UNB_HOPS => {
self.hops = Some(
value
.parse()
.map_err(|_| CoreError::Malformed(format!("{name}: {value:?}")))?,
);
}
UNB_PATH => {
self.path = value.split(',').map(str::to_string).collect();
}
UNB_CODE if response => {}
UNB_KIND if !response => {}
other if other.starts_with("unb-") => {
return Err(CoreError::Malformed(format!(
"{other}: unknown reserved header"
)));
}
other => {
self.headers
.insert(other.to_string(), Value::String(value.to_string()));
}
}
}
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
fn hex_bytes(hex: &str) -> Bytes {
let hex = hex.trim();
(0..hex.len())
.step_by(2)
.map(|i| u8::from_str_radix(&hex[i..i + 2], 16).expect("hex fixture"))
.collect::<Vec<u8>>()
.into()
}
fn envelope(kind: Kind) -> Envelope {
Envelope {
v: PROTOCOL_VERSION,
id: "f1".into(),
target: if kind.is_application_request() {
"node-a".into()
} else {
String::new()
},
subject: String::new(),
kind,
corr: None,
seq: None,
hops: None,
body_token: None,
payload: Bytes::new(),
path: Vec::new(),
headers: Default::default(),
}
}
#[test]
fn a_v1_request_line_frame_round_trips() {
let mut headers = serde_json::Map::new();
headers.insert("authorization".into(), json!("Bearer jwt-abc"));
let request = Envelope {
v: PROTOCOL_VERSION,
id: "f2".into(),
target: "node-b".into(),
subject: "chess.move".into(),
kind: Kind::Request,
corr: Some("s1".into()),
seq: None,
hops: Some(DEFAULT_HOPS),
body_token: None,
payload: Envelope::encode_payload(&json!({"from": "e2", "to": "e4"})),
path: vec!["node-a".into()],
headers,
};
let frame = request.encode();
let text = std::str::from_utf8(&frame).unwrap();
assert_eq!(
text,
"POST /node-b/chess/move HTTP/1.1\r\nunb-version: 1\r\nunb-id: f2\r\nunb-corr: s1\r\nunb-hops: 8\r\n\
unb-path: node-a\r\nauthorization: Bearer jwt-abc\r\n\r\n\
{\"from\":\"e2\",\"to\":\"e4\"}"
);
assert_eq!(Envelope::decode(frame).unwrap(), request);
}
#[test]
fn every_kind_takes_its_lane_on_the_wire() {
for kind in [
Kind::Hello,
Kind::Welcome,
Kind::Request,
Kind::Response,
Kind::Subscribe,
Kind::Event,
Kind::Channel,
Kind::Cancel,
Kind::Error,
Kind::Ping,
Kind::Pong,
Kind::Discover,
Kind::Identify,
Kind::IdentityAccepted,
Kind::RouteSnapshot,
Kind::RouteDelta,
Kind::RouteAck,
] {
let mut envelope = envelope(kind);
if kind.is_application_request() && kind != Kind::Discover {
envelope.subject = "chess".into();
envelope.corr = Some("s1".into());
}
if kind == Kind::Discover {
envelope.corr = Some("s1".into());
}
if kind.is_application_response() {
envelope.corr = Some("s1".into());
}
if kind == Kind::Event {
envelope.seq = Some(7);
}
if kind == Kind::Error {
envelope.payload =
Envelope::encode_payload(&json!({"code": "BUSY", "message": "full"}));
}
let frame = envelope.encode();
let text = std::str::from_utf8(&frame).unwrap();
if kind.is_application_response() {
assert!(text.starts_with("HTTP/1.1 "), "{kind:?}: {text}");
} else if kind == Kind::Discover {
assert!(
text.starts_with("POST /node-a HTTP/1.1"),
"{kind:?}: {text}"
);
} else if kind.is_application_request() {
assert!(
text.starts_with("POST /node-a/chess HTTP/1.1"),
"{kind:?}: {text}"
);
} else {
assert!(text.starts_with("POST * HTTP/1.1"), "{kind:?}: {text}");
}
assert_eq!(Envelope::decode(frame).unwrap(), envelope, "{kind:?}");
}
}
#[test]
fn a_control_frame_routes_by_az_kind_over_the_asterisk_target() {
let mut hello = envelope(Kind::Hello);
hello.payload = Envelope::encode_payload(&json!({"versions": [1]}));
assert_eq!(
std::str::from_utf8(&hello.encode()).unwrap(),
"POST * HTTP/1.1\r\nunb-version: 1\r\nunb-id: f1\r\nunb-kind: hello\r\n\r\n{\"versions\":[1]}"
);
}
#[test]
fn a_ping_frame_has_an_empty_body() {
let ping = envelope(Kind::Ping);
let frame = ping.encode();
assert_eq!(
std::str::from_utf8(&frame).unwrap(),
"POST * HTTP/1.1\r\nunb-version: 1\r\nunb-id: f1\r\nunb-kind: ping\r\n\r\n"
);
assert_eq!(Envelope::decode(frame).unwrap(), ping);
}
#[test]
fn an_error_status_line_names_the_error_code() {
let mut error = envelope(Kind::Error);
error.corr = Some("s1".into());
error.payload =
Envelope::encode_payload(&json!({"code": "BUSY", "message": "node at capacity"}));
let frame = error.encode();
let text = std::str::from_utf8(&frame).unwrap();
assert!(text.starts_with("HTTP/1.1 503 BUSY\r\n"), "{text}");
assert!(text.contains("\r\nunb-code: BUSY\r\n"), "{text}");
assert_eq!(Envelope::decode(frame).unwrap(), error);
}
#[test]
fn every_error_code_has_canonical_wire_status_reason_and_restoration() {
for (code, status) in [
(crate::ErrorCode::VersionMismatch, 505),
(crate::ErrorCode::Protocol, 400),
(crate::ErrorCode::UnknownSubject, 404),
(crate::ErrorCode::UnresolvedAtPeer, 421),
(crate::ErrorCode::PeerUnreachable, 502),
(crate::ErrorCode::HopLimitExceeded, 508),
(crate::ErrorCode::InvalidInput, 400),
(crate::ErrorCode::Unauthorized, 401),
(crate::ErrorCode::Conflict, 409),
(crate::ErrorCode::Busy, 503),
(crate::ErrorCode::Cancelled, 499),
(crate::ErrorCode::Internal, 500),
] {
let mut error = envelope(Kind::Error);
error.corr = Some("s1".into());
error.payload = Envelope::encode_payload(&json!({
"code": code.token(),
"message": "failure"
}));
let frame = error.encode();
let text = std::str::from_utf8(&frame).unwrap();
assert!(
text.starts_with(&format!("HTTP/1.1 {status} {}\r\n", code.token())),
"{code:?}: {text}"
);
assert!(
text.contains(&format!("\r\nunb-code: {}\r\n", code.token())),
"{code:?}: {text}"
);
let decoded = Envelope::decode(frame.clone()).unwrap();
assert_eq!(decoded, error, "{code:?}");
let restored = Envelope::from_response(decoded.to_response().unwrap()).unwrap();
assert_eq!(restored.payload_json()["code"], code.token(), "{code:?}");
assert_eq!(restored.encode(), frame, "{code:?}");
}
}
#[test]
fn an_unmapped_error_payload_still_frames_as_a_server_error() {
let mut error = envelope(Kind::Error);
error.corr = Some("s1".into());
error.payload = Envelope::encode_payload(&json!({"note": "no code field"}));
let frame = error.encode();
let text = std::str::from_utf8(&frame).unwrap();
assert!(text.starts_with("HTTP/1.1 500 INTERNAL\r\n"), "{text}");
let decoded = Envelope::decode(frame).unwrap();
assert_eq!(decoded.kind, Kind::Error);
}
#[test]
fn decode_slices_the_body_zero_copy() {
let mut request = envelope(Kind::Request);
request.subject = "chess".into();
request.corr = Some("s1".into());
request.payload = Envelope::encode_payload(&json!({"from": "e2"}));
let frame = request.encode();
let body_start = frame.len() - request.payload.len();
let decoded = Envelope::decode(frame.clone()).unwrap();
assert_eq!(decoded.payload.as_ptr(), frame[body_start..].as_ptr());
}
#[test]
fn dot_and_slash_subject_forms_are_equivalent() {
for uri in ["/node-a/chess.move", "/node-a/chess/move"] {
let request = http::Request::builder()
.method("POST")
.uri(uri)
.body(Bytes::new())
.unwrap();
let envelope = Envelope::from_request(request).unwrap();
assert_eq!(envelope.target, "node-a");
assert_eq!(envelope.subject, "chess.move");
}
}
#[test]
fn a_malformed_head_is_a_typed_decode_error() {
for frame in [
&b"GARBAGE\r\n\r\n"[..],
b"POST /x HTTP/1.1\r\nunb-id: f1",
b"GET /x HTTP/1.1\r\n\r\n",
b"POST x HTTP/1.1\r\n\r\n",
b"POST /x?side=w HTTP/1.1\r\nunb-version: 1\r\n\r\n",
b"POST /az/teleport HTTP/1.1\r\nunb-version: 1\r\n\r\n",
b"POST /az HTTP/1.1\r\nunb-version: 1\r\n\r\n",
b"POST /az/move HTTP/1.1\r\nunb-version: 1\r\n\r\n",
b"POST /x HTTP/1.1\r\nunb-kind: response\r\n\r\n",
b"POST /x HTTP/1.1\r\nunb-magic: 1\r\n\r\n",
b"POST /x HTTP/1.1\r\nunb-corr: a\r\nunb-corr: b\r\n\r\n",
b"POST /x HTTP/1.1\r\nactor: a\r\nactor: b\r\n\r\n",
b"POST * HTTP/1.1\r\nunb-kind: request\r\n\r\n",
b"POST * HTTP/1.1\r\n\r\n",
b"POST /x HTTP/1.1\r\ncontent-length: 5\r\n\r\nhello",
b"POST /x HTTP/1.1\r\ntransfer-encoding: chunked\r\n\r\n",
b"HTTP/1.1 302 FOUND\r\n\r\n",
b"HTTP/1.1 200 OK\r\nunb-kind: response\r\n\r\n",
] {
let error = Envelope::decode(Bytes::copy_from_slice(frame)).unwrap_err();
assert!(
matches!(error, CoreError::Malformed(_)),
"{:?}: {error}",
std::str::from_utf8(frame)
);
}
}
#[test]
fn decode_rejects_bare_carriage_returns_and_accepts_supported_line_endings() {
for frame in [
&b"POST /x\rHTTP/1.1\r\n\r\n"[..],
&b"POST /x HTTP/1.1\r\nact\ror: a\r\n\r\n"[..],
&b"POST /x HTTP/1.1\r\nactor: a\rb\r\n\r\n"[..],
] {
assert!(matches!(
Envelope::decode(Bytes::copy_from_slice(frame)),
Err(CoreError::Malformed(_))
));
}
assert_eq!(
Envelope::decode(Bytes::from_static(
b"POST /node-a/x HTTP/1.1\r\nactor: a\r\n\r\n",
))
.unwrap()
.headers["actor"],
"a"
);
}
#[test]
fn the_exact_head_bound_is_accepted_and_one_byte_over_is_rejected() {
let prefix = "POST /node-a/x HTTP/1.1\r\nactor: ";
let suffix = "\r\n\r\n";
let value_len = MAX_HEAD_BYTES - prefix.len() - suffix.len();
let exact = format!("{prefix}{}{suffix}", "x".repeat(value_len));
assert_eq!(exact.len(), MAX_HEAD_BYTES);
assert_eq!(
Envelope::decode(Bytes::from(exact)).unwrap().headers["actor"],
"x".repeat(value_len)
);
let text = format!("{prefix}{}{suffix}", "x".repeat(value_len + 1));
assert_eq!(text.len(), MAX_HEAD_BYTES + 1);
assert!(matches!(
Envelope::decode(Bytes::from(text)),
Err(CoreError::Malformed(_))
));
}
#[test]
fn custom_headers_stay_headers_not_core_fields() {
let frame = Bytes::from_static(
b"POST /node-a/chess HTTP/1.1\r\nunb-corr: s1\r\nsubject: sneaky\r\n\r\n",
);
let decoded = Envelope::decode(frame).unwrap();
assert_eq!(decoded.target, "node-a");
assert_eq!(decoded.subject, "chess");
assert_eq!(decoded.corr.as_deref(), Some("s1"));
assert_eq!(decoded.headers["subject"], "sneaky");
}
#[test]
fn a_non_json_payload_round_trips_verbatim() {
let mut opaque = envelope(Kind::Request);
opaque.subject = "files".into();
opaque.corr = Some("s1".into());
opaque.payload = Bytes::from_static(&[0x00, 0x01, 0xff, 0xfe, b'!', 0x80]);
let decoded = Envelope::decode(opaque.encode()).unwrap();
assert_eq!(decoded, opaque);
assert_eq!(decoded.payload_json(), Value::Null);
}
#[test]
fn a_custom_header_round_trips_verbatim() {
let mut headers = serde_json::Map::new();
headers.insert("actor".into(), json!("jwt-abc"));
headers.insert("x-trace".into(), json!("span-7"));
let mut request = envelope(Kind::Request);
request.subject = "chess".into();
request.corr = Some("s1".into());
request.headers = headers;
let decoded = Envelope::decode(request.encode()).unwrap();
assert_eq!(decoded, request);
assert_eq!(decoded.headers["actor"], "jwt-abc");
}
#[test]
fn a_mixed_case_unb_header_is_rejected_as_reserved() {
for name in ["Unb-Corr", "UNB-KIND", "uNb-hOpS"] {
let mut headers = serde_json::Map::new();
headers.insert(name.into(), json!("forged"));
assert!(matches!(
Envelope::ensure_headers_wire_safe(&headers),
Err(CoreError::Malformed(message)) if message.contains("reserved")
));
let mut request = envelope(Kind::Request);
request.subject = "chess".into();
request.corr = Some("s1".into());
request.headers.insert(name.into(), json!("forged"));
assert!(matches!(
request.to_request(),
Err(CoreError::Malformed(message)) if message.contains("reserved")
));
}
}
#[test]
fn an_application_envelope_round_trips_through_the_request_model() {
let mut headers = serde_json::Map::new();
headers.insert("authorization".into(), json!("Bearer jwt-abc"));
let envelope = Envelope {
v: PROTOCOL_VERSION,
id: "f7".into(),
target: "node-a".into(),
subject: "chess.move".into(),
kind: Kind::Request,
corr: Some("s1".into()),
seq: None,
hops: Some(DEFAULT_HOPS),
body_token: None,
payload: Envelope::encode_payload(&json!({"from": "e2", "to": "e4"})),
path: vec!["node-a".into()],
headers,
};
let request = envelope.to_request().unwrap();
assert_eq!(request.method(), http::Method::POST);
assert_eq!(request.uri().path(), "/node-a/chess/move");
assert_eq!(request.headers()["authorization"], "Bearer jwt-abc");
assert_eq!(request.headers()[UNB_CORR], "s1");
assert_eq!(request.headers()[UNB_ID], "f7");
let back = Envelope::from_request(request).unwrap();
assert_eq!(back.encode(), envelope.encode());
}
#[test]
fn every_request_kind_rides_az_kind_over_post_and_back() {
for (kind, marker) in [
(Kind::Request, None),
(Kind::Subscribe, Some("subscribe")),
(Kind::Channel, Some("channel")),
(Kind::Discover, Some("discover")),
] {
let mut envelope = envelope(kind);
if kind != Kind::Discover {
envelope.subject = "chess".into();
}
envelope.corr = Some("s1".into());
let request = envelope.to_request().unwrap();
assert_eq!(request.method(), http::Method::POST);
match marker {
Some(marker) => assert_eq!(request.headers()[UNB_KIND], marker),
None => assert!(!request.headers().contains_key(UNB_KIND)),
}
assert_eq!(Envelope::from_request(request).unwrap().kind, kind);
}
}
#[test]
fn request_conversion_rejects_what_the_model_forbids() {
let mut base = envelope(Kind::Request);
base.subject = "chess".into();
base.corr = Some("s1".into());
let mut control = base.clone();
control.kind = Kind::Ping;
assert!(matches!(control.to_request(), Err(CoreError::Malformed(_))));
let mut nested = base.clone();
nested.headers.insert("trace".into(), json!({"span": 7}));
assert!(matches!(nested.to_request(), Err(CoreError::Malformed(_))));
let mut shadowing = base.clone();
shadowing.headers.insert(UNB_CORR.into(), json!("spoof"));
assert!(matches!(
shadowing.to_request(),
Err(CoreError::Malformed(_))
));
let mut queried = base.clone();
queried.subject = "chess.move?side=white".into();
assert!(matches!(queried.to_request(), Err(CoreError::Malformed(_))));
let mut reserved = base.clone();
reserved.target = "az".into();
assert!(matches!(
reserved.to_request(),
Err(CoreError::Malformed(_))
));
let wrong_method = http::Request::builder()
.method("GET")
.uri("/node-a/chess")
.body(Bytes::new())
.unwrap();
assert!(matches!(
Envelope::from_request(wrong_method),
Err(CoreError::Malformed(_))
));
let unknown_kind = http::Request::builder()
.method("POST")
.uri("/node-a/chess")
.header(UNB_KIND, "teleport")
.body(Bytes::new())
.unwrap();
assert!(matches!(
Envelope::from_request(unknown_kind),
Err(CoreError::Malformed(_))
));
let queried = http::Request::builder()
.method("POST")
.uri("/node-a/chess/move?draft=1")
.body(Bytes::new())
.unwrap();
assert!(matches!(
Envelope::from_request(queried),
Err(CoreError::Malformed(message)) if message.contains("query string")
));
let reserved_subject = http::Request::builder()
.method("POST")
.uri("/az/hello")
.body(Bytes::new())
.unwrap();
assert!(matches!(
Envelope::from_request(reserved_subject),
Err(CoreError::Malformed(_))
));
}
#[test]
fn a_fresh_user_request_converts_with_wire_defaults() {
let request = http::Request::builder()
.method("POST")
.uri("/node-a/chess/move")
.header("authorization", "jwt-abc")
.body(Bytes::from_static(b"{}"))
.unwrap();
let envelope = Envelope::from_request(request).unwrap();
assert_eq!(envelope.v, PROTOCOL_VERSION);
assert_eq!(envelope.kind, Kind::Request);
assert_eq!(envelope.target, "node-a");
assert_eq!(envelope.subject, "chess.move");
assert!(envelope.id.is_empty());
assert_eq!(envelope.corr, None);
assert_eq!(envelope.headers["authorization"], "jwt-abc");
}
#[test]
fn model_conversion_shares_the_payload_allocation() {
let mut source = envelope(Kind::Request);
source.subject = "chess.move".into();
source.corr = Some("s1".into());
source.payload = Envelope::encode_payload(&json!({"from": "e2"}));
let request = source.to_request().unwrap();
assert_eq!(request.body().as_ptr(), source.payload.as_ptr());
let back = Envelope::from_request(request).unwrap();
assert_eq!(back.payload.as_ptr(), source.payload.as_ptr());
}
#[test]
fn response_event_and_error_envelopes_round_trip_through_the_response_model() {
let mut response = envelope(Kind::Response);
response.id = "f8".into();
response.corr = Some("s1".into());
response.payload = Envelope::encode_payload(&json!({"ok": true}));
let converted = response.to_response().unwrap();
assert_eq!(converted.status(), http::StatusCode::OK);
let back = Envelope::from_response(converted).unwrap();
assert_eq!(back.encode(), response.encode());
let mut event = response.clone();
event.kind = Kind::Event;
event.seq = Some(42);
let converted = event.to_response().unwrap();
assert_eq!(converted.headers()[UNB_SEQ], "42");
let back = Envelope::from_response(converted).unwrap();
assert_eq!(back.kind, Kind::Event);
assert_eq!(back.encode(), event.encode());
let mut error = response.clone();
error.kind = Kind::Error;
error.payload =
Envelope::encode_payload(&json!({"code": "BUSY", "message": "node at capacity"}));
let converted = error.to_response().unwrap();
assert_eq!(converted.status(), http::StatusCode::SERVICE_UNAVAILABLE);
assert_eq!(converted.headers()[UNB_CODE], "BUSY");
let back = Envelope::from_response(converted).unwrap();
assert_eq!(back.encode(), error.encode());
}
#[test]
fn reserved_fields_project_and_restore_across_legal_application_lanes() {
for kind in [Kind::Request, Kind::Event, Kind::Error] {
let mut source = envelope(kind);
source.id = "f-reserved".into();
source.subject = if kind == Kind::Request {
"chess.move".into()
} else {
String::new()
};
source.corr = Some("s-reserved".into());
source.seq = (kind == Kind::Event).then_some(17);
source.hops = Some(6);
source.path = vec!["leaf-a".into(), "hub-b".into(), "root-c".into()];
source.headers.insert("x-trace".into(), json!("span-7"));
source.payload = if kind == Kind::Error {
Envelope::encode_payload(&json!({"code": "PROTOCOL", "message": "bad"}))
} else {
Bytes::from_static(b"opaque")
};
let restored = if kind == Kind::Request {
let projected = source.to_request().unwrap();
assert_eq!(projected.headers()[UNB_VERSION], "1");
assert_eq!(projected.headers()[UNB_ID], "f-reserved");
assert_eq!(projected.headers()[UNB_CORR], "s-reserved");
assert_eq!(projected.headers()[UNB_HOPS], "6");
assert_eq!(projected.headers()[UNB_PATH], "leaf-a,hub-b,root-c");
assert_eq!(projected.headers()["x-trace"], "span-7");
Envelope::from_request(projected).unwrap()
} else {
let projected = source.to_response().unwrap();
assert_eq!(projected.headers()[UNB_VERSION], "1");
assert_eq!(projected.headers()[UNB_ID], "f-reserved");
assert_eq!(projected.headers()[UNB_CORR], "s-reserved");
assert_eq!(projected.headers()[UNB_HOPS], "6");
assert_eq!(projected.headers()[UNB_PATH], "leaf-a,hub-b,root-c");
assert_eq!(projected.headers()["x-trace"], "span-7");
if kind == Kind::Event {
assert_eq!(projected.headers()[UNB_SEQ], "17");
} else {
assert_eq!(projected.headers()[UNB_CODE], "PROTOCOL");
}
Envelope::from_response(projected).unwrap()
};
assert_eq!(restored, source, "{kind:?}");
}
}
#[test]
fn comma_path_elements_are_rejected_on_every_legal_projection_lane() {
for kind in [Kind::Request, Kind::Event, Kind::Error] {
let mut source = envelope(kind);
source.subject = if kind == Kind::Request {
"chess".into()
} else {
String::new()
};
source.corr = Some("s1".into());
source.seq = (kind == Kind::Event).then_some(1);
source.path = vec!["leaf-a,forged-hop".into()];
if kind == Kind::Error {
source.payload = Envelope::encode_payload(&json!({"code": "INTERNAL"}));
}
let result = if kind == Kind::Request {
source.to_request().map(|_| ())
} else {
source.to_response().map(|_| ())
};
assert!(matches!(result, Err(CoreError::Malformed(_))), "{kind:?}");
}
}
#[test]
fn error_status_collisions_restore_the_exact_code_from_az_code() {
for code in ["INVALID_INPUT", "PROTOCOL"] {
let mut error = envelope(Kind::Error);
error.id = "f9".into();
error.corr = Some("s1".into());
error.payload = Envelope::encode_payload(&json!({"code": code, "message": "bad"}));
let converted = error.to_response().unwrap();
assert_eq!(converted.status(), http::StatusCode::BAD_REQUEST);
assert_eq!(converted.headers()[UNB_CODE], code);
let back = Envelope::from_response(converted).unwrap();
assert_eq!(back.encode(), error.encode());
}
}
#[test]
fn header_validation_rejects_frame_splitting_input() {
let clean = {
let mut headers = serde_json::Map::new();
headers.insert("x-trace".into(), json!("span-7"));
headers
};
assert!(Envelope::ensure_headers_wire_safe(&clean).is_ok());
let crlf_value = {
let mut headers = serde_json::Map::new();
headers.insert("x-trace".into(), json!("span\r\nunb-corr: forged"));
headers
};
assert!(matches!(
Envelope::ensure_headers_wire_safe(&crlf_value),
Err(CoreError::Malformed(_))
));
let crlf_name = {
let mut headers = serde_json::Map::new();
headers.insert("x\r\nInjected".into(), json!("v"));
headers
};
assert!(matches!(
Envelope::ensure_headers_wire_safe(&crlf_name),
Err(CoreError::Malformed(_))
));
let non_string = {
let mut headers = serde_json::Map::new();
headers.insert("x-trace".into(), json!({ "nested": true }));
headers
};
assert!(matches!(
Envelope::ensure_headers_wire_safe(&non_string),
Err(CoreError::Malformed(_))
));
}
fn golden_envelopes() -> Vec<(&'static str, Envelope)> {
let build = |id: &str, subject: &str, kind: Kind, corr: Option<&str>| Envelope {
v: PROTOCOL_VERSION,
id: id.into(),
target: if kind.is_application_request() {
"node-a".into()
} else {
String::new()
},
subject: subject.into(),
kind,
corr: corr.map(str::to_string),
seq: None,
hops: None,
body_token: None,
payload: Bytes::new(),
path: Vec::new(),
headers: Default::default(),
};
let mut request = build("f2", "chess", Kind::Request, Some("s1"));
request.hops = Some(DEFAULT_HOPS);
request.payload = Envelope::encode_payload(
&json!({"action": "move", "input": {"from": "e2", "to": "e4"}}),
);
let mut event = build("f9", "", Kind::Event, Some("s1"));
event.seq = Some(3);
event.payload =
Envelope::encode_payload(&json!({"as_of": 4711, "value": {"done": false, "id": "t1"}}));
let mut response = build("f3", "", Kind::Response, Some("s1"));
response.payload = Envelope::encode_payload(&json!({"ok": true}));
let mut subscribe = build("f8", "todo.changes", Kind::Subscribe, Some("s2"));
subscribe.hops = Some(DEFAULT_HOPS);
subscribe.payload = Envelope::encode_payload(&json!({"after": 4711}));
let mut error = build("f4", "", Kind::Error, Some("s1"));
error.path = vec!["node-a".into(), "node-b".into()];
error.payload = Envelope::encode_payload(&json!({
"code": "UNKNOWN_SUBJECT",
"message": "Unknown subject \"ches\". Did you mean \"chess\"?"
}));
let ping = build("f1", "", Kind::Ping, None);
let mut identify = build("f1", "", Kind::Identify, None);
identify.payload = Envelope::encode_payload(
&json!({"node_id": "node-a", "instance_id": "node-a-1", "epoch": 1}),
);
let identity_accepted = build("f2", "", Kind::IdentityAccepted, None);
let mut route_snapshot = build("f3", "", Kind::RouteSnapshot, None);
route_snapshot.payload = Envelope::encode_payload(&json!({
"generation": 1,
"routes": [{
"subject": "chess", "owner": "leaf-a", "owner_instance": "leaf-a-1",
"owner_epoch": 1, "owner_revision": 0, "distance": 1,
"path": ["leaf-a", "hub"]
}]
}));
let mut route_delta = build("f4", "", Kind::RouteDelta, None);
route_delta.payload = Envelope::encode_payload(&json!({
"generation": 2,
"upsert": [],
"withdraw": [{
"subject": "chess", "owner": "leaf-a", "owner_instance": "leaf-a-1",
"owner_epoch": 1, "owner_revision": 0
}]
}));
let mut route_ack = build("f5", "", Kind::RouteAck, None);
route_ack.payload =
Envelope::encode_payload(&json!({"generation": 2, "status": "applied"}));
vec![
("request", request),
("response", response),
("subscribe", subscribe),
("event", event),
("error", error),
("ping", ping),
("identify", identify),
("identity_accepted", identity_accepted),
("route_snapshot", route_snapshot),
("route_delta", route_delta),
("route_ack", route_ack),
]
}
#[test]
#[ignore = "regenerates the golden fixtures from the canonical encoder"]
fn regenerate_golden_fixtures() {
let root = concat!(
env!("CARGO_MANIFEST_DIR"),
"/../../conformance/fixtures/envelope"
);
for (name, envelope) in golden_envelopes() {
let frame = envelope.encode();
let hex: String = frame.iter().map(|byte| format!("{byte:02x}")).collect();
std::fs::write(format!("{root}/{name}.frame.hex"), format!("{hex}\n")).unwrap();
std::fs::write(
format!("{root}/{name}.header.json"),
format!("{}\n", serde_json::to_string(&envelope).unwrap()),
)
.unwrap();
}
}
#[test]
fn az_body_is_a_core_field_on_the_wire_and_reserved_for_users() {
let mut headers = serde_json::Map::new();
headers.insert(UNB_BODY.into(), Value::String("token".into()));
assert!(matches!(
Envelope::ensure_headers_wire_safe(&headers),
Err(CoreError::Malformed(_))
));
let frame = Bytes::from_static(
b"POST /node-a/echo HTTP/1.1\r\nunb-id: a\r\nunb-body: token\r\n\r\n",
);
let decoded = Envelope::decode(frame).unwrap();
assert_eq!(decoded.body_token.as_deref(), Some("token"));
assert!(!decoded.headers.contains_key(UNB_BODY));
let round = Envelope::decode(decoded.encode()).unwrap();
assert_eq!(round.body_token.as_deref(), Some("token"));
let oversized = format!(
"POST /node-a/echo HTTP/1.1\r\nunb-id: a\r\nunb-body: {}\r\n\r\n",
"x".repeat(257)
);
assert!(matches!(
Envelope::decode(Bytes::from(oversized)),
Err(CoreError::Malformed(_))
));
}
#[test]
fn head_first_decode_matches_whole_message_decode() {
for (name, envelope) in golden_envelopes() {
let frame = envelope.encode();
let whole = Envelope::decode(frame.clone()).unwrap();
let (mut head, head_len) = Envelope::decode_head(&frame).unwrap();
assert!(
head.payload.is_empty(),
"{name}: head decode carries no payload"
);
head.payload = frame.slice(head_len..);
assert_eq!(head, whole, "{name}");
}
}
#[test]
fn golden_fixtures_round_trip_byte_identically() {
for (frame_hex, header_json) in [
(
include_str!("../../../conformance/fixtures/envelope/request.frame.hex"),
include_str!("../../../conformance/fixtures/envelope/request.header.json"),
),
(
include_str!("../../../conformance/fixtures/envelope/response.frame.hex"),
include_str!("../../../conformance/fixtures/envelope/response.header.json"),
),
(
include_str!("../../../conformance/fixtures/envelope/subscribe.frame.hex"),
include_str!("../../../conformance/fixtures/envelope/subscribe.header.json"),
),
(
include_str!("../../../conformance/fixtures/envelope/event.frame.hex"),
include_str!("../../../conformance/fixtures/envelope/event.header.json"),
),
(
include_str!("../../../conformance/fixtures/envelope/error.frame.hex"),
include_str!("../../../conformance/fixtures/envelope/error.header.json"),
),
(
include_str!("../../../conformance/fixtures/envelope/ping.frame.hex"),
include_str!("../../../conformance/fixtures/envelope/ping.header.json"),
),
(
include_str!("../../../conformance/fixtures/envelope/identify.frame.hex"),
include_str!("../../../conformance/fixtures/envelope/identify.header.json"),
),
(
include_str!("../../../conformance/fixtures/envelope/identity_accepted.frame.hex"),
include_str!(
"../../../conformance/fixtures/envelope/identity_accepted.header.json"
),
),
(
include_str!("../../../conformance/fixtures/envelope/route_snapshot.frame.hex"),
include_str!("../../../conformance/fixtures/envelope/route_snapshot.header.json"),
),
(
include_str!("../../../conformance/fixtures/envelope/route_delta.frame.hex"),
include_str!("../../../conformance/fixtures/envelope/route_delta.header.json"),
),
(
include_str!("../../../conformance/fixtures/envelope/route_ack.frame.hex"),
include_str!("../../../conformance/fixtures/envelope/route_ack.header.json"),
),
] {
let frame = hex_bytes(frame_hex);
let envelope = Envelope::decode(frame.clone()).unwrap();
assert_eq!(
serde_json::to_string(&envelope).unwrap(),
header_json.trim(),
"header fixture must be the canonical envelope encoding"
);
assert_eq!(
envelope.encode(),
frame,
"fixture must be the canonical frame encoding"
);
}
}
#[test]
fn application_golden_fixtures_survive_the_model_byte_identically() {
let request_frame = hex_bytes(include_str!(
"../../../conformance/fixtures/envelope/request.frame.hex"
));
let envelope = Envelope::decode(request_frame.clone()).unwrap();
let back = Envelope::from_request(envelope.to_request().unwrap()).unwrap();
assert_eq!(back.encode(), request_frame);
let subscribe_frame = hex_bytes(include_str!(
"../../../conformance/fixtures/envelope/subscribe.frame.hex"
));
let envelope = Envelope::decode(subscribe_frame.clone()).unwrap();
let back = Envelope::from_request(envelope.to_request().unwrap()).unwrap();
assert_eq!(back.encode(), subscribe_frame);
for fixture in [
include_str!("../../../conformance/fixtures/envelope/response.frame.hex"),
include_str!("../../../conformance/fixtures/envelope/event.frame.hex"),
include_str!("../../../conformance/fixtures/envelope/error.frame.hex"),
] {
let frame = hex_bytes(fixture);
let envelope = Envelope::decode(frame.clone()).unwrap();
let back = Envelope::from_response(envelope.to_response().unwrap()).unwrap();
assert_eq!(back.encode(), frame);
}
}
}