nautilus_common/python/
msgbus.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//! Python bindings for the message bus, including configuration types and the
17//! [`PyMessageBus`] wrapper that routes Python events through the Rust
18//! thread-local [`MessageBus`] via the Any-based dispatch path.
19
20use std::{any::Any, fmt::Debug, rc::Rc, sync::LazyLock};
21
22use ahash::AHashMap;
23use nautilus_core::{
24    UUID4,
25    python::{to_pyruntime_err, to_pyvalue_err},
26};
27use nautilus_model::identifiers::TraderId;
28use pyo3::{Py, Python, prelude::*, types::PyBytes};
29use ustr::Ustr;
30
31use crate::{
32    enums::SerializationEncoding,
33    msgbus::{
34        self as msgbus_api, BusMessage, MessageBus, MessageBusBackingFactory, MessageBusConfig,
35        core::Subscription,
36        get_message_bus,
37        matching::is_matching,
38        mstr::{Endpoint, MStr, Pattern, Topic},
39        typed_handler::{Handler, ShareableMessageHandler, TypedHandler},
40    },
41    python::{
42        config_error_to_pyvalue_err,
43        factory::{FactoryExtractor, FactoryRegistry},
44    },
45};
46
47/// Function type for extracting a Python object into a boxed message bus backing factory.
48pub type MessageBusFactoryExtractor = FactoryExtractor<dyn MessageBusBackingFactory>;
49
50/// Registry for Python message bus backing factory extractors.
51#[derive(Debug)]
52pub struct MessageBusFactoryRegistry {
53    inner: FactoryRegistry<dyn MessageBusBackingFactory>,
54}
55
56impl MessageBusFactoryRegistry {
57    /// Creates an empty registry.
58    #[must_use]
59    pub fn new() -> Self {
60        Self {
61            inner: FactoryRegistry::new("message bus factory"),
62        }
63    }
64
65    // panics-doc-ok (transitive via FactoryRegistry mutex locking)
66    /// Registers an extractor for a Python factory type name.
67    ///
68    /// # Errors
69    ///
70    /// Returns an error if a different extractor is already registered for the type name.
71    pub fn register(
72        &self,
73        type_name: String,
74        extractor: MessageBusFactoryExtractor,
75    ) -> anyhow::Result<()> {
76        self.inner.register(type_name, extractor)
77    }
78
79    // panics-doc-ok (transitive via FactoryRegistry mutex locking)
80    /// Extracts a Python object into a boxed message bus backing factory.
81    ///
82    /// # Errors
83    ///
84    /// Returns an error if no extractor is registered for the Python type or extraction fails.
85    pub fn extract(
86        &self,
87        py: Python<'_>,
88        factory: Py<PyAny>,
89    ) -> PyResult<Box<dyn MessageBusBackingFactory>> {
90        self.inner.extract(py, factory)
91    }
92}
93
94impl Default for MessageBusFactoryRegistry {
95    fn default() -> Self {
96        Self::new()
97    }
98}
99
100static GLOBAL_MSGBUS_FACTORY_REGISTRY: LazyLock<MessageBusFactoryRegistry> =
101    LazyLock::new(MessageBusFactoryRegistry::new);
102
103/// Returns the global Python message bus backing factory registry.
104#[must_use]
105pub fn get_global_msgbus_factory_registry() -> &'static MessageBusFactoryRegistry {
106    &GLOBAL_MSGBUS_FACTORY_REGISTRY
107}
108
109#[pymethods]
110#[pyo3_stub_gen::derive::gen_stub_pymethods]
111impl BusMessage {
112    #[getter]
113    #[pyo3(name = "topic")]
114    fn py_topic(&self) -> String {
115        self.topic.to_string()
116    }
117
118    #[getter]
119    #[pyo3(name = "payload_type")]
120    fn py_payload_type(&self) -> String {
121        self.payload_type.to_string()
122    }
123
124    #[getter]
125    #[pyo3(name = "payload")]
126    fn py_payload(&self, py: Python<'_>) -> Py<PyBytes> {
127        PyBytes::new(py, self.payload.as_ref()).into()
128    }
129
130    #[getter]
131    #[pyo3(name = "encoding")]
132    fn py_encoding(&self) -> SerializationEncoding {
133        self.encoding
134    }
135
136    fn __repr__(&self) -> String {
137        format!("{}('{}')", stringify!(BusMessage), self)
138    }
139
140    fn __str__(&self) -> String {
141        self.to_string()
142    }
143}
144
145#[pymethods]
146#[pyo3_stub_gen::derive::gen_stub_pymethods]
147impl MessageBusConfig {
148    /// Configuration for `MessageBus` instances.
149    #[new]
150    #[expect(clippy::too_many_arguments)]
151    #[pyo3(signature = (encoding=None, encoding_market_data=None, encoding_builtin=None, timestamps_as_iso8601=None, buffer_interval_ms=None, autotrim_mins=None, autotrim_maxlen=None, use_trader_prefix=None, use_trader_id=None, use_instance_id=None, streams_prefix=None, stream_per_topic=None, external_streams=None, types_filter=None, heartbeat_interval_secs=None))]
152    fn py_new(
153        encoding: Option<SerializationEncoding>,
154        encoding_market_data: Option<SerializationEncoding>,
155        encoding_builtin: Option<SerializationEncoding>,
156        timestamps_as_iso8601: Option<bool>,
157        buffer_interval_ms: Option<u32>,
158        autotrim_mins: Option<u32>,
159        autotrim_maxlen: Option<u32>,
160        use_trader_prefix: Option<bool>,
161        use_trader_id: Option<bool>,
162        use_instance_id: Option<bool>,
163        streams_prefix: Option<String>,
164        stream_per_topic: Option<bool>,
165        external_streams: Option<Vec<String>>,
166        types_filter: Option<Vec<String>>,
167        heartbeat_interval_secs: Option<u16>,
168    ) -> PyResult<Self> {
169        let default = Self::default();
170        let config = Self {
171            encoding: encoding.unwrap_or(default.encoding),
172            encoding_market_data,
173            encoding_builtin,
174            timestamps_as_iso8601: timestamps_as_iso8601.unwrap_or(default.timestamps_as_iso8601),
175            buffer_interval_ms,
176            autotrim_mins,
177            autotrim_maxlen,
178            use_trader_prefix: use_trader_prefix.unwrap_or(default.use_trader_prefix),
179            use_trader_id: use_trader_id.unwrap_or(default.use_trader_id),
180            use_instance_id: use_instance_id.unwrap_or(default.use_instance_id),
181            streams_prefix: streams_prefix.unwrap_or(default.streams_prefix),
182            stream_per_topic: stream_per_topic.unwrap_or(default.stream_per_topic),
183            external_streams,
184            types_filter,
185            heartbeat_interval_secs,
186        };
187
188        config.validate().map_err(config_error_to_pyvalue_err)?;
189        Ok(config)
190    }
191
192    fn __repr__(&self) -> String {
193        format!("{self:?}")
194    }
195
196    fn __str__(&self) -> String {
197        format!("{self:?}")
198    }
199
200    #[getter]
201    fn encoding(&self) -> SerializationEncoding {
202        self.encoding
203    }
204
205    #[getter]
206    fn encoding_market_data(&self) -> Option<SerializationEncoding> {
207        self.encoding_market_data
208    }
209
210    #[getter]
211    fn encoding_builtin(&self) -> Option<SerializationEncoding> {
212        self.encoding_builtin
213    }
214
215    #[getter]
216    fn timestamps_as_iso8601(&self) -> bool {
217        self.timestamps_as_iso8601
218    }
219
220    #[getter]
221    fn buffer_interval_ms(&self) -> Option<u32> {
222        self.buffer_interval_ms
223    }
224
225    #[getter]
226    fn autotrim_mins(&self) -> Option<u32> {
227        self.autotrim_mins
228    }
229
230    #[getter]
231    fn autotrim_maxlen(&self) -> Option<u32> {
232        self.autotrim_maxlen
233    }
234
235    #[getter]
236    fn use_trader_prefix(&self) -> bool {
237        self.use_trader_prefix
238    }
239
240    #[getter]
241    fn use_trader_id(&self) -> bool {
242        self.use_trader_id
243    }
244
245    #[getter]
246    fn use_instance_id(&self) -> bool {
247        self.use_instance_id
248    }
249
250    #[getter]
251    fn streams_prefix(&self) -> &str {
252        &self.streams_prefix
253    }
254
255    #[getter]
256    fn stream_per_topic(&self) -> bool {
257        self.stream_per_topic
258    }
259
260    #[getter]
261    fn external_streams(&self) -> Option<Vec<String>> {
262        self.external_streams.clone()
263    }
264
265    #[getter]
266    fn types_filter(&self) -> Option<Vec<String>> {
267        self.types_filter.clone()
268    }
269
270    #[getter]
271    fn heartbeat_interval_secs(&self) -> Option<u16> {
272        self.heartbeat_interval_secs
273    }
274}
275
276/// Wraps a Python object so it can travel through the Rust Any-based message bus.
277pub struct PyMessage(pub Py<PyAny>);
278
279impl Debug for PyMessage {
280    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
281        f.debug_tuple(stringify!(PyMessage))
282            .field(&"<PyObject>")
283            .finish()
284    }
285}
286
287/// Adapts a Python callable as a [`ShareableMessageHandler`].
288///
289/// Expects messages to be [`PyMessage`] instances. Acquires the GIL and calls
290/// the Python callable with the inner Python object.
291pub struct PyCallableHandler {
292    id: Ustr,
293    callable: Py<PyAny>,
294}
295
296impl Debug for PyCallableHandler {
297    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
298        f.debug_struct(stringify!(PyCallableHandler))
299            .field("id", &self.id)
300            .finish()
301    }
302}
303
304impl PyCallableHandler {
305    /// Creates a new handler from a Python callable.
306    ///
307    /// The handler ID is derived from `repr(callable)` for stable identity
308    /// across subscribe/unsubscribe calls.
309    pub fn new(py: Python<'_>, callable: Py<PyAny>) -> PyResult<Self> {
310        let repr_str = callable.bind(py).repr()?.to_string();
311        let id = Ustr::from(&repr_str);
312        Ok(Self { id, callable })
313    }
314}
315
316impl Handler<dyn Any> for PyCallableHandler {
317    fn id(&self) -> Ustr {
318        self.id
319    }
320
321    fn handle(&self, message: &dyn Any) {
322        if let Some(py_msg) = message.downcast_ref::<PyMessage>() {
323            Python::attach(|py| {
324                if let Err(e) = self.callable.call1(py, (&py_msg.0,)) {
325                    log::error!("Python handler {id} failed: {e}", id = self.id);
326                }
327            });
328        } else {
329            log::error!(
330                "Python handler {id} received non-PyMessage type",
331                id = self.id
332            );
333        }
334    }
335}
336
337fn make_handler(py: Python<'_>, callable: Py<PyAny>) -> PyResult<ShareableMessageHandler> {
338    let handler = PyCallableHandler::new(py, callable)?;
339    Ok(TypedHandler(Rc::new(handler) as Rc<dyn Handler<dyn Any>>))
340}
341
342/// Python message bus backed by the Rust thread-local [`MessageBus`].
343///
344/// Publish, subscribe, and request/response calls from Python route through the
345/// single Rust bus. Python custom events travel through the Any-based dispatch
346/// path via [`PyMessage`] wrappers.
347#[pyclass(module = "nautilus_trader.common", name = "MessageBus", unsendable)]
348#[pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.common")]
349pub struct PyMessageBus {
350    trader_id: TraderId,
351    instance_id: UUID4,
352    name: String,
353    has_backing: bool,
354    serializer: Option<Py<PyAny>>,
355    backing: Option<Py<PyAny>>,
356    listeners: Vec<Py<PyAny>>,
357    types_filter: Option<Py<PyAny>>,
358    streaming_types: Vec<Py<PyAny>>,
359    correlation_index: AHashMap<UUID4, Py<PyAny>>,
360    sent_count: u64,
361    req_count: u64,
362    res_count: u64,
363    pub_count: u64,
364}
365
366impl Debug for PyMessageBus {
367    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
368        f.debug_struct(stringify!(PyMessageBus))
369            .field("trader_id", &self.trader_id)
370            .field("name", &self.name)
371            .finish()
372    }
373}
374
375#[pymethods]
376#[pyo3_stub_gen::derive::gen_stub_pymethods]
377impl PyMessageBus {
378    /// Creates a new `MessageBus` instance.
379    ///
380    /// This creates and registers the underlying Rust `MessageBus` as the
381    /// thread-local bus, then wraps it for Python access.
382    #[new]
383    #[pyo3(signature = (trader_id, clock=None, instance_id=None, name=None, serializer=None, backing=None, config=None))]
384    #[expect(clippy::too_many_arguments, clippy::needless_pass_by_value)]
385    fn py_new(
386        py: Python<'_>,
387        trader_id: TraderId,
388        clock: Option<Py<PyAny>>,
389        instance_id: Option<UUID4>,
390        name: Option<String>,
391        serializer: Option<Py<PyAny>>,
392        backing: Option<Py<PyAny>>,
393        config: Option<Py<PyAny>>,
394    ) -> PyResult<Self> {
395        let _ = clock;
396        let instance_id = instance_id.unwrap_or_default();
397        let bus_name = name.clone();
398        let has_backing = backing.is_some();
399
400        let msgbus = MessageBus::new(trader_id, instance_id, bus_name, None);
401        msgbus.register_message_bus();
402
403        let types_filter = if let Some(ref cfg) = config {
404            let tf = cfg.getattr(py, "types_filter")?;
405            if tf.is_none(py) {
406                None
407            } else {
408                // Convert to tuple for isinstance() checks
409                let tuple = py
410                    .import("builtins")?
411                    .call_method1("tuple", (tf,))?
412                    .unbind();
413                Some(tuple)
414            }
415        } else {
416            None
417        };
418
419        Ok(Self {
420            trader_id,
421            instance_id,
422            name: name.unwrap_or_else(|| "MessageBus".to_owned()),
423            has_backing,
424            serializer,
425            backing,
426            listeners: Vec::new(),
427            types_filter,
428            streaming_types: Vec::new(),
429            correlation_index: AHashMap::new(),
430            sent_count: 0,
431            req_count: 0,
432            res_count: 0,
433            pub_count: 0,
434        })
435    }
436
437    /// Returns the trader ID associated with the message bus.
438    #[getter]
439    #[pyo3(name = "trader_id")]
440    fn py_trader_id(&self) -> TraderId {
441        self.trader_id
442    }
443
444    /// Returns the instance ID associated with the message bus.
445    #[getter]
446    #[pyo3(name = "instance_id")]
447    fn py_instance_id(&self) -> UUID4 {
448        self.instance_id
449    }
450
451    /// Returns the name of the message bus.
452    #[getter]
453    #[pyo3(name = "name")]
454    fn py_name(&self) -> &str {
455        &self.name
456    }
457
458    /// Returns whether the message bus has an external backing.
459    #[getter]
460    #[pyo3(name = "has_backing")]
461    fn py_has_backing(&self) -> bool {
462        self.has_backing
463    }
464
465    /// Returns the count of messages sent via point-to-point.
466    #[getter]
467    #[pyo3(name = "sent_count")]
468    fn py_sent_count(&self) -> u64 {
469        self.sent_count
470    }
471
472    /// Returns the count of requests made.
473    #[getter]
474    #[pyo3(name = "req_count")]
475    fn py_req_count(&self) -> u64 {
476        self.req_count
477    }
478
479    /// Returns the count of responses handled.
480    #[getter]
481    #[pyo3(name = "res_count")]
482    fn py_res_count(&self) -> u64 {
483        self.res_count
484    }
485
486    /// Returns the count of messages published.
487    #[getter]
488    #[pyo3(name = "pub_count")]
489    fn py_pub_count(&self) -> u64 {
490        self.pub_count
491    }
492
493    /// Returns all registered endpoint addresses.
494    #[pyo3(name = "endpoints")]
495    fn py_endpoints(&self) -> Vec<String> {
496        let bus = get_message_bus();
497        let bus_ref = bus.borrow();
498        bus_ref.endpoints().into_iter().map(String::from).collect()
499    }
500
501    /// Returns all topics with active subscribers.
502    #[pyo3(name = "topics")]
503    fn py_topics(&self) -> Vec<String> {
504        let bus = get_message_bus();
505        let bus_ref = bus.borrow();
506        let mut topics: Vec<String> = bus_ref.patterns().into_iter().map(String::from).collect();
507        topics.sort();
508        topics.dedup();
509        topics
510    }
511
512    /// Returns subscriptions matching the given topic pattern.
513    #[pyo3(name = "subscriptions")]
514    #[pyo3(signature = (pattern=None))]
515    fn py_subscriptions(&self, pattern: Option<&str>) -> PyResult<Vec<String>> {
516        let filter = pattern.map(parse_pattern).transpose()?;
517
518        let bus = get_message_bus();
519        let bus_ref = bus.borrow();
520        let subs: Vec<&Subscription> = bus_ref.subscriptions();
521
522        Ok(subs
523            .into_iter()
524            .filter(|s| filter.is_none_or(|f| is_matching(s.pattern.as_bytes(), f.as_bytes())))
525            .map(|s| {
526                format!(
527                    "Subscription(topic={}, handler={})",
528                    s.pattern, s.handler_id
529                )
530            })
531            .collect())
532    }
533
534    /// Returns whether there are subscribers for the given topic pattern.
535    #[pyo3(name = "has_subscribers")]
536    #[pyo3(signature = (pattern=None))]
537    fn py_has_subscribers(&self, pattern: Option<&str>) -> PyResult<bool> {
538        let filter = pattern.map(parse_pattern).transpose()?;
539
540        let bus = get_message_bus();
541        let bus_ref = bus.borrow();
542
543        Ok(match filter {
544            Some(filter) => bus_ref
545                .subscriptions()
546                .iter()
547                .any(|s| is_matching(s.pattern.as_bytes(), filter.as_bytes())),
548            None => !bus_ref.subscriptions().is_empty(),
549        })
550    }
551
552    /// Returns whether the given topic and handler is subscribed.
553    #[pyo3(name = "is_subscribed")]
554    fn py_is_subscribed(&self, py: Python<'_>, topic: &str, handler: Py<PyAny>) -> PyResult<bool> {
555        let pattern = parse_pattern(topic)?;
556        let handler = make_handler(py, handler)?;
557        let sub = Subscription::new(pattern, handler, None);
558        Ok(get_message_bus().borrow().subscriptions.contains(&sub))
559    }
560
561    /// Returns whether the given request ID is pending a response.
562    #[pyo3(name = "is_pending_request")]
563    fn py_is_pending_request(&self, request_id: UUID4) -> bool {
564        self.correlation_index.contains_key(&request_id)
565    }
566
567    /// Returns whether the given type is registered for streaming.
568    #[pyo3(name = "is_streaming_type")]
569    #[expect(clippy::needless_pass_by_value)]
570    fn py_is_streaming_type(&self, py: Python<'_>, cls: Py<PyAny>) -> bool {
571        let cls_ref = cls.bind(py);
572        self.streaming_types.iter().any(|t| t.bind(py).is(cls_ref))
573    }
574
575    /// Returns all types registered for streaming.
576    #[pyo3(name = "streaming_types")]
577    fn py_streaming_types(&self, py: Python<'_>) -> Vec<Py<PyAny>> {
578        self.streaming_types
579            .iter()
580            .map(|t| t.clone_ref(py))
581            .collect()
582    }
583
584    /// Registers a handler at the given endpoint address.
585    #[pyo3(name = "register")]
586    fn py_register(&self, py: Python<'_>, endpoint: &str, handler: Py<PyAny>) -> PyResult<()> {
587        let endpoint = parse_endpoint(endpoint)?;
588        let handler = make_handler(py, handler)?;
589        msgbus_api::register_any(endpoint, handler);
590        Ok(())
591    }
592
593    /// Deregisters the handler from the given endpoint address.
594    #[pyo3(name = "deregister")]
595    #[pyo3(signature = (endpoint, handler=None))]
596    #[expect(clippy::needless_pass_by_value)]
597    fn py_deregister(&self, endpoint: &str, handler: Option<Py<PyAny>>) -> PyResult<()> {
598        let _ = handler;
599        let endpoint = parse_endpoint(endpoint)?;
600        msgbus_api::deregister_any(endpoint);
601        Ok(())
602    }
603
604    /// Sends a message to the given endpoint address.
605    #[pyo3(name = "send")]
606    fn py_send(&mut self, endpoint: &str, msg: Py<PyAny>) -> PyResult<()> {
607        let endpoint = parse_endpoint(endpoint)?;
608        let py_msg = PyMessage(msg);
609        msgbus_api::send_any(endpoint, &py_msg);
610        self.sent_count += 1;
611        Ok(())
612    }
613
614    /// Sends a request to the given endpoint with correlation tracking.
615    #[pyo3(name = "request")]
616    fn py_request(&mut self, py: Python<'_>, endpoint: &str, request: Py<PyAny>) -> PyResult<()> {
617        let endpoint = parse_endpoint(endpoint)?;
618        let request_ref = request.bind(py);
619
620        let request_id: UUID4 = request_ref.getattr("id")?.extract()?;
621        let callback = request_ref.getattr("callback")?;
622
623        if self.correlation_index.contains_key(&request_id) {
624            log::error!(
625                "Cannot handle request: duplicate ID {request_id} found in correlation index"
626            );
627            return Ok(());
628        }
629
630        if !callback.is_none() {
631            self.correlation_index.insert(request_id, callback.unbind());
632        }
633
634        let py_msg = PyMessage(request);
635        msgbus_api::send_any(endpoint, &py_msg);
636        self.req_count += 1;
637
638        Ok(())
639    }
640
641    /// Handles a response by invoking the correlated callback.
642    #[pyo3(name = "response")]
643    #[expect(clippy::needless_pass_by_value)]
644    fn py_response(&mut self, py: Python<'_>, response: Py<PyAny>) -> PyResult<()> {
645        let correlation_id: UUID4 = response.getattr(py, "correlation_id")?.extract(py)?;
646
647        if let Some(callback) = self.correlation_index.remove(&correlation_id) {
648            callback.call1(py, (&response,))?;
649        } else {
650            log::debug!("No callback for correlation_id {correlation_id}");
651        }
652
653        self.res_count += 1;
654        Ok(())
655    }
656
657    /// Subscribes to the given topic with the given handler.
658    #[pyo3(name = "subscribe")]
659    #[pyo3(signature = (topic, handler, priority=0))]
660    fn py_subscribe(
661        &self,
662        py: Python<'_>,
663        topic: &str,
664        handler: Py<PyAny>,
665        priority: u32,
666    ) -> PyResult<()> {
667        let pattern = parse_pattern(topic)?;
668        let handler = make_handler(py, handler)?;
669        msgbus_api::subscribe_any(pattern, handler, Some(priority));
670        Ok(())
671    }
672
673    /// Unsubscribes the given handler from the given topic.
674    #[pyo3(name = "unsubscribe")]
675    fn py_unsubscribe(&self, py: Python<'_>, topic: &str, handler: Py<PyAny>) -> PyResult<()> {
676        let pattern = parse_pattern(topic)?;
677        let handler = make_handler(py, handler)?;
678        msgbus_api::unsubscribe_any(pattern, &handler);
679        Ok(())
680    }
681
682    /// Publishes a message for the given topic.
683    #[pyo3(name = "publish")]
684    #[pyo3(signature = (topic, msg, external_pub=true))]
685    #[expect(clippy::needless_pass_by_value)]
686    fn py_publish(
687        &mut self,
688        py: Python<'_>,
689        topic: &str,
690        msg: Py<PyAny>,
691        external_pub: bool,
692    ) -> PyResult<()> {
693        let topic_mstr = MStr::<Topic>::topic(topic).map_err(to_pyruntime_err)?;
694
695        let py_msg = PyMessage(msg.clone_ref(py));
696        msgbus_api::publish_any(topic_mstr, &py_msg);
697
698        if external_pub {
699            self.publish_external(py, topic, &msg)?;
700        }
701
702        self.pub_count += 1;
703        Ok(())
704    }
705
706    /// Disposes of the message bus, clearing all state.
707    #[pyo3(name = "dispose")]
708    fn py_dispose(&mut self, py: Python<'_>) -> PyResult<()> {
709        log::debug!("Closing message bus");
710
711        get_message_bus().borrow_mut().dispose();
712
713        self.correlation_index.clear();
714        self.listeners.clear();
715        self.streaming_types.clear();
716
717        if let Some(ref backing) = self.backing {
718            let db = backing.bind(py);
719            if !db.call_method0("is_closed")?.extract::<bool>()? {
720                db.call_method0("close")?;
721            }
722        }
723
724        log::info!("Closed message bus");
725        Ok(())
726    }
727
728    /// Registers a type for external-to-internal message streaming.
729    #[pyo3(name = "add_streaming_type")]
730    fn py_add_streaming_type(&mut self, cls: Py<PyAny>) {
731        self.streaming_types.push(cls);
732    }
733
734    /// Adds a listener to the message bus.
735    #[pyo3(name = "add_listener")]
736    fn py_add_listener(&mut self, listener: Py<PyAny>) {
737        self.listeners.push(listener);
738    }
739}
740
741impl PyMessageBus {
742    fn publish_external(&self, py: Python<'_>, topic: &str, msg: &Py<PyAny>) -> PyResult<()> {
743        if let Some(ref filter) = self.types_filter {
744            let is_excluded = py
745                .import("builtins")?
746                .call_method1("isinstance", (msg, filter))?
747                .extract::<bool>()?;
748
749            if is_excluded {
750                return Ok(());
751            }
752        }
753
754        // Serialize: raw bytes pass through, other types need a serializer
755        let msg_ref = msg.bind(py);
756        let payload: Py<PyAny> = if msg_ref.is_instance_of::<pyo3::types::PyBytes>() {
757            msg.clone_ref(py)
758        } else if let Some(ref serializer) = self.serializer {
759            serializer.call_method1(py, "serialize", (msg,))?
760        } else {
761            return Ok(());
762        };
763
764        if let Some(ref backing) = self.backing {
765            let db = backing.bind(py);
766            if !db.call_method0("is_closed")?.extract::<bool>()? {
767                db.call_method1("publish", (topic, &payload))?;
768            }
769        }
770
771        for listener in &self.listeners {
772            let l = listener.bind(py);
773            if l.call_method0("is_closed")?.extract::<bool>()? {
774                continue;
775            }
776            l.call_method1("publish", (topic, &payload))?;
777        }
778
779        Ok(())
780    }
781}
782
783fn parse_endpoint(endpoint: &str) -> PyResult<MStr<Endpoint>> {
784    MStr::<Endpoint>::endpoint(endpoint).map_err(to_pyvalue_err)
785}
786
787fn parse_pattern(pattern: &str) -> PyResult<MStr<Pattern>> {
788    MStr::<Pattern>::pattern_checked(pattern).map_err(to_pyvalue_err)
789}
790
791#[cfg(test)]
792mod tests {
793    use std::any::Any;
794
795    use pyo3::{exceptions::PyValueError, ffi::c_str};
796    use rstest::rstest;
797
798    use super::*;
799
800    #[rstest]
801    fn test_message_bus_factory_registry_compatibility_constructors() {
802        let registry = MessageBusFactoryRegistry::new();
803        let default_registry = MessageBusFactoryRegistry::default();
804
805        assert_eq!(format!("{registry:?}"), format!("{default_registry:?}"));
806        assert!(format!("{registry:?}").contains("message bus factory"));
807    }
808
809    #[rstest]
810    fn message_bus_config_py_new_maps_validate_error_to_value_error() {
811        pyo3::Python::initialize();
812        Python::attach(|py| {
813            let err = MessageBusConfig::py_new(
814                Some(SerializationEncoding::Json),
815                None,
816                Some(SerializationEncoding::Capnp),
817                None,
818                None,
819                None,
820                None,
821                None,
822                None,
823                None,
824                None,
825                None,
826                None,
827                None,
828                None,
829            )
830            .unwrap_err();
831
832            assert!(err.is_instance_of::<PyValueError>(py));
833            assert_eq!(
834                err.value(py).to_string(),
835                format!(
836                    "MessageBusConfig.encoding_builtin has unsupported value: {} is not supported by AccountState, OrderEventAny, PositionEvent, PortfolioSnapshot",
837                    SerializationEncoding::Capnp
838                )
839            );
840        });
841    }
842
843    #[rstest]
844    fn test_py_message_downcast() {
845        pyo3::Python::initialize();
846        Python::attach(|py| {
847            let py_obj = py.eval(c_str!("42"), None, None).unwrap();
848            let msg = PyMessage(py_obj.unbind());
849
850            let any_ref: &dyn Any = &msg;
851            let downcasted = any_ref.downcast_ref::<PyMessage>();
852            assert!(downcasted.is_some());
853
854            let inner = &downcasted.unwrap().0;
855            let value: i64 = inner.extract(py).unwrap();
856            assert_eq!(value, 42);
857        });
858    }
859
860    #[rstest]
861    fn test_py_callable_handler_id_stability() {
862        pyo3::Python::initialize();
863        Python::attach(|py| {
864            let callable = py.eval(c_str!("lambda x: x"), None, None).unwrap().unbind();
865
866            let handler1 = PyCallableHandler::new(py, callable.clone_ref(py)).unwrap();
867            let handler2 = PyCallableHandler::new(py, callable).unwrap();
868
869            assert_eq!(handler1.id(), handler2.id());
870        });
871    }
872
873    #[rstest]
874    fn test_py_callable_handler_dispatch() {
875        pyo3::Python::initialize();
876        Python::attach(|py| {
877            let main = py.import("__main__").unwrap();
878            let globals = main.dict();
879            py.run(
880                c_str!("results = []\ndef handler(x): results.append(x)"),
881                Some(&globals),
882                None,
883            )
884            .unwrap();
885
886            let handler_fn = globals.get_item("handler").unwrap().unwrap().unbind();
887            let handler = PyCallableHandler::new(py, handler_fn).unwrap();
888
889            let py_obj = py.eval(c_str!("'hello'"), None, None).unwrap();
890            let msg = PyMessage(py_obj.unbind());
891
892            let any_ref: &dyn Any = &msg;
893            handler.handle(any_ref);
894
895            let results = globals.get_item("results").unwrap().unwrap();
896            let len: usize = results.len().unwrap();
897            assert_eq!(len, 1);
898        });
899    }
900}
nautilus_common/python/msgbus.rs

nautilus_common/python/
msgbus.rs
