nautilus_common/msgbus/
typed_handler.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Compile-time type-safe message handler infrastructure.
17//!
18//! This module provides generic handler traits and types that enable type-safe
19//! message dispatch without runtime downcasting for built-in message types.
20
21use std::{any::Any, fmt::Debug, marker::PhantomData, rc::Rc};
22
23use nautilus_core::UUID4;
24use ustr::Ustr;
25
26/// Compile-time type-safe message handler trait.
27///
28/// Provides zero-cost dispatch for statically typed messages. Can also be used
29/// with `dyn Any` for dynamic dispatch when type flexibility is needed.
30pub trait Handler<T: ?Sized>: 'static {
31    /// Returns the unique identifier for this handler.
32    fn id(&self) -> Ustr;
33
34    /// Handles a message of type `T`.
35    fn handle(&self, message: &T);
36}
37
38impl<T: ?Sized, H: Handler<T>> Handler<T> for Rc<H> {
39    fn id(&self) -> Ustr {
40        (**self).id()
41    }
42
43    fn handle(&self, message: &T) {
44        (**self).handle(message);
45    }
46}
47
48/// A shareable wrapper for typed handlers.
49///
50/// Provides reference-counted access to handlers. Supports both concrete types
51/// for zero-cost dispatch and `dyn Any` for dynamic dispatch.
52///
53/// # Thread Safety
54///
55/// Uses `Rc` intentionally (not `Arc`) for single-threaded use within each
56/// async runtime. The `MessageBus` uses thread-local storage to ensure each
57/// thread gets its own handlers.
58pub struct TypedHandler<T: 'static + ?Sized>(pub Rc<dyn Handler<T>>);
59
60impl<T: 'static + ?Sized> Clone for TypedHandler<T> {
61    fn clone(&self) -> Self {
62        Self(Rc::clone(&self.0))
63    }
64}
65
66impl<T: 'static + ?Sized> TypedHandler<T> {
67    /// Returns the handler ID.
68    pub fn id(&self) -> Ustr {
69        self.0.id()
70    }
71
72    /// Handles a message by delegating to the inner handler.
73    pub fn handle(&self, message: &T) {
74        self.0.handle(message);
75    }
76}
77
78impl<T: 'static> TypedHandler<T> {
79    /// Creates a new typed handler from any type implementing `Handler<T>`.
80    pub fn new<H: Handler<T>>(handler: H) -> Self {
81        Self(Rc::new(handler))
82    }
83
84    /// Creates a new typed handler from a callback function.
85    pub fn from<F>(callback: F) -> Self
86    where
87        F: Fn(&T) + 'static,
88    {
89        Self::new(CallbackHandler::new(None::<&str>, callback))
90    }
91
92    /// Creates a new typed handler from a callback function with a custom ID.
93    pub fn from_with_id<S: AsRef<str>, F>(id: S, callback: F) -> Self
94    where
95        F: Fn(&T) + 'static,
96    {
97        Self::new(CallbackHandler::new(Some(id), callback))
98    }
99}
100
101impl<T: 'static + ?Sized> Debug for TypedHandler<T> {
102    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
103        f.debug_struct(stringify!(TypedHandler))
104            .field("id", &self.0.id())
105            .field("type", &std::any::type_name::<T>())
106            .finish()
107    }
108}
109
110impl<T: 'static + ?Sized> PartialEq for TypedHandler<T> {
111    fn eq(&self, other: &Self) -> bool {
112        self.0.id() == other.0.id()
113    }
114}
115
116impl<T: 'static + ?Sized> Eq for TypedHandler<T> {}
117
118impl<T: 'static + ?Sized> std::hash::Hash for TypedHandler<T> {
119    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
120        self.0.id().hash(state);
121    }
122}
123
124impl From<Rc<dyn Handler<dyn Any>>> for TypedHandler<dyn Any> {
125    fn from(handler: Rc<dyn Handler<dyn Any>>) -> Self {
126        Self(handler)
127    }
128}
129
130/// Type alias for handlers that work with dynamically-typed messages.
131///
132/// This replaces the legacy `MessageHandler` trait with a unified handler
133/// type that supports both typed and dynamic dispatch through the same trait.
134pub type ShareableMessageHandler = TypedHandler<dyn Any>;
135
136/// Creates a new `ShareableMessageHandler` from an `Rc<dyn Handler<dyn Any>>`.
137#[must_use]
138pub fn shareable_handler(handler: Rc<dyn Handler<dyn Any>>) -> ShareableMessageHandler {
139    TypedHandler(handler)
140}
141
142impl ShareableMessageHandler {
143    /// Creates a handler from a typed closure that internally downcasts.
144    ///
145    /// Use this when you need Any-based routing but want type-safe handling.
146    /// The callback will only be invoked if the message downcasts successfully.
147    pub fn from_typed<T, F>(f: F) -> Self
148    where
149        T: 'static,
150        F: Fn(&T) + 'static,
151    {
152        TypedHandler(Rc::new(DowncastingHandler::new(f)))
153    }
154
155    /// Creates a handler from an Any-typed closure.
156    pub fn from_any<F>(f: F) -> Self
157    where
158        F: Fn(&dyn Any) + 'static,
159    {
160        TypedHandler(Rc::new(AnyCallbackHandler::new(f)))
161    }
162}
163
164/// Handler that downcasts `&dyn Any` to a concrete type before calling the callback.
165struct DowncastingHandler<T, F: Fn(&T)> {
166    id: Ustr,
167    callback: F,
168    _marker: PhantomData<T>,
169}
170
171impl<T: 'static, F: Fn(&T) + 'static> DowncastingHandler<T, F> {
172    fn new(callback: F) -> Self {
173        Self {
174            id: generate_handler_id(&callback),
175            callback,
176            _marker: PhantomData,
177        }
178    }
179}
180
181impl<T: 'static, F: Fn(&T) + 'static> Handler<dyn Any> for DowncastingHandler<T, F> {
182    fn id(&self) -> Ustr {
183        self.id
184    }
185
186    fn handle(&self, message: &dyn Any) {
187        if let Some(typed_msg) = message.downcast_ref::<T>() {
188            (self.callback)(typed_msg);
189        } else {
190            log::error!(
191                "DowncastingHandler downcast failed: expected {} got {:?}",
192                std::any::type_name::<T>(),
193                message.type_id()
194            );
195        }
196    }
197}
198
199/// Handler that directly receives `&dyn Any` without downcasting.
200struct AnyCallbackHandler<F: Fn(&dyn Any)> {
201    id: Ustr,
202    callback: F,
203}
204
205impl<F: Fn(&dyn Any) + 'static> AnyCallbackHandler<F> {
206    fn new(callback: F) -> Self {
207        Self {
208            id: generate_handler_id(&callback),
209            callback,
210        }
211    }
212}
213
214impl<F: Fn(&dyn Any) + 'static> Handler<dyn Any> for AnyCallbackHandler<F> {
215    fn id(&self) -> Ustr {
216        self.id
217    }
218
219    fn handle(&self, message: &dyn Any) {
220        (self.callback)(message);
221    }
222}
223
224/// A callback-based handler implementation.
225///
226/// This is the typed equivalent of `TypedMessageHandler`,
227/// but without runtime downcasting overhead.
228pub struct CallbackHandler<T, F: Fn(&T)> {
229    id: Ustr,
230    callback: F,
231    _marker: PhantomData<T>,
232}
233
234impl<T: 'static, F: Fn(&T) + 'static> CallbackHandler<T, F> {
235    /// Creates a new callback handler with an optional custom ID.
236    pub fn new<S: AsRef<str>>(id: Option<S>, callback: F) -> Self {
237        let id_ustr = id.map_or_else(
238            || generate_handler_id(&callback),
239            |s| Ustr::from(s.as_ref()),
240        );
241
242        Self {
243            id: id_ustr,
244            callback,
245            _marker: PhantomData,
246        }
247    }
248}
249
250impl<T: 'static, F: Fn(&T) + 'static> Handler<T> for CallbackHandler<T, F> {
251    fn id(&self) -> Ustr {
252        self.id
253    }
254
255    fn handle(&self, message: &T) {
256        (self.callback)(message);
257    }
258}
259
260impl<T, F: Fn(&T)> Debug for CallbackHandler<T, F> {
261    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
262        f.debug_struct(stringify!(CallbackHandler))
263            .field("id", &self.id)
264            .field("type", &std::any::type_name::<T>())
265            .finish()
266    }
267}
268
269fn generate_handler_id<F: ?Sized>(callback: &F) -> Ustr {
270    let callback_ptr = std::ptr::from_ref(callback);
271    let uuid = UUID4::new();
272    Ustr::from(&format!("<{callback_ptr:?}>-{uuid}"))
273}
274
275/// Compile-time type-safe message handler trait that takes ownership.
276///
277/// Unlike [`Handler<T>`] which borrows messages, this trait takes ownership
278/// of messages, enabling zero-copy processing when the handler needs to store
279/// or forward the message.
280pub trait IntoHandler<T>: 'static {
281    /// Returns the unique identifier for this handler.
282    fn id(&self) -> Ustr;
283
284    /// Handles a message of type `T`, taking ownership.
285    fn handle(&self, message: T);
286}
287
288impl<T, H: IntoHandler<T>> IntoHandler<T> for Rc<H> {
289    fn id(&self) -> Ustr {
290        (**self).id()
291    }
292
293    fn handle(&self, message: T) {
294        (**self).handle(message);
295    }
296}
297
298/// A shareable wrapper for ownership-based typed handlers.
299///
300/// This is the ownership-based equivalent of [`TypedHandler`], used for
301/// point-to-point messaging where the sender transfers ownership of the message.
302///
303/// # Thread Safety
304///
305/// Uses `Rc` intentionally (not `Arc`) for single-threaded use within each
306/// async runtime. The `MessageBus` uses thread-local storage to ensure each
307/// thread gets its own handlers.
308pub struct TypedIntoHandler<T: 'static>(pub Rc<dyn IntoHandler<T>>);
309
310impl<T: 'static> Clone for TypedIntoHandler<T> {
311    fn clone(&self) -> Self {
312        Self(Rc::clone(&self.0))
313    }
314}
315
316impl<T: 'static> TypedIntoHandler<T> {
317    /// Creates a new typed into handler from any type implementing `IntoHandler<T>`.
318    pub fn new<H: IntoHandler<T>>(handler: H) -> Self {
319        Self(Rc::new(handler))
320    }
321
322    /// Creates a new typed into handler from a callback function.
323    pub fn from<F>(callback: F) -> Self
324    where
325        F: Fn(T) + 'static,
326    {
327        Self::new(IntoCallbackHandler::new(None::<&str>, callback))
328    }
329
330    /// Creates a new typed into handler from a callback function with a custom ID.
331    pub fn from_with_id<S: AsRef<str>, F>(id: S, callback: F) -> Self
332    where
333        F: Fn(T) + 'static,
334    {
335        Self::new(IntoCallbackHandler::new(Some(id), callback))
336    }
337
338    /// Returns the handler ID.
339    pub fn id(&self) -> Ustr {
340        self.0.id()
341    }
342
343    /// Handles a message by delegating to the inner handler, taking ownership.
344    pub fn handle(&self, message: T) {
345        self.0.handle(message);
346    }
347}
348
349impl<T: 'static> Debug for TypedIntoHandler<T> {
350    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
351        f.debug_struct(stringify!(TypedIntoHandler))
352            .field("id", &self.0.id())
353            .field("type", &std::any::type_name::<T>())
354            .finish()
355    }
356}
357
358impl<T: 'static> PartialEq for TypedIntoHandler<T> {
359    fn eq(&self, other: &Self) -> bool {
360        self.0.id() == other.0.id()
361    }
362}
363
364impl<T: 'static> Eq for TypedIntoHandler<T> {}
365
366impl<T: 'static> std::hash::Hash for TypedIntoHandler<T> {
367    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
368        self.0.id().hash(state);
369    }
370}
371
372/// A callback-based handler implementation that takes ownership.
373///
374/// This is the ownership-based equivalent of `CallbackHandler`.
375pub struct IntoCallbackHandler<T, F: Fn(T)> {
376    id: Ustr,
377    callback: F,
378    _marker: PhantomData<T>,
379}
380
381impl<T: 'static, F: Fn(T) + 'static> IntoCallbackHandler<T, F> {
382    /// Creates a new into callback handler with an optional custom ID.
383    pub fn new<S: AsRef<str>>(id: Option<S>, callback: F) -> Self {
384        let id_ustr = id.map_or_else(
385            || generate_handler_id(&callback),
386            |s| Ustr::from(s.as_ref()),
387        );
388
389        Self {
390            id: id_ustr,
391            callback,
392            _marker: PhantomData,
393        }
394    }
395}
396
397impl<T: 'static, F: Fn(T) + 'static> IntoHandler<T> for IntoCallbackHandler<T, F> {
398    fn id(&self) -> Ustr {
399        self.id
400    }
401
402    fn handle(&self, message: T) {
403        (self.callback)(message);
404    }
405}
406
407impl<T, F: Fn(T)> Debug for IntoCallbackHandler<T, F> {
408    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
409        f.debug_struct(stringify!(IntoCallbackHandler))
410            .field("id", &self.id)
411            .field("type", &std::any::type_name::<T>())
412            .finish()
413    }
414}
415
416#[cfg(test)]
417mod tests {
418    use std::{cell::RefCell, collections::HashSet};
419
420    use rstest::rstest;
421
422    use super::*;
423
424    #[rstest]
425    fn test_typed_handler_from_fn() {
426        let received = Rc::new(RefCell::new(Vec::new()));
427        let received_clone = received.clone();
428
429        let handler = TypedHandler::from(move |msg: &String| {
430            received_clone.borrow_mut().push(msg.clone());
431        });
432
433        handler.handle(&"test1".to_string());
434        handler.handle(&"test2".to_string());
435
436        assert_eq!(*received.borrow(), vec!["test1", "test2"]);
437    }
438
439    #[rstest]
440    fn test_generated_handler_ids_are_unique() {
441        let ids = (0..2)
442            .flat_map(|index| {
443                [
444                    TypedHandler::<u32>::from(move |_: &u32| {
445                        std::hint::black_box(index);
446                    })
447                    .id(),
448                    ShareableMessageHandler::from_typed::<u32, _>(move |_: &u32| {
449                        std::hint::black_box(index);
450                    })
451                    .id(),
452                    ShareableMessageHandler::from_any(move |_: &dyn Any| {
453                        std::hint::black_box(index);
454                    })
455                    .id(),
456                    TypedIntoHandler::<u32>::from(move |_: u32| {
457                        std::hint::black_box(index);
458                    })
459                    .id(),
460                ]
461            })
462            .collect::<Vec<_>>();
463
464        assert_eq!(ids.iter().copied().collect::<HashSet<_>>().len(), ids.len());
465    }
466
467    #[rstest]
468    fn test_typed_handler_with_custom_id() {
469        let handler = TypedHandler::from_with_id("custom-id", |_msg: &i32| {});
470
471        assert_eq!(handler.id(), "custom-id");
472    }
473
474    #[rstest]
475    fn test_typed_handler_equality() {
476        let handler1 = TypedHandler::from_with_id("same-id", |_: &u32| {});
477        let handler2 = TypedHandler::from_with_id("same-id", |_: &u32| {});
478        let handler3 = TypedHandler::from_with_id("different-id", |_: &u32| {});
479
480        assert_eq!(handler1, handler2);
481        assert_ne!(handler1, handler3);
482    }
483
484    #[rstest]
485    fn test_typed_handler_hash() {
486        use std::collections::HashSet;
487
488        let handler1 = TypedHandler::from_with_id("id-a", |_: &u32| {});
489        let handler2 = TypedHandler::from_with_id("id-a", |_: &u32| {});
490        let handler3 = TypedHandler::from_with_id("id-b", |_: &u32| {});
491
492        let mut set = HashSet::new();
493        set.insert(handler1);
494
495        // Same ID should be considered duplicate
496        assert!(!set.insert(handler2));
497        // Different ID should be added
498        assert!(set.insert(handler3));
499        assert_eq!(set.len(), 2);
500    }
501
502    #[rstest]
503    fn test_typed_handler_debug() {
504        let handler = TypedHandler::from_with_id("debug-test", |_: &String| {});
505        let debug_str = format!("{handler:?}");
506
507        assert!(debug_str.contains("TypedHandler"));
508        assert!(debug_str.contains("debug-test"));
509    }
510
511    // Tests for Handler<T> impl for Rc<H>
512    struct TestHandler {
513        id: Ustr,
514        call_count: RefCell<usize>,
515    }
516
517    impl TestHandler {
518        fn new(id: &str) -> Self {
519            Self {
520                id: Ustr::from(id),
521                call_count: RefCell::new(0),
522            }
523        }
524    }
525
526    impl Handler<i32> for TestHandler {
527        fn id(&self) -> Ustr {
528            self.id
529        }
530
531        fn handle(&self, _message: &i32) {
532            *self.call_count.borrow_mut() += 1;
533        }
534    }
535
536    #[rstest]
537    fn test_rc_handler_delegates_id() {
538        let handler = TestHandler::new("rc-test-id");
539        let rc_handler: Rc<TestHandler> = Rc::new(handler);
540
541        assert_eq!(rc_handler.id(), Ustr::from("rc-test-id"));
542    }
543
544    #[rstest]
545    fn test_rc_handler_delegates_handle() {
546        let handler = TestHandler::new("rc-handle-test");
547        let rc_handler: Rc<TestHandler> = Rc::new(handler);
548
549        rc_handler.handle(&42);
550        rc_handler.handle(&100);
551
552        assert_eq!(*rc_handler.call_count.borrow(), 2);
553    }
554
555    #[rstest]
556    fn test_rc_handler_shared_state() {
557        let handler = TestHandler::new("shared-state");
558        let rc1: Rc<TestHandler> = Rc::new(handler);
559        let rc2 = rc1.clone();
560
561        // Both Rc's point to same handler
562        rc1.handle(&1);
563        rc2.handle(&2);
564        rc1.handle(&3);
565
566        // All calls should be counted on the same handler
567        assert_eq!(*rc1.call_count.borrow(), 3);
568        assert_eq!(*rc2.call_count.borrow(), 3);
569    }
570
571    #[rstest]
572    fn test_typed_handler_from_rc() {
573        let handler = Rc::new(TestHandler::new("from-rc-test"));
574        let typed: TypedHandler<i32> = TypedHandler::new(handler.clone());
575
576        typed.handle(&42);
577
578        assert_eq!(*handler.call_count.borrow(), 1);
579        assert_eq!(typed.id(), Ustr::from("from-rc-test"));
580    }
581}
nautilus_common/msgbus/typed_handler.rs

nautilus_common/msgbus/
typed_handler.rs
