1use serde::{Deserialize, Serialize};
2
3use crate::error::{Error, Result};
4use crate::field_map::FieldMap;
5
6#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
7pub struct ContextError {
8 kind: String,
9 message: String,
10}
11
12impl ContextError {
13 pub fn new(kind: impl Into<String>, message: impl Into<String>) -> Self {
14 Self {
15 kind: kind.into(),
16 message: message.into(),
17 }
18 }
19
20 pub fn kind(&self) -> &str {
21 &self.kind
22 }
23
24 pub fn message(&self) -> &str {
25 &self.message
26 }
27}
28
29impl From<Error> for ContextError {
30 fn from(err: Error) -> Self {
31 Self::from(&err)
32 }
33}
34
35impl From<&Error> for ContextError {
36 fn from(err: &Error) -> Self {
37 let (kind, message) = match err {
38 Error::InvalidTopic(m) => ("invalid_topic", m.clone()),
39 Error::InvalidNamespace(m) => ("invalid_namespace", m.clone()),
40 Error::InvalidOrganization(m) => ("invalid_organization", m.clone()),
41 Error::InvalidMetadataKey(m) => ("invalid_metadata_key", m.clone()),
42 Error::InvalidPayload(m) => ("invalid_payload", m.clone()),
43 Error::InvalidEventKey(m) => ("invalid_event_key", m.clone()),
44 Error::InvalidConsumerGroupId(m) => ("invalid_consumer_group_id", m.clone()),
45 Error::InvalidOwnerId(m) => ("invalid_owner_id", m.clone()),
46 Error::OwnershipLost(m) => ("ownership_lost", m.clone()),
47 Error::InvalidStartFrom(m) => ("invalid_start_from", m.clone()),
48 Error::InvalidCursor(m) => ("invalid_cursor", m.clone()),
49 Error::Serialization(m) => ("serialization", m.clone()),
50 Error::Store(m) => ("store", m.clone()),
51 Error::Handler(m) => ("handler", m.clone()),
52 Error::Timeout(m) => ("timeout", m.clone()),
53 Error::Config(m) => ("config", m.clone()),
54 };
55 Self::new(kind, message)
56 }
57}
58
59#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
60#[serde(tag = "type", content = "value", rename_all = "snake_case")]
61pub enum ContextValue {
62 String(String),
63 Bool(bool),
64 U64(u64),
65 I64(i64),
66 F64(f64),
67 Error(ContextError),
68 Json(serde_json::Value),
69}
70
71impl ContextValue {
72 fn should_store(&self) -> bool {
73 match self {
74 ContextValue::F64(v) => v.is_finite(),
75 _ => true,
76 }
77 }
78}
79
80impl From<String> for ContextValue {
81 fn from(value: String) -> Self {
82 ContextValue::String(value)
83 }
84}
85
86impl From<&str> for ContextValue {
87 fn from(value: &str) -> Self {
88 ContextValue::String(value.to_owned())
89 }
90}
91
92impl From<bool> for ContextValue {
93 fn from(value: bool) -> Self {
94 ContextValue::Bool(value)
95 }
96}
97
98impl From<u8> for ContextValue {
99 fn from(value: u8) -> Self {
100 ContextValue::U64(value.into())
101 }
102}
103
104impl From<u16> for ContextValue {
105 fn from(value: u16) -> Self {
106 ContextValue::U64(value.into())
107 }
108}
109
110impl From<u32> for ContextValue {
111 fn from(value: u32) -> Self {
112 ContextValue::U64(value.into())
113 }
114}
115
116impl From<u64> for ContextValue {
117 fn from(value: u64) -> Self {
118 ContextValue::U64(value)
119 }
120}
121
122impl From<usize> for ContextValue {
123 fn from(value: usize) -> Self {
124 ContextValue::U64(value as u64)
125 }
126}
127
128impl From<i8> for ContextValue {
129 fn from(value: i8) -> Self {
130 ContextValue::I64(value.into())
131 }
132}
133
134impl From<i16> for ContextValue {
135 fn from(value: i16) -> Self {
136 ContextValue::I64(value.into())
137 }
138}
139
140impl From<i32> for ContextValue {
141 fn from(value: i32) -> Self {
142 ContextValue::I64(value.into())
143 }
144}
145
146impl From<i64> for ContextValue {
147 fn from(value: i64) -> Self {
148 ContextValue::I64(value)
149 }
150}
151
152impl From<isize> for ContextValue {
153 fn from(value: isize) -> Self {
154 ContextValue::I64(value as i64)
155 }
156}
157
158impl From<f32> for ContextValue {
159 fn from(value: f32) -> Self {
160 ContextValue::F64(value.into())
161 }
162}
163
164impl From<f64> for ContextValue {
165 fn from(value: f64) -> Self {
166 ContextValue::F64(value)
167 }
168}
169
170impl From<ContextError> for ContextValue {
171 fn from(value: ContextError) -> Self {
172 ContextValue::Error(value)
173 }
174}
175
176impl From<Error> for ContextValue {
177 fn from(value: Error) -> Self {
178 ContextValue::Error(ContextError::from(value))
179 }
180}
181
182impl From<&Error> for ContextValue {
183 fn from(value: &Error) -> Self {
184 ContextValue::Error(ContextError::from(value))
185 }
186}
187
188impl From<serde_json::Value> for ContextValue {
189 fn from(value: serde_json::Value) -> Self {
190 ContextValue::Json(value)
191 }
192}
193
194#[derive(Debug, Clone, Serialize, Deserialize)]
195pub struct Context {
196 message: String,
197 fields: FieldMap<ContextValue>,
198}
199
200impl Context {
201 pub fn new(message: impl Into<String>) -> Self {
202 Self {
203 message: message.into(),
204 fields: FieldMap::new(),
205 }
206 }
207
208 pub fn with<V: Into<ContextValue>>(mut self, key: impl Into<String>, value: V) -> Result<Self> {
209 let key = key.into();
210 FieldMap::<ContextValue>::validate_key(&key)?;
211 let value = value.into();
212 if !value.should_store() {
213 return Ok(self);
214 }
215 self.fields = self.fields.with(key, value)?;
216 Ok(self)
217 }
218
219 pub fn message(&self) -> &str {
220 &self.message
221 }
222
223 pub fn get(&self, key: &str) -> Option<&ContextValue> {
224 self.fields.get(key)
225 }
226
227 pub fn has(&self, key: &str) -> bool {
228 self.fields.has(key)
229 }
230
231 pub fn fields(&self) -> &FieldMap<ContextValue> {
232 &self.fields
233 }
234}
235
236#[cfg(test)]
237mod tests {
238 use super::*;
239
240 #[test]
241 fn context_with_accepts_common_value_types() {
242 let ctx = Context::new("processing failed")
243 .with("handler_id", "billing")
244 .unwrap()
245 .with("retry", true)
246 .unwrap()
247 .with("attempt", 3u32)
248 .unwrap()
249 .with("offset", -5i64)
250 .unwrap()
251 .with("ratio", 0.5f64)
252 .unwrap()
253 .with("source_error", Error::Handler("boom".to_owned()))
254 .unwrap();
255
256 assert_eq!(ctx.message(), "processing failed");
257 assert_eq!(
258 ctx.get("handler_id"),
259 Some(&ContextValue::String("billing".to_owned()))
260 );
261 assert_eq!(ctx.get("retry"), Some(&ContextValue::Bool(true)));
262 assert_eq!(ctx.get("attempt"), Some(&ContextValue::U64(3)));
263 assert_eq!(ctx.get("offset"), Some(&ContextValue::I64(-5)));
264 assert_eq!(ctx.get("ratio"), Some(&ContextValue::F64(0.5)));
265 assert!(matches!(
266 ctx.get("source_error"),
267 Some(ContextValue::Error(e)) if e.kind() == "handler" && e.message() == "boom"
268 ));
269 assert!(ctx.has("handler_id"));
270 }
271
272 #[test]
273 fn serde_json_value_keeps_json_variant_for_any_shape() {
274 let ctx = Context::new("payload")
275 .with(
276 "json_string",
277 serde_json::Value::String("billing".to_owned()),
278 )
279 .unwrap()
280 .with(
281 "json_object",
282 serde_json::json!({"nested": {"k": [1, 2, 3]}}),
283 )
284 .unwrap();
285
286 assert!(matches!(
287 ctx.get("json_string"),
288 Some(ContextValue::Json(serde_json::Value::String(s))) if s == "billing"
289 ));
290 assert!(matches!(
291 ctx.get("json_object"),
292 Some(ContextValue::Json(serde_json::Value::Object(_)))
293 ));
294
295 let json = serde_json::to_value(&ctx).unwrap();
296 assert_eq!(
297 json["fields"]["json_string"],
298 serde_json::json!({"type": "json", "value": "billing"})
299 );
300 }
301
302 #[test]
303 fn non_finite_f64_values_are_omitted() {
304 let ctx = Context::new("nan")
305 .with("nan_value", f64::NAN)
306 .unwrap()
307 .with("inf_value", f64::INFINITY)
308 .unwrap()
309 .with("neg_inf", f64::NEG_INFINITY)
310 .unwrap()
311 .with("finite", 1.5f64)
312 .unwrap();
313
314 assert!(!ctx.has("nan_value"));
315 assert!(!ctx.has("inf_value"));
316 assert!(!ctx.has("neg_inf"));
317 assert_eq!(ctx.get("finite"), Some(&ContextValue::F64(1.5)));
318 assert_eq!(ctx.get("nan_value"), None);
319 }
320
321 #[test]
322 fn context_value_serialization_is_tagged() {
323 let ctx = Context::new("processing failed")
324 .with("handler_id", "billing")
325 .unwrap()
326 .with("attempt", 3u32)
327 .unwrap();
328
329 let json = serde_json::to_value(&ctx).unwrap();
330 assert_eq!(
331 json["message"],
332 serde_json::Value::String("processing failed".to_owned())
333 );
334 assert_eq!(
335 json["fields"]["handler_id"],
336 serde_json::json!({"type": "string", "value": "billing"})
337 );
338 assert_eq!(
339 json["fields"]["attempt"],
340 serde_json::json!({"type": "u64", "value": 3})
341 );
342 }
343
344 #[test]
345 fn context_roundtrips_through_json() {
346 let ctx = Context::new("processing failed")
347 .with("handler_id", "billing")
348 .unwrap()
349 .with("attempt", 3u32)
350 .unwrap()
351 .with("ratio", 0.25f64)
352 .unwrap()
353 .with("source_error", Error::Store("disk full".to_owned()))
354 .unwrap()
355 .with("payload", serde_json::json!({"k": "v"}))
356 .unwrap();
357
358 let serialized = serde_json::to_string(&ctx).unwrap();
359 let deserialized: Context = serde_json::from_str(&serialized).unwrap();
360
361 assert_eq!(deserialized.message(), ctx.message());
362 assert_eq!(deserialized.get("handler_id"), ctx.get("handler_id"));
363 assert_eq!(deserialized.get("attempt"), ctx.get("attempt"));
364 assert_eq!(deserialized.get("ratio"), ctx.get("ratio"));
365 assert_eq!(deserialized.get("source_error"), ctx.get("source_error"));
366 assert_eq!(deserialized.get("payload"), ctx.get("payload"));
367 }
368}