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

nautilus_common/python/
msgbus.rs
