nautilus_common/msgbus/
typed_endpoints.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//! Type-safe endpoint mapping for point-to-point messaging.
17//!
18//! This module provides [`EndpointMap<T>`] for registering handlers at named
19//! endpoints and sending typed messages directly to them.
20
21use std::fmt::Debug;
22
23use indexmap::IndexMap;
24
25use super::{
26    mstr::{Endpoint, MStr},
27    typed_handler::{TypedHandler, TypedIntoHandler},
28};
29
30/// Maps endpoints to typed handlers for point-to-point messaging.
31///
32/// Provides O(1) lookup for registered endpoints and type-safe message dispatch.
33#[derive(Debug)]
34pub struct EndpointMap<T: 'static> {
35    handlers: IndexMap<MStr<Endpoint>, TypedHandler<T>>,
36}
37
38impl<T: 'static> Default for EndpointMap<T> {
39    fn default() -> Self {
40        Self::new()
41    }
42}
43
44impl<T: 'static> EndpointMap<T> {
45    /// Creates a new empty endpoint map.
46    #[must_use]
47    pub fn new() -> Self {
48        Self {
49            handlers: IndexMap::new(),
50        }
51    }
52
53    /// Returns the number of registered endpoints.
54    #[must_use]
55    pub fn len(&self) -> usize {
56        self.handlers.len()
57    }
58
59    /// Returns whether there are any registered endpoints.
60    #[must_use]
61    pub fn is_empty(&self) -> bool {
62        self.handlers.is_empty()
63    }
64
65    /// Returns all registered endpoint addresses.
66    #[must_use]
67    pub fn endpoints(&self) -> Vec<&str> {
68        self.handlers.keys().map(|e| e.as_str()).collect()
69    }
70
71    /// Returns whether an endpoint is registered.
72    #[must_use]
73    pub fn is_registered(&self, endpoint: MStr<Endpoint>) -> bool {
74        self.handlers.contains_key(&endpoint)
75    }
76
77    /// Registers a handler at an endpoint.
78    ///
79    /// If the endpoint already has a handler, it will be replaced.
80    pub fn register(&mut self, endpoint: MStr<Endpoint>, handler: TypedHandler<T>) {
81        log::debug!(
82            "Registering endpoint '{endpoint}' with handler ID {}",
83            handler.id()
84        );
85        self.handlers.insert(endpoint, handler);
86    }
87
88    /// Deregisters the handler at an endpoint.
89    pub fn deregister(&mut self, endpoint: MStr<Endpoint>) {
90        log::debug!("Deregistering endpoint '{endpoint}'");
91        self.handlers.shift_remove(&endpoint);
92    }
93
94    /// Gets the handler registered at an endpoint.
95    #[must_use]
96    pub fn get(&self, endpoint: MStr<Endpoint>) -> Option<&TypedHandler<T>> {
97        self.handlers.get(&endpoint)
98    }
99
100    /// Sends a message to an endpoint.
101    ///
102    /// Logs an error if no handler is registered for the endpoint.
103    pub fn send(&self, endpoint: MStr<Endpoint>, message: &T) {
104        if let Some(handler) = self.handlers.get(&endpoint) {
105            handler.handle(message);
106        } else {
107            log::error!("send: no registered endpoint '{endpoint}'");
108        }
109    }
110
111    /// Sends a message to an endpoint, returning whether a handler was found.
112    #[must_use]
113    pub fn try_send(&self, endpoint: MStr<Endpoint>, message: &T) -> bool {
114        if let Some(handler) = self.handlers.get(&endpoint) {
115            handler.handle(message);
116            true
117        } else {
118            false
119        }
120    }
121
122    /// Clears all registered endpoints.
123    pub fn clear(&mut self) {
124        self.handlers.clear();
125    }
126}
127
128/// Maps endpoints to ownership-based typed handlers for point-to-point messaging.
129///
130/// Unlike [`EndpointMap`] which borrows messages, this map transfers ownership
131/// of messages to handlers, enabling zero-copy processing.
132#[derive(Debug)]
133pub struct IntoEndpointMap<T: 'static> {
134    handlers: IndexMap<MStr<Endpoint>, TypedIntoHandler<T>>,
135}
136
137impl<T: 'static> Default for IntoEndpointMap<T> {
138    fn default() -> Self {
139        Self::new()
140    }
141}
142
143impl<T: 'static> IntoEndpointMap<T> {
144    /// Creates a new empty endpoint map.
145    #[must_use]
146    pub fn new() -> Self {
147        Self {
148            handlers: IndexMap::new(),
149        }
150    }
151
152    /// Returns the number of registered endpoints.
153    #[must_use]
154    pub fn len(&self) -> usize {
155        self.handlers.len()
156    }
157
158    /// Returns whether there are any registered endpoints.
159    #[must_use]
160    pub fn is_empty(&self) -> bool {
161        self.handlers.is_empty()
162    }
163
164    /// Returns all registered endpoint addresses.
165    #[must_use]
166    pub fn endpoints(&self) -> Vec<&str> {
167        self.handlers.keys().map(|e| e.as_str()).collect()
168    }
169
170    /// Returns whether an endpoint is registered.
171    #[must_use]
172    pub fn is_registered(&self, endpoint: MStr<Endpoint>) -> bool {
173        self.handlers.contains_key(&endpoint)
174    }
175
176    /// Registers a handler at an endpoint.
177    ///
178    /// If the endpoint already has a handler, it will be replaced.
179    pub fn register(&mut self, endpoint: MStr<Endpoint>, handler: TypedIntoHandler<T>) {
180        log::debug!(
181            "Registering endpoint '{endpoint}' with handler ID {}",
182            handler.id()
183        );
184        self.handlers.insert(endpoint, handler);
185    }
186
187    /// Deregisters the handler at an endpoint.
188    pub fn deregister(&mut self, endpoint: MStr<Endpoint>) {
189        log::debug!("Deregistering endpoint '{endpoint}'");
190        self.handlers.shift_remove(&endpoint);
191    }
192
193    /// Gets the handler registered at an endpoint.
194    #[must_use]
195    pub fn get(&self, endpoint: MStr<Endpoint>) -> Option<&TypedIntoHandler<T>> {
196        self.handlers.get(&endpoint)
197    }
198
199    /// Sends a message to an endpoint, transferring ownership.
200    ///
201    /// Logs an error if no handler is registered for the endpoint.
202    pub fn send(&self, endpoint: MStr<Endpoint>, message: T) {
203        if let Some(handler) = self.handlers.get(&endpoint) {
204            handler.handle(message);
205        } else {
206            log::error!("send: no registered endpoint '{endpoint}'");
207        }
208    }
209
210    /// Sends a message to an endpoint, returning whether a handler was found.
211    ///
212    /// # Errors
213    ///
214    /// Returns an error if no handler is registered, allowing the caller to recover the message.
215    /// The error contains `message`.
216    pub fn try_send(&self, endpoint: MStr<Endpoint>, message: T) -> Result<(), T> {
217        if let Some(handler) = self.handlers.get(&endpoint) {
218            handler.handle(message);
219            Ok(())
220        } else {
221            Err(message)
222        }
223    }
224
225    /// Clears all registered endpoints.
226    pub fn clear(&mut self) {
227        self.handlers.clear();
228    }
229}
230
231#[cfg(test)]
232mod tests {
233    use std::{cell::RefCell, rc::Rc};
234
235    use rstest::rstest;
236    use ustr::Ustr;
237
238    use super::*;
239
240    #[rstest]
241    fn test_endpoint_map_register_and_send() {
242        let mut endpoints = EndpointMap::<String>::new();
243        let received = Rc::new(RefCell::new(Vec::new()));
244        let received_clone = received.clone();
245
246        let handler = TypedHandler::from(move |msg: &String| {
247            received_clone.borrow_mut().push(msg.clone());
248        });
249
250        let endpoint: MStr<Endpoint> = "DataEngine.execute".into();
251        endpoints.register(endpoint, handler);
252
253        endpoints.send(endpoint, &"command1".to_string());
254        endpoints.send(endpoint, &"command2".to_string());
255
256        assert_eq!(*received.borrow(), vec!["command1", "command2"]);
257    }
258
259    #[rstest]
260    fn test_endpoint_map_deregister() {
261        let mut endpoints = EndpointMap::<i32>::new();
262        let received = Rc::new(RefCell::new(Vec::new()));
263        let received_clone = received.clone();
264
265        let handler = TypedHandler::from(move |msg: &i32| {
266            received_clone.borrow_mut().push(*msg);
267        });
268
269        let endpoint: MStr<Endpoint> = "Test.endpoint".into();
270        endpoints.register(endpoint, handler);
271        assert!(endpoints.is_registered(endpoint));
272
273        endpoints.deregister(endpoint);
274        assert!(!endpoints.is_registered(endpoint));
275
276        // Should not receive (no handler)
277        endpoints.send(endpoint, &42);
278        assert!(received.borrow().is_empty());
279    }
280
281    #[rstest]
282    fn test_endpoint_map_try_send() {
283        let mut endpoints = EndpointMap::<String>::new();
284        let received = Rc::new(RefCell::new(Vec::new()));
285        let received_clone = received.clone();
286
287        let handler = TypedHandler::from(move |msg: &String| {
288            received_clone.borrow_mut().push(msg.clone());
289        });
290
291        let registered: MStr<Endpoint> = "Registered.endpoint".into();
292        let unregistered: MStr<Endpoint> = "Unregistered.endpoint".into();
293
294        endpoints.register(registered, handler);
295
296        assert!(endpoints.try_send(registered, &"test".to_string()));
297        assert!(!endpoints.try_send(unregistered, &"test".to_string()));
298
299        assert_eq!(*received.borrow(), vec!["test"]);
300    }
301
302    #[rstest]
303    fn test_endpoint_map_replace_handler() {
304        let mut endpoints = EndpointMap::<i32>::new();
305        let first_received = Rc::new(RefCell::new(false));
306        let second_received = Rc::new(RefCell::new(false));
307
308        let first_clone = first_received.clone();
309        let handler1 = TypedHandler::from(move |_: &i32| {
310            *first_clone.borrow_mut() = true;
311        });
312
313        let second_clone = second_received.clone();
314        let handler2 = TypedHandler::from(move |_: &i32| {
315            *second_clone.borrow_mut() = true;
316        });
317
318        let endpoint: MStr<Endpoint> = "Test.endpoint".into();
319
320        endpoints.register(endpoint, handler1);
321        endpoints.register(endpoint, handler2);
322
323        endpoints.send(endpoint, &1);
324
325        // Only second handler should receive (replaced first)
326        assert!(!*first_received.borrow());
327        assert!(*second_received.borrow());
328    }
329
330    #[rstest]
331    fn test_endpoint_map_tracks_registered_endpoints() {
332        let mut endpoints = EndpointMap::<i32>::new();
333        let first: MStr<Endpoint> = "First.endpoint".into();
334        let second: MStr<Endpoint> = "Second.endpoint".into();
335
336        assert_eq!(endpoints.len(), 0);
337        assert!(endpoints.is_empty());
338        assert!(endpoints.endpoints().is_empty());
339        assert!(endpoints.get(first).is_none());
340
341        endpoints.register(first, TypedHandler::from_with_id("first", |_: &i32| {}));
342        endpoints.register(second, TypedHandler::from_with_id("second", |_: &i32| {}));
343
344        assert_eq!(endpoints.len(), 2);
345        assert!(!endpoints.is_empty());
346        assert_eq!(
347            endpoints.endpoints(),
348            vec!["First.endpoint", "Second.endpoint"]
349        );
350        assert_eq!(endpoints.get(first).unwrap().id(), Ustr::from("first"));
351        assert_eq!(endpoints.get(second).unwrap().id(), Ustr::from("second"));
352
353        endpoints.deregister(first);
354
355        assert_eq!(endpoints.endpoints(), vec!["Second.endpoint"]);
356
357        endpoints.clear();
358
359        assert_eq!(endpoints.len(), 0);
360        assert!(endpoints.get(second).is_none());
361    }
362
363    #[rstest]
364    fn test_into_endpoint_map_register_and_send_transfers_ownership() {
365        let mut endpoints = IntoEndpointMap::<String>::new();
366        let received = Rc::new(RefCell::new(Vec::new()));
367        let received_clone = received.clone();
368
369        let handler = TypedIntoHandler::from(move |msg: String| {
370            received_clone.borrow_mut().push(msg);
371        });
372
373        let endpoint: MStr<Endpoint> = "ExecEngine.execute".into();
374        endpoints.register(endpoint, handler);
375
376        endpoints.send(endpoint, "command1".to_string());
377        endpoints.send(endpoint, "command2".to_string());
378
379        assert_eq!(*received.borrow(), vec!["command1", "command2"]);
380    }
381
382    #[rstest]
383    fn test_into_endpoint_map_send_to_unregistered_endpoint_drops_the_message() {
384        let mut endpoints = IntoEndpointMap::<i32>::new();
385        let received = Rc::new(RefCell::new(Vec::new()));
386        let received_clone = received.clone();
387
388        let endpoint: MStr<Endpoint> = "Test.endpoint".into();
389        endpoints.register(
390            endpoint,
391            TypedIntoHandler::from(move |msg: i32| received_clone.borrow_mut().push(msg)),
392        );
393        endpoints.deregister(endpoint);
394
395        endpoints.send(endpoint, 42);
396
397        assert!(received.borrow().is_empty());
398        assert!(!endpoints.is_registered(endpoint));
399    }
400
401    #[rstest]
402    fn test_into_endpoint_map_try_send_returns_the_message_when_unregistered() {
403        let mut endpoints = IntoEndpointMap::<String>::new();
404        let received = Rc::new(RefCell::new(Vec::new()));
405        let received_clone = received.clone();
406
407        let registered: MStr<Endpoint> = "Registered.endpoint".into();
408        let unregistered: MStr<Endpoint> = "Unregistered.endpoint".into();
409        endpoints.register(
410            registered,
411            TypedIntoHandler::from(move |msg: String| received_clone.borrow_mut().push(msg)),
412        );
413
414        assert_eq!(
415            endpoints.try_send(registered, "delivered".to_string()),
416            Ok(())
417        );
418        assert_eq!(
419            endpoints.try_send(unregistered, "recovered".to_string()),
420            Err("recovered".to_string()),
421            "an unregistered endpoint must hand the message back intact"
422        );
423
424        assert_eq!(*received.borrow(), vec!["delivered"]);
425    }
426
427    #[rstest]
428    fn test_into_endpoint_map_register_replaces_the_existing_handler() {
429        let mut endpoints = IntoEndpointMap::<i32>::new();
430        let received = Rc::new(RefCell::new(Vec::new()));
431
432        let first_clone = received.clone();
433        let second_clone = received.clone();
434        let endpoint: MStr<Endpoint> = "Test.endpoint".into();
435
436        endpoints.register(
437            endpoint,
438            TypedIntoHandler::from_with_id("first", move |_: i32| {
439                first_clone.borrow_mut().push("first");
440            }),
441        );
442
443        endpoints.register(
444            endpoint,
445            TypedIntoHandler::from_with_id("second", move |_: i32| {
446                second_clone.borrow_mut().push("second");
447            }),
448        );
449
450        endpoints.send(endpoint, 1);
451
452        assert_eq!(endpoints.len(), 1);
453        assert_eq!(endpoints.get(endpoint).unwrap().id(), Ustr::from("second"));
454        assert_eq!(*received.borrow(), vec!["second"]);
455    }
456
457    #[rstest]
458    fn test_into_endpoint_map_tracks_registered_endpoints() {
459        let mut endpoints = IntoEndpointMap::<i32>::new();
460        let first: MStr<Endpoint> = "First.endpoint".into();
461        let second: MStr<Endpoint> = "Second.endpoint".into();
462
463        assert_eq!(endpoints.len(), 0);
464        assert!(endpoints.is_empty());
465        assert!(endpoints.endpoints().is_empty());
466        assert!(endpoints.get(first).is_none());
467
468        endpoints.register(first, TypedIntoHandler::from_with_id("first", |_: i32| {}));
469        endpoints.register(
470            second,
471            TypedIntoHandler::from_with_id("second", |_: i32| {}),
472        );
473
474        assert_eq!(endpoints.len(), 2);
475        assert!(!endpoints.is_empty());
476        assert_eq!(
477            endpoints.endpoints(),
478            vec!["First.endpoint", "Second.endpoint"]
479        );
480        assert!(endpoints.is_registered(first));
481
482        endpoints.deregister(first);
483
484        assert!(!endpoints.is_registered(first));
485        assert_eq!(endpoints.endpoints(), vec!["Second.endpoint"]);
486
487        endpoints.clear();
488
489        assert!(endpoints.is_empty());
490        assert!(endpoints.get(second).is_none());
491    }
492
493    #[rstest]
494    fn test_endpoint_maps_default_to_empty() {
495        assert!(EndpointMap::<i32>::default().is_empty());
496        assert!(IntoEndpointMap::<i32>::default().is_empty());
497    }
498}
nautilus_common/msgbus/typed_endpoints.rs

nautilus_common/msgbus/
typed_endpoints.rs
