eventuary_core/io/acker/
nack_context.rs

1use serde::{Deserialize, Serialize};
2
3use crate::context::{Context, ContextValue};
4use crate::error::{Error, Result};
5
6#[derive(Debug, Clone, Copy, Eq, PartialEq, Hash, Serialize, Deserialize)]
7#[serde(rename_all = "snake_case")]
8pub enum NackReason {
9    HandlerError,
10    HandlerTimeout,
11    ProcessingRejected,
12    DeliveryExpired,
13    RouteFailed,
14    Unknown,
15}
16
17#[derive(Debug, Clone, Serialize, Deserialize)]
18pub struct NackContext {
19    reason: NackReason,
20    context: Context,
21}
22
23impl NackContext {
24    pub fn new(reason: NackReason, context: Context) -> Self {
25        Self { reason, context }
26    }
27
28    pub fn message(reason: NackReason, message: impl Into<String>) -> Self {
29        Self::new(reason, Context::new(message))
30    }
31
32    pub fn handler_error(handler_id: impl Into<String>, error: Error) -> Result<Self> {
33        let context = Context::new("handler failed")
34            .with("handler_id", handler_id.into())?
35            .with("error", error)?;
36        Ok(Self::new(NackReason::HandlerError, context))
37    }
38
39    pub fn handler_timeout(
40        handler_id: impl Into<String>,
41        message: impl Into<String>,
42    ) -> Result<Self> {
43        let context = Context::new("handler timed out")
44            .with("handler_id", handler_id.into())?
45            .with("error", Error::Timeout(message.into()))?;
46        Ok(Self::new(NackReason::HandlerTimeout, context))
47    }
48
49    pub fn delivery_expired(message: impl Into<String>) -> Self {
50        Self::message(NackReason::DeliveryExpired, message)
51    }
52
53    pub fn route_failed(destination: impl Into<String>, error: Error) -> Result<Self> {
54        let context = Context::new("route failed")
55            .with("destination", destination.into())?
56            .with("error", error)?;
57        Ok(Self::new(NackReason::RouteFailed, context))
58    }
59
60    pub fn processing_rejected(message: impl Into<String>) -> Result<Self> {
61        Ok(Self::message(NackReason::ProcessingRejected, message))
62    }
63
64    pub fn with<V: Into<ContextValue>>(mut self, key: impl Into<String>, value: V) -> Result<Self> {
65        self.context = self.context.with(key, value)?;
66        Ok(self)
67    }
68
69    pub fn reason(&self) -> NackReason {
70        self.reason
71    }
72
73    pub fn context(&self) -> &Context {
74        &self.context
75    }
76}
77
78impl Default for NackContext {
79    fn default() -> Self {
80        Self::message(NackReason::Unknown, "message nacked")
81    }
82}
83
84impl From<Error> for NackContext {
85    fn from(error: Error) -> Self {
86        let context = Context::new("message nacked")
87            .with("error", error)
88            .expect("valid context key");
89        Self::new(NackReason::Unknown, context)
90    }
91}
92
93#[cfg(test)]
94mod tests {
95    use super::*;
96
97    #[test]
98    fn handler_error_builds_context_fields() {
99        let nack =
100            NackContext::handler_error("billing", Error::Handler("boom".to_owned())).unwrap();
101
102        assert_eq!(nack.reason(), NackReason::HandlerError);
103        assert_eq!(nack.context().message(), "handler failed");
104        assert_eq!(
105            nack.context().get("handler_id"),
106            Some(&ContextValue::String("billing".to_owned()))
107        );
108        assert!(matches!(
109            nack.context().get("error"),
110            Some(ContextValue::Error(e)) if e.kind() == "handler" && e.message() == "boom"
111        ));
112    }
113
114    #[test]
115    fn route_failed_builds_destination_and_error_fields() {
116        let nack = NackContext::route_failed("dlq", Error::Store("offline".to_owned())).unwrap();
117
118        assert_eq!(nack.reason(), NackReason::RouteFailed);
119        assert_eq!(nack.context().message(), "route failed");
120        assert_eq!(
121            nack.context().get("destination"),
122            Some(&ContextValue::String("dlq".to_owned()))
123        );
124        assert!(matches!(
125            nack.context().get("error"),
126            Some(ContextValue::Error(e)) if e.kind() == "store" && e.message() == "offline"
127        ));
128    }
129
130    #[test]
131    fn nack_context_roundtrips_json() {
132        let nack = NackContext::handler_timeout("billing", "exceeded 5s")
133            .unwrap()
134            .with("attempt", 2u32)
135            .unwrap();
136
137        let serialized = serde_json::to_string(&nack).unwrap();
138        let deserialized: NackContext = serde_json::from_str(&serialized).unwrap();
139
140        assert_eq!(deserialized.reason(), NackReason::HandlerTimeout);
141        assert_eq!(deserialized.context().message(), "handler timed out");
142        assert_eq!(
143            deserialized.context().get("handler_id"),
144            Some(&ContextValue::String("billing".to_owned()))
145        );
146        assert_eq!(
147            deserialized.context().get("attempt"),
148            Some(&ContextValue::U64(2))
149        );
150        assert!(matches!(
151            deserialized.context().get("error"),
152            Some(ContextValue::Error(e)) if e.kind() == "timeout" && e.message() == "exceeded 5s"
153        ));
154    }
155}
eventuary_core/io/acker/nack_context.rs

eventuary_core/io/acker/
nack_context.rs
