ruststream/runtime/
typed.rs

1//! Typed handler adapter: turns a [`Handler<T>`](Handler) over a decoded value into a
2//! [`Handler<M>`](Handler) by decoding the message payload via a [`Codec`].
3//!
4//! This is the decode boundary between the two middleware levels: raw (pre-decode) middleware
5//! wrap the produced `Handler<M>`; typed (post-decode) middleware wrap the `inner: Handler<T>`
6//! passed in here. Both use the same [`Layer`](super::Layer) / [`HandlerExt`](super::HandlerExt)
7//! machinery, just at different inputs.
8
9use std::{fmt, marker::PhantomData};
10
11use crate::IncomingMessage;
12use crate::codec::Codec;
13use serde::de::DeserializeOwned;
14use tracing::warn;
15
16use super::context::Context;
17use super::failure::FailurePolicy;
18use super::handler::{Handler, HandlerResult, Settle};
19
20/// Build a `Handler<M>` that decodes the payload with `codec` into `T` and forwards `&T` to
21/// `inner`.
22///
23/// `inner` is any [`Handler<T>`](Handler) - a closure `Fn(&T) -> _` or a typed middleware stack
24/// built with [`HandlerExt::with`](super::HandlerExt::with).
25pub fn typed<M, T, C, H>(codec: C, inner: H) -> Typed<M, T, C, H>
26where
27    M: IncomingMessage,
28    T: DeserializeOwned + Send + Sync,
29    C: Codec,
30{
31    Typed {
32        codec,
33        inner,
34        decode: FailurePolicy::Drop,
35        _phantom: PhantomData,
36    }
37}
38
39/// Handler produced by [`typed`]. Override the decode-failure policy with
40/// [`Typed::on_decode_failure`].
41pub struct Typed<M, T, C, H> {
42    codec: C,
43    inner: H,
44    decode: FailurePolicy,
45    _phantom: PhantomData<fn(M, T)>,
46}
47
48impl<M, T, C, H> Typed<M, T, C, H> {
49    /// Sets the [`FailurePolicy`] applied when the codec fails to decode an incoming payload. The
50    /// default is [`FailurePolicy::Drop`].
51    #[must_use]
52    pub fn on_decode_failure(mut self, decode: FailurePolicy) -> Self {
53        self.decode = decode;
54        self
55    }
56}
57
58impl<M, T, C, H> fmt::Debug for Typed<M, T, C, H> {
59    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
60        f.debug_struct("Typed")
61            .field("decode", &self.decode)
62            .finish_non_exhaustive()
63    }
64}
65
66impl<M, T, C, H, Cx, St> Handler<M, Cx, St> for Typed<M, T, C, H>
67where
68    M: IncomingMessage,
69    T: DeserializeOwned + Send + Sync,
70    C: Codec,
71    Cx: Send,
72    St: Send + Sync,
73    H: Handler<T, Cx, St>,
74{
75    async fn handle(&self, msg: &M, ctx: &mut Context<'_, Cx, St>) -> Settle {
76        match self.codec.decode::<T>(msg.payload()) {
77            Ok(value) => self.inner.handle(&value, ctx).await,
78            Err(err) => {
79                warn!(
80                    target: "ruststream::dispatch",
81                    subscription = %ctx.name(),
82                    message_type = std::any::type_name::<T>(),
83                    error = %err,
84                    "codec decode failed",
85                );
86                #[cfg(feature = "testing")]
87                ctx.mark_decode_failed();
88                match self.decode {
89                    FailurePolicy::FailFast => {
90                        ctx.fail_fast(&format!("decode failed: {err}"));
91                        HandlerResult::drop()
92                    }
93                    other => other.settlement().unwrap_or_else(HandlerResult::drop),
94                }
95                .into()
96            }
97        }
98    }
99}
100
101#[cfg(all(test, feature = "json"))]
102mod tests {
103    use std::sync::{
104        Arc,
105        atomic::{AtomicU32, Ordering},
106    };
107
108    use super::typed;
109    use crate::codec::JsonCodec;
110    use crate::runtime::context::Context;
111    use crate::runtime::dispatch::Delivery;
112    use crate::runtime::failure::FailurePolicy;
113    use crate::runtime::handler::{Handler, HandlerResult};
114    use crate::{AckError, Headers, IncomingMessage};
115
116    struct StubMsg(Vec<u8>, Headers);
117
118    impl IncomingMessage for StubMsg {
119        fn payload(&self) -> &[u8] {
120            &self.0
121        }
122
123        fn headers(&self) -> &Headers {
124            &self.1
125        }
126
127        async fn ack(self) -> Result<(), AckError> {
128            Ok(())
129        }
130
131        async fn nack(self, _requeue: bool) -> Result<(), AckError> {
132            Ok(())
133        }
134    }
135
136    fn counting_inner(seen: &Arc<AtomicU32>) -> impl Handler<u32> {
137        let seen = Arc::clone(seen);
138        move |value: &u32, _ctx: &mut Context| {
139            let seen = Arc::clone(&seen);
140            let value = *value;
141            async move {
142                seen.store(value, Ordering::SeqCst);
143                HandlerResult::Ack
144            }
145        }
146    }
147
148    // Plain #[tokio::test]: nothing is spawned, the handler future is awaited inline.
149    #[tokio::test]
150    async fn decoded_value_reaches_inner() {
151        let seen = Arc::new(AtomicU32::new(0));
152        let handler = typed(JsonCodec, counting_inner(&seen));
153        let state = ();
154        let delivery = Delivery::empty();
155        let headers = Headers::new();
156        let mut ctx = Context::new("typed", &headers, &state, (), &delivery);
157
158        let msg = StubMsg(b"7".to_vec(), Headers::new());
159        assert_eq!(
160            handler.handle(&msg, &mut ctx).await.outcome(),
161            HandlerResult::Ack
162        );
163        assert_eq!(seen.load(Ordering::SeqCst), 7);
164    }
165
166    #[tokio::test]
167    async fn decode_failure_drops_by_default() {
168        let seen = Arc::new(AtomicU32::new(0));
169        let handler = typed(JsonCodec, counting_inner(&seen));
170        let state = ();
171        let delivery = Delivery::empty();
172        let headers = Headers::new();
173        let mut ctx = Context::new("typed", &headers, &state, (), &delivery);
174
175        let msg = StubMsg(b"not json".to_vec(), Headers::new());
176        assert_eq!(
177            handler.handle(&msg, &mut ctx).await.outcome(),
178            HandlerResult::drop()
179        );
180        assert_eq!(seen.load(Ordering::SeqCst), 0, "inner must not run");
181    }
182
183    #[tokio::test]
184    async fn decode_failure_requeues_when_overridden() {
185        let seen = Arc::new(AtomicU32::new(0));
186        let handler =
187            typed(JsonCodec, counting_inner(&seen)).on_decode_failure(FailurePolicy::Retry);
188        let state = ();
189        let delivery = Delivery::empty();
190        let headers = Headers::new();
191        let mut ctx = Context::new("typed", &headers, &state, (), &delivery);
192
193        let msg = StubMsg(b"not json".to_vec(), Headers::new());
194        assert_eq!(
195            handler.handle(&msg, &mut ctx).await.outcome(),
196            HandlerResult::retry()
197        );
198        assert_eq!(seen.load(Ordering::SeqCst), 0, "inner must not run");
199    }
200
201    #[tokio::test]
202    async fn typed_handler_is_debug_and_stub_acks() {
203        let seen = Arc::new(AtomicU32::new(0));
204        let handler = typed(JsonCodec, counting_inner(&seen));
205        let state = ();
206        let delivery = Delivery::empty();
207        let headers = Headers::new();
208        let mut ctx = Context::new("typed", &headers, &state, (), &delivery);
209        // Drive one delivery to pin the message type, then check the Debug rendering.
210        let msg = StubMsg(b"5".to_vec(), Headers::new());
211        let _ = handler.handle(&msg, &mut ctx).await;
212        assert!(format!("{handler:?}").contains("Typed"));
213
214        // Exercise the StubMsg fixture's own IncomingMessage surface.
215        let other = StubMsg(b"x".to_vec(), Headers::new());
216        assert!(other.headers().is_empty());
217        other.ack().await.unwrap();
218        StubMsg(Vec::new(), Headers::new())
219            .nack(true)
220            .await
221            .unwrap();
222    }
223
224    // Captures the fields of the one event emitted on a decode failure, so the test can assert the
225    // diagnostic carries the subscription name and target type (needs a tracing subscriber, hence
226    // the `logging` feature gate).
227    #[cfg(feature = "logging")]
228    #[tokio::test]
229    async fn decode_failure_log_names_subscription_and_type() {
230        use std::collections::HashMap;
231        use std::sync::Mutex;
232
233        use tracing::field::{Field, Visit};
234        use tracing_subscriber::Layer;
235        use tracing_subscriber::layer::{Context as LayerContext, SubscriberExt as _};
236
237        #[derive(Default)]
238        struct FieldGrab(HashMap<String, String>);
239
240        impl Visit for FieldGrab {
241            fn record_str(&mut self, field: &Field, value: &str) {
242                self.0.insert(field.name().to_owned(), value.to_owned());
243            }
244
245            fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
246                self.0
247                    .entry(field.name().to_owned())
248                    .or_insert_with(|| format!("{value:?}"));
249            }
250        }
251
252        struct Capture(Arc<Mutex<Vec<HashMap<String, String>>>>);
253
254        impl<S: tracing::Subscriber> Layer<S> for Capture {
255            fn on_event(&self, event: &tracing::Event<'_>, _ctx: LayerContext<'_, S>) {
256                let mut grab = FieldGrab::default();
257                event.record(&mut grab);
258                self.0.lock().unwrap().push(grab.0);
259            }
260        }
261
262        let events = Arc::new(Mutex::new(Vec::new()));
263        let guard = tracing::subscriber::set_default(
264            tracing_subscriber::registry().with(Capture(Arc::clone(&events))),
265        );
266
267        let seen = Arc::new(AtomicU32::new(0));
268        let handler = typed(JsonCodec, counting_inner(&seen));
269        let state = ();
270        let delivery = Delivery::empty();
271        let headers = Headers::new();
272        let mut ctx = Context::new("orders.inbound", &headers, &state, (), &delivery);
273        let msg = StubMsg(b"not json".to_vec(), Headers::new());
274        assert_eq!(
275            handler.handle(&msg, &mut ctx).await.outcome(),
276            HandlerResult::drop()
277        );
278        drop(guard);
279
280        let decode_event = {
281            let captured = events.lock().unwrap();
282            captured
283                .iter()
284                .find(|f| f.get("message").is_some_and(|m| m == "codec decode failed"))
285                .cloned()
286                .expect("a codec-decode-failed event must be emitted")
287        };
288        assert_eq!(
289            decode_event.get("subscription").map(String::as_str),
290            Some("orders.inbound")
291        );
292        assert_eq!(
293            decode_event.get("message_type").map(String::as_str),
294            Some("u32")
295        );
296    }
297}
ruststream/runtime/typed.rs

ruststream/runtime/
typed.rs
