nautilus_common/msgbus/
backing.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//! External message bus backing traits.
17//!
18//! Ingress and egress are named from the local message bus boundary. Egress carries serialized
19//! [`BusMessage`] values from the local bus to external streams. Ingress exposes serialized
20//! [`BusMessage`] values read from external streams so a live bridge can republish them on the
21//! local bus.
22//!
23//! - `MessageBusBackingFactory` creates concrete backing technology for a bus runtime.
24//! - `MessageBusBacking` owns the runtime facade used by core bus wiring.
25//! - `MessageBusExternalEgress` accepts outbound messages from the local bus.
26//! - `MessageBusExternalIngress` exposes the inbound external stream receiver.
27//! - `external_egress_from_backing` adapts a backing into an egress surface.
28//! - `external_io_from_backing` adapts one backing into shared egress and ingress surfaces.
29
30use std::{
31    cell::{Cell, RefCell},
32    fmt::Debug,
33    rc::Rc,
34};
35
36use nautilus_core::UUID4;
37use nautilus_model::identifiers::TraderId;
38
39use super::config::MessageBusConfig;
40use crate::msgbus::BusMessage;
41
42/// Receiver for external message bus ingress publications.
43#[cfg(feature = "live")]
44pub type MessageBusExternalReceiver = tokio::sync::mpsc::Receiver<BusMessage>;
45
46/// Factory for constructing external message bus backings.
47///
48/// Implementations own concrete backing configuration and return the [`MessageBusBacking`] surface
49/// used by the core bus runtime.
50pub trait MessageBusBackingFactory: Debug + Send + Sync {
51    /// Creates a message bus backing for the given bus runtime.
52    ///
53    /// # Errors
54    ///
55    /// Returns an error if backing construction or connection setup fails.
56    fn create(
57        &self,
58        trader_id: TraderId,
59        instance_id: UUID4,
60        config: MessageBusConfig,
61    ) -> anyhow::Result<Box<dyn MessageBusBacking>>;
62}
63
64/// External message bus backing facade.
65///
66/// Implementations own the concrete backing technology and provide the runtime-facing publication
67/// surface used by the core bus. With the `live` feature, the same backing can also hand an
68/// inbound receiver to the live bridge.
69pub trait MessageBusBacking {
70    /// Returns `true` if the backing has been closed.
71    fn is_closed(&self) -> bool;
72
73    /// Queues a serialized bus message for external egress.
74    fn publish(&self, message: BusMessage);
75
76    /// Takes the inbound message receiver for live bridge consumption.
77    ///
78    /// # Errors
79    ///
80    /// Returns an error if the receiver has already been taken or is unavailable.
81    #[cfg(feature = "live")]
82    fn take_receiver(&mut self) -> anyhow::Result<MessageBusExternalReceiver> {
83        anyhow::bail!("external ingress receiver unavailable")
84    }
85
86    /// Closes the backing and releases any owned resources.
87    fn close(&mut self);
88}
89
90/// External egress surface for serialized message bus publications.
91///
92/// The core bus passes each outbound message as a [`BusMessage`] carrying the
93/// `topic`, `payload_type`, and serialized `payload`. Implementations must not block the publishing
94/// thread. If the underlying channel is full, drop the message in the implementation rather than
95/// applying back-pressure to the node.
96pub trait MessageBusExternalEgress {
97    /// Returns `true` if egress has been closed.
98    fn is_closed(&self) -> bool;
99
100    /// Queues a serialized bus message for external egress.
101    fn publish(&self, message: BusMessage);
102
103    /// Closes egress and stops accepting outbound messages.
104    fn close(&mut self);
105}
106
107/// External ingress surface for serialized message bus publications.
108///
109/// The live bridge consumes each inbound [`BusMessage`] as a topic and serialized
110/// payload. The receiver can be taken only once so ingress can hand ownership of the external stream
111/// to the bridge without exposing concrete backing details.
112#[cfg(feature = "live")]
113pub trait MessageBusExternalIngress {
114    /// Returns `true` if ingress has been closed.
115    fn is_closed(&self) -> bool;
116
117    /// Takes the inbound message receiver for live bridge consumption.
118    ///
119    /// # Errors
120    ///
121    /// Returns an error if the receiver has already been taken or is unavailable.
122    fn take_receiver(&mut self) -> anyhow::Result<MessageBusExternalReceiver>;
123
124    /// Closes ingress and stops accepting inbound messages.
125    fn close(&mut self);
126}
127
128type SharedMessageBusBacking = Rc<RefCell<Box<dyn MessageBusBacking>>>;
129type SharedMessageBusCloseState = Rc<Cell<bool>>;
130
131/// Wraps a message bus backing for external egress installation.
132#[must_use]
133pub fn external_egress_from_backing(
134    backing: Box<dyn MessageBusBacking>,
135) -> Box<dyn MessageBusExternalEgress> {
136    Box::new(BackingExternalEgress {
137        backing: Rc::new(RefCell::new(backing)),
138        closed: Rc::new(Cell::new(false)),
139    })
140}
141
142/// Splits a message bus backing into external egress and ingress surfaces.
143#[cfg(feature = "live")]
144#[must_use]
145pub fn external_io_from_backing(
146    backing: Box<dyn MessageBusBacking>,
147) -> (
148    Box<dyn MessageBusExternalEgress>,
149    Box<dyn MessageBusExternalIngress>,
150) {
151    let backing = Rc::new(RefCell::new(backing));
152    let closed = Rc::new(Cell::new(false));
153    (
154        Box::new(BackingExternalEgress {
155            backing: backing.clone(),
156            closed: closed.clone(),
157        }),
158        Box::new(BackingExternalIngress { backing, closed }),
159    )
160}
161
162struct BackingExternalEgress {
163    backing: SharedMessageBusBacking,
164    closed: SharedMessageBusCloseState,
165}
166
167impl MessageBusExternalEgress for BackingExternalEgress {
168    fn is_closed(&self) -> bool {
169        self.backing.borrow().is_closed()
170    }
171
172    fn publish(&self, message: BusMessage) {
173        self.backing.borrow().publish(message);
174    }
175
176    fn close(&mut self) {
177        if !self.closed.replace(true) {
178            self.backing.borrow_mut().close();
179        }
180    }
181}
182
183#[cfg(feature = "live")]
184struct BackingExternalIngress {
185    backing: SharedMessageBusBacking,
186    closed: SharedMessageBusCloseState,
187}
188
189#[cfg(feature = "live")]
190impl MessageBusExternalIngress for BackingExternalIngress {
191    fn is_closed(&self) -> bool {
192        self.backing.borrow().is_closed()
193    }
194
195    fn take_receiver(&mut self) -> anyhow::Result<MessageBusExternalReceiver> {
196        self.backing.borrow_mut().take_receiver()
197    }
198
199    fn close(&mut self) {
200        if !self.closed.replace(true) {
201            self.backing.borrow_mut().close();
202        }
203    }
204}
205
206#[cfg(all(test, feature = "live"))]
207mod tests {
208    use std::{
209        cell::{Cell, RefCell},
210        rc::Rc,
211    };
212
213    use bytes::Bytes;
214    use rstest::*;
215
216    use super::{MessageBusBacking, external_egress_from_backing, external_io_from_backing};
217    use crate::{
218        enums::SerializationEncoding,
219        msgbus::{
220            BusMessage, BusPayloadType,
221            MessageBusExternalIngress as ReexportedMessageBusExternalIngress,
222        },
223    };
224
225    struct CapturingBacking {
226        rx: Option<tokio::sync::mpsc::Receiver<BusMessage>>,
227        closed: bool,
228    }
229
230    impl MessageBusBacking for CapturingBacking {
231        fn is_closed(&self) -> bool {
232            self.closed
233        }
234
235        fn publish(&self, _message: BusMessage) {}
236
237        fn take_receiver(&mut self) -> anyhow::Result<tokio::sync::mpsc::Receiver<BusMessage>> {
238            self.rx
239                .take()
240                .ok_or_else(|| anyhow::anyhow!("Stream receiver already taken"))
241        }
242
243        fn close(&mut self) {
244            self.closed = true;
245        }
246    }
247
248    struct CapturingPublishBacking {
249        publications: Rc<RefCell<Vec<BusMessage>>>,
250        closed: Rc<Cell<bool>>,
251        close_count: Rc<Cell<u32>>,
252    }
253
254    impl MessageBusBacking for CapturingPublishBacking {
255        fn is_closed(&self) -> bool {
256            self.closed.get()
257        }
258
259        fn publish(&self, message: BusMessage) {
260            self.publications.borrow_mut().push(message);
261        }
262
263        fn close(&mut self) {
264            self.close_count.set(self.close_count.get() + 1);
265            self.closed.set(true);
266        }
267    }
268
269    struct CapturingExternalIngress {
270        rx: Option<tokio::sync::mpsc::Receiver<BusMessage>>,
271        closed: bool,
272    }
273
274    impl ReexportedMessageBusExternalIngress for CapturingExternalIngress {
275        fn is_closed(&self) -> bool {
276            self.closed
277        }
278
279        fn take_receiver(&mut self) -> anyhow::Result<tokio::sync::mpsc::Receiver<BusMessage>> {
280            self.rx
281                .take()
282                .ok_or_else(|| anyhow::anyhow!("Stream receiver already taken"))
283        }
284
285        fn close(&mut self) {
286            self.closed = true;
287        }
288    }
289
290    #[rstest]
291    fn test_message_bus_external_ingress_reexport_accepts_bus_messages() {
292        let (tx, rx) = tokio::sync::mpsc::channel::<BusMessage>(1);
293        let mut ingress = CapturingExternalIngress {
294            rx: Some(rx),
295            closed: false,
296        };
297        let message = BusMessage::with_str_topic(
298            "events/data",
299            BusPayloadType::QuoteTick,
300            Bytes::from_static(b"payload"),
301            SerializationEncoding::Json,
302        );
303
304        tx.try_send(message.clone()).unwrap();
305        let mut stream_rx =
306            ReexportedMessageBusExternalIngress::take_receiver(&mut ingress).unwrap();
307        let received = stream_rx.try_recv().unwrap();
308
309        assert_eq!(received.topic, message.topic);
310        assert_eq!(received.payload, message.payload);
311        assert!(ReexportedMessageBusExternalIngress::take_receiver(&mut ingress).is_err());
312
313        ReexportedMessageBusExternalIngress::close(&mut ingress);
314        assert!(ReexportedMessageBusExternalIngress::is_closed(&ingress));
315    }
316
317    #[rstest]
318    fn test_external_io_from_backing_shares_close_state() {
319        let (tx, rx) = tokio::sync::mpsc::channel::<BusMessage>(1);
320        let backing = CapturingBacking {
321            rx: Some(rx),
322            closed: false,
323        };
324        let message = BusMessage::with_str_topic(
325            "events/data",
326            BusPayloadType::QuoteTick,
327            Bytes::from_static(b"payload"),
328            SerializationEncoding::Json,
329        );
330        let (mut egress, mut ingress) = external_io_from_backing(Box::new(backing));
331
332        tx.try_send(message.clone()).unwrap();
333        let mut stream_rx = ingress.take_receiver().unwrap();
334        let received = stream_rx.try_recv().unwrap();
335
336        assert_eq!(received.topic, message.topic);
337        assert!(!egress.is_closed());
338        assert!(!ingress.is_closed());
339
340        egress.close();
341
342        assert!(egress.is_closed());
343        assert!(ingress.is_closed());
344    }
345
346    #[rstest]
347    fn test_external_egress_from_backing_forwards_publications() {
348        let publications = Rc::new(RefCell::new(Vec::new()));
349        let closed = Rc::new(Cell::new(false));
350        let close_count = Rc::new(Cell::new(0));
351        let backing = CapturingPublishBacking {
352            publications: publications.clone(),
353            closed: closed.clone(),
354            close_count,
355        };
356        let mut egress = external_egress_from_backing(Box::new(backing));
357        let message = BusMessage::with_str_topic(
358            "events/data",
359            BusPayloadType::QuoteTick,
360            Bytes::from_static(b"payload"),
361            SerializationEncoding::Json,
362        );
363
364        egress.publish(message.clone());
365        egress.close();
366
367        let publications = publications.borrow();
368        assert_eq!(publications.len(), 1);
369        assert_eq!(publications[0].topic, message.topic);
370        assert!(closed.get());
371    }
372
373    #[rstest]
374    fn test_external_io_from_backing_closes_shared_backing_once() {
375        let publications = Rc::new(RefCell::new(Vec::new()));
376        let closed = Rc::new(Cell::new(false));
377        let close_count = Rc::new(Cell::new(0));
378        let backing = CapturingPublishBacking {
379            publications,
380            closed,
381            close_count: close_count.clone(),
382        };
383        let (mut egress, mut ingress) = external_io_from_backing(Box::new(backing));
384
385        egress.close();
386        ingress.close();
387
388        assert_eq!(close_count.get(), 1);
389    }
390
391    #[rstest]
392    fn test_external_io_from_backing_close_does_not_depend_on_backing_is_closed() {
393        let publications = Rc::new(RefCell::new(Vec::new()));
394        let closed = Rc::new(Cell::new(true));
395        let close_count = Rc::new(Cell::new(0));
396        let backing = CapturingPublishBacking {
397            publications,
398            closed,
399            close_count: close_count.clone(),
400        };
401        let (mut egress, mut ingress) = external_io_from_backing(Box::new(backing));
402
403        egress.close();
404        ingress.close();
405
406        assert_eq!(close_count.get(), 1);
407    }
408
409    #[rstest]
410    fn test_external_io_from_backing_default_receiver_is_unavailable() {
411        let publications = Rc::new(RefCell::new(Vec::new()));
412        let closed = Rc::new(Cell::new(false));
413        let close_count = Rc::new(Cell::new(0));
414        let backing = CapturingPublishBacking {
415            publications,
416            closed,
417            close_count,
418        };
419        let (_egress, mut ingress) = external_io_from_backing(Box::new(backing));
420
421        let error = ingress
422            .take_receiver()
423            .expect_err("egress-only backing should not provide ingress receiver");
424
425        assert_eq!(error.to_string(), "external ingress receiver unavailable");
426    }
427}
nautilus_common/msgbus/backing.rs

nautilus_common/msgbus/
backing.rs
