use std::fmt;
use serde_json::Value;
use crate::{CoreError, Envelope, ErrorCode, Kind, PROTOCOL_VERSION};
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct BodyId(String);
impl BodyId {
pub fn as_str(&self) -> &str {
&self.0
}
}
impl From<String> for BodyId {
fn from(value: String) -> Self {
Self(value)
}
}
impl From<&str> for BodyId {
fn from(value: &str) -> Self {
Self(value.to_owned())
}
}
impl fmt::Display for BodyId {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
self.0.fmt(formatter)
}
}
#[derive(Debug, Clone, PartialEq)]
pub struct ApplicationHead {
pub v: u16,
pub id: String,
pub target: String,
pub subject: String,
pub kind: Kind,
pub corr: Option<String>,
pub seq: Option<u64>,
pub hops: Option<u8>,
pub path: Vec<String>,
pub headers: serde_json::Map<String, Value>,
pub error: Option<ApplicationError>,
}
impl ApplicationHead {
pub fn from_envelope(envelope: &Envelope) -> Result<Self, CoreError> {
if !is_application_kind(envelope.kind) {
return Err(CoreError::BadKind(format!("{:?}", envelope.kind)));
}
let error = if envelope.kind == Kind::Error {
let payload = envelope.payload_json();
Some(ApplicationError {
code: serde_json::from_value(payload["code"].clone())
.unwrap_or(ErrorCode::Protocol),
message: payload["message"]
.as_str()
.unwrap_or("protocol error")
.to_owned(),
})
} else {
None
};
Ok(Self {
v: envelope.v,
id: envelope.id.clone(),
target: envelope.target.clone(),
subject: envelope.subject.clone(),
kind: envelope.kind,
corr: envelope.corr.clone(),
seq: envelope.seq,
hops: envelope.hops,
path: envelope.path.clone(),
headers: envelope.headers.clone(),
error,
})
}
pub fn into_envelope(self) -> Envelope {
let payload = self
.error
.as_ref()
.map(|error| {
Envelope::encode_payload(&serde_json::json!({
"code": error.code,
"message": error.message,
}))
})
.unwrap_or_default();
Envelope {
v: self.v,
id: self.id,
target: self.target,
subject: self.subject,
kind: self.kind,
corr: self.corr,
seq: self.seq,
hops: self.hops,
body_token: None,
payload,
path: self.path,
headers: self.headers,
}
}
}
impl Default for ApplicationHead {
fn default() -> Self {
Self {
v: PROTOCOL_VERSION,
id: String::new(),
target: String::new(),
subject: String::new(),
kind: Kind::Request,
corr: None,
seq: None,
hops: None,
path: Vec::new(),
headers: serde_json::Map::new(),
error: None,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ApplicationError {
pub code: ErrorCode,
pub message: String,
}
#[derive(Debug, Clone, PartialEq)]
pub struct ApplicationFrame {
pub head: ApplicationHead,
pub body: Option<BodyId>,
}
impl ApplicationFrame {
pub fn new(head: ApplicationHead, body: Option<BodyId>) -> Self {
Self { head, body }
}
pub fn from_envelope(envelope: &Envelope) -> Result<Self, CoreError> {
Ok(Self {
head: ApplicationHead::from_envelope(envelope)?,
body: envelope.body_token.clone().map(BodyId::from),
})
}
pub fn into_envelope(self) -> Envelope {
let mut envelope = self.head.into_envelope();
envelope.body_token = self.body.map(|body| body.to_string());
envelope
}
}
fn is_application_kind(kind: Kind) -> bool {
kind.is_application_request() || kind.is_application_response() || kind == Kind::Cancel
}
#[cfg(test)]
mod tests {
use bytes::Bytes;
use super::*;
#[test]
fn application_head_drops_payload_and_transport_body_token() {
let source = Envelope {
v: PROTOCOL_VERSION,
id: "f1".into(),
target: "node-a".into(),
subject: "files.upload".into(),
kind: Kind::Request,
corr: Some("s1".into()),
seq: None,
hops: Some(4),
body_token: Some("transport-token".into()),
payload: Bytes::from_static(b"secret application bytes"),
path: vec!["edge".into()],
headers: serde_json::Map::from_iter([(
"content-type".into(),
Value::String("application/octet-stream".into()),
)]),
};
let frame = ApplicationFrame::new(
ApplicationHead::from_envelope(&source).unwrap(),
Some(BodyId::from("body-1")),
);
let projected = frame.into_envelope();
assert!(projected.payload.is_empty());
assert_eq!(projected.body_token.as_deref(), Some("body-1"));
assert_eq!(projected.target, source.target);
assert_eq!(projected.subject, source.subject);
assert_eq!(projected.corr, source.corr);
assert_eq!(projected.headers, source.headers);
}
#[test]
fn protocol_error_metadata_survives_without_exposing_its_encoded_document() {
let source = Envelope {
kind: Kind::Error,
payload: Envelope::encode_payload(&serde_json::json!({
"code": ErrorCode::Busy,
"message": "capacity exhausted",
})),
..ApplicationHead::default().into_envelope()
};
let frame = ApplicationFrame::new(ApplicationHead::from_envelope(&source).unwrap(), None);
assert_eq!(
frame.head.error,
Some(ApplicationError {
code: ErrorCode::Busy,
message: "capacity exhausted".into(),
})
);
assert_eq!(frame.into_envelope().payload, source.payload);
}
#[test]
fn control_frames_cannot_cross_the_application_boundary() {
let control = Envelope {
kind: Kind::Hello,
..ApplicationHead::default().into_envelope()
};
assert!(matches!(
ApplicationHead::from_envelope(&control),
Err(CoreError::BadKind(_))
));
}
}