1use std::fmt;
2
3use serde_json::Value;
4
5use crate::{CoreError, Envelope, ErrorCode, Kind, PROTOCOL_VERSION};
6
7#[derive(Debug, Clone, PartialEq, Eq, Hash)]
8pub struct BodyId(String);
9
10impl BodyId {
11 pub fn as_str(&self) -> &str {
12 &self.0
13 }
14}
15
16impl From<String> for BodyId {
17 fn from(value: String) -> Self {
18 Self(value)
19 }
20}
21
22impl From<&str> for BodyId {
23 fn from(value: &str) -> Self {
24 Self(value.to_owned())
25 }
26}
27
28impl fmt::Display for BodyId {
29 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
30 self.0.fmt(formatter)
31 }
32}
33
34#[derive(Debug, Clone, PartialEq)]
35pub struct ApplicationHead {
36 pub v: u16,
37 pub id: String,
38 pub target: String,
39 pub subject: String,
40 pub kind: Kind,
41 pub corr: Option<String>,
42 pub seq: Option<u64>,
43 pub hops: Option<u8>,
44 pub path: Vec<String>,
45 pub headers: serde_json::Map<String, Value>,
46 pub error: Option<ApplicationError>,
47}
48
49impl ApplicationHead {
50 pub fn from_envelope(envelope: &Envelope) -> Result<Self, CoreError> {
51 if !is_application_kind(envelope.kind) {
52 return Err(CoreError::BadKind(format!("{:?}", envelope.kind)));
53 }
54 let error = if envelope.kind == Kind::Error {
55 let payload = envelope.payload_json();
56 Some(ApplicationError {
57 code: serde_json::from_value(payload["code"].clone())
58 .unwrap_or(ErrorCode::Protocol),
59 message: payload["message"]
60 .as_str()
61 .unwrap_or("protocol error")
62 .to_owned(),
63 })
64 } else {
65 None
66 };
67 Ok(Self {
68 v: envelope.v,
69 id: envelope.id.clone(),
70 target: envelope.target.clone(),
71 subject: envelope.subject.clone(),
72 kind: envelope.kind,
73 corr: envelope.corr.clone(),
74 seq: envelope.seq,
75 hops: envelope.hops,
76 path: envelope.path.clone(),
77 headers: envelope.headers.clone(),
78 error,
79 })
80 }
81
82 pub fn into_envelope(self) -> Envelope {
83 let payload = self
84 .error
85 .as_ref()
86 .map(|error| {
87 Envelope::encode_payload(&serde_json::json!({
88 "code": error.code,
89 "message": error.message,
90 }))
91 })
92 .unwrap_or_default();
93 Envelope {
94 v: self.v,
95 id: self.id,
96 target: self.target,
97 subject: self.subject,
98 kind: self.kind,
99 corr: self.corr,
100 seq: self.seq,
101 hops: self.hops,
102 body_token: None,
103 payload,
104 path: self.path,
105 headers: self.headers,
106 }
107 }
108}
109
110impl Default for ApplicationHead {
111 fn default() -> Self {
112 Self {
113 v: PROTOCOL_VERSION,
114 id: String::new(),
115 target: String::new(),
116 subject: String::new(),
117 kind: Kind::Request,
118 corr: None,
119 seq: None,
120 hops: None,
121 path: Vec::new(),
122 headers: serde_json::Map::new(),
123 error: None,
124 }
125 }
126}
127
128#[derive(Debug, Clone, PartialEq, Eq)]
129pub struct ApplicationError {
130 pub code: ErrorCode,
131 pub message: String,
132}
133
134#[derive(Debug, Clone, PartialEq)]
135pub struct ApplicationFrame {
136 pub head: ApplicationHead,
137 pub body: Option<BodyId>,
138}
139
140impl ApplicationFrame {
141 pub fn new(head: ApplicationHead, body: Option<BodyId>) -> Self {
142 Self { head, body }
143 }
144
145 pub fn from_envelope(envelope: &Envelope) -> Result<Self, CoreError> {
146 Ok(Self {
147 head: ApplicationHead::from_envelope(envelope)?,
148 body: envelope.body_token.clone().map(BodyId::from),
149 })
150 }
151
152 pub fn into_envelope(self) -> Envelope {
153 let mut envelope = self.head.into_envelope();
154 envelope.body_token = self.body.map(|body| body.to_string());
155 envelope
156 }
157}
158
159fn is_application_kind(kind: Kind) -> bool {
160 kind.is_application_request() || kind.is_application_response() || kind == Kind::Cancel
161}
162
163#[cfg(test)]
164mod tests {
165 use bytes::Bytes;
166
167 use super::*;
168
169 #[test]
170 fn application_head_drops_payload_and_transport_body_token() {
171 let source = Envelope {
172 v: PROTOCOL_VERSION,
173 id: "f1".into(),
174 target: "node-a".into(),
175 subject: "files.upload".into(),
176 kind: Kind::Request,
177 corr: Some("s1".into()),
178 seq: None,
179 hops: Some(4),
180 body_token: Some("transport-token".into()),
181 payload: Bytes::from_static(b"secret application bytes"),
182 path: vec!["edge".into()],
183 headers: serde_json::Map::from_iter([(
184 "content-type".into(),
185 Value::String("application/octet-stream".into()),
186 )]),
187 };
188
189 let frame = ApplicationFrame::new(
190 ApplicationHead::from_envelope(&source).unwrap(),
191 Some(BodyId::from("body-1")),
192 );
193 let projected = frame.into_envelope();
194
195 assert!(projected.payload.is_empty());
196 assert_eq!(projected.body_token.as_deref(), Some("body-1"));
197 assert_eq!(projected.target, source.target);
198 assert_eq!(projected.subject, source.subject);
199 assert_eq!(projected.corr, source.corr);
200 assert_eq!(projected.headers, source.headers);
201 }
202
203 #[test]
204 fn protocol_error_metadata_survives_without_exposing_its_encoded_document() {
205 let source = Envelope {
206 kind: Kind::Error,
207 payload: Envelope::encode_payload(&serde_json::json!({
208 "code": ErrorCode::Busy,
209 "message": "capacity exhausted",
210 })),
211 ..ApplicationHead::default().into_envelope()
212 };
213 let frame = ApplicationFrame::new(ApplicationHead::from_envelope(&source).unwrap(), None);
214 assert_eq!(
215 frame.head.error,
216 Some(ApplicationError {
217 code: ErrorCode::Busy,
218 message: "capacity exhausted".into(),
219 })
220 );
221 assert_eq!(frame.into_envelope().payload, source.payload);
222 }
223
224 #[test]
225 fn control_frames_cannot_cross_the_application_boundary() {
226 let control = Envelope {
227 kind: Kind::Hello,
228 ..ApplicationHead::default().into_envelope()
229 };
230 assert!(matches!(
231 ApplicationHead::from_envelope(&control),
232 Err(CoreError::BadKind(_))
233 ));
234 }
235}