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//! [`PyMessageBusScope`] owns each component's Python subscriptions.
20
21use std::{
22    any::Any,
23    cell::{Cell, RefCell},
24    fmt::Debug,
25    rc::{Rc, Weak},
26    sync::LazyLock,
27};
28
29use ahash::AHashMap;
30use nautilus_core::{
31    UUID4,
32    python::{to_pyruntime_err, to_pytype_err, to_pyvalue_err},
33};
34use nautilus_model::identifiers::TraderId;
35use pyo3::{Py, Python, prelude::*, types::PyBytes};
36use ustr::Ustr;
37
38use crate::{
39    enums::SerializationEncoding,
40    msgbus::{
41        self as msgbus_api, BusMessage, MessageBus, MessageBusBackingFactory, MessageBusConfig,
42        core::Subscription,
43        get_message_bus,
44        matching::is_matching,
45        mstr::{Endpoint, MStr, Pattern, Topic},
46        try_get_message_bus,
47        typed_handler::{Handler, ShareableMessageHandler, TypedHandler},
48    },
49    python::{
50        config_error_to_pyvalue_err,
51        factory::{FactoryExtractor, FactoryRegistry},
52    },
53};
54
55/// Function type for extracting a Python object into a boxed message bus backing factory.
56pub type MessageBusFactoryExtractor = FactoryExtractor<dyn MessageBusBackingFactory>;
57
58/// Registry for Python message bus backing factory extractors.
59#[derive(Debug)]
60pub struct MessageBusFactoryRegistry {
61    inner: FactoryRegistry<dyn MessageBusBackingFactory>,
62}
63
64impl MessageBusFactoryRegistry {
65    /// Creates an empty registry.
66    #[must_use]
67    pub fn new() -> Self {
68        Self {
69            inner: FactoryRegistry::new("message bus factory"),
70        }
71    }
72
73    // panics-doc-ok (transitive via FactoryRegistry mutex locking)
74    /// Registers an extractor for a Python factory type name.
75    ///
76    /// # Errors
77    ///
78    /// Returns an error if a different extractor is already registered for the type name.
79    pub fn register(
80        &self,
81        type_name: String,
82        extractor: MessageBusFactoryExtractor,
83    ) -> anyhow::Result<()> {
84        self.inner.register(type_name, extractor)
85    }
86
87    // panics-doc-ok (transitive via FactoryRegistry mutex locking)
88    /// Extracts a Python object into a boxed message bus backing factory.
89    ///
90    /// # Errors
91    ///
92    /// Returns an error if no extractor is registered for the Python type or extraction fails.
93    pub fn extract(
94        &self,
95        py: Python<'_>,
96        factory: Py<PyAny>,
97    ) -> PyResult<Box<dyn MessageBusBackingFactory>> {
98        self.inner.extract(py, factory)
99    }
100}
101
102impl Default for MessageBusFactoryRegistry {
103    fn default() -> Self {
104        Self::new()
105    }
106}
107
108static GLOBAL_MSGBUS_FACTORY_REGISTRY: LazyLock<MessageBusFactoryRegistry> =
109    LazyLock::new(MessageBusFactoryRegistry::new);
110
111/// Returns the global Python message bus backing factory registry.
112#[must_use]
113pub fn get_global_msgbus_factory_registry() -> &'static MessageBusFactoryRegistry {
114    &GLOBAL_MSGBUS_FACTORY_REGISTRY
115}
116
117#[pymethods]
118#[pyo3_stub_gen::derive::gen_stub_pymethods]
119impl BusMessage {
120    #[getter]
121    #[pyo3(name = "topic")]
122    fn py_topic(&self) -> String {
123        self.topic.to_string()
124    }
125
126    #[getter]
127    #[pyo3(name = "payload_type")]
128    fn py_payload_type(&self) -> String {
129        self.payload_type.to_string()
130    }
131
132    #[getter]
133    #[pyo3(name = "payload")]
134    fn py_payload(&self, py: Python<'_>) -> Py<PyBytes> {
135        PyBytes::new(py, self.payload.as_ref()).into()
136    }
137
138    #[getter]
139    #[pyo3(name = "encoding")]
140    fn py_encoding(&self) -> SerializationEncoding {
141        self.encoding
142    }
143
144    fn __repr__(&self) -> String {
145        format!("{}('{}')", stringify!(BusMessage), self)
146    }
147
148    fn __str__(&self) -> String {
149        self.to_string()
150    }
151}
152
153#[pymethods]
154#[pyo3_stub_gen::derive::gen_stub_pymethods]
155impl MessageBusConfig {
156    /// Configuration for `MessageBus` instances.
157    #[new]
158    #[expect(clippy::too_many_arguments)]
159    #[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))]
160    fn py_new(
161        encoding: Option<SerializationEncoding>,
162        encoding_market_data: Option<SerializationEncoding>,
163        encoding_builtin: Option<SerializationEncoding>,
164        timestamps_as_iso8601: Option<bool>,
165        buffer_interval_ms: Option<u32>,
166        autotrim_mins: Option<u32>,
167        autotrim_maxlen: Option<u32>,
168        use_trader_prefix: Option<bool>,
169        use_trader_id: Option<bool>,
170        use_instance_id: Option<bool>,
171        streams_prefix: Option<String>,
172        stream_per_topic: Option<bool>,
173        external_streams: Option<Vec<String>>,
174        types_filter: Option<Vec<String>>,
175        heartbeat_interval_secs: Option<u16>,
176    ) -> PyResult<Self> {
177        let default = Self::default();
178        let config = Self {
179            encoding: encoding.unwrap_or(default.encoding),
180            encoding_market_data,
181            encoding_builtin,
182            timestamps_as_iso8601: timestamps_as_iso8601.unwrap_or(default.timestamps_as_iso8601),
183            buffer_interval_ms,
184            autotrim_mins,
185            autotrim_maxlen,
186            use_trader_prefix: use_trader_prefix.unwrap_or(default.use_trader_prefix),
187            use_trader_id: use_trader_id.unwrap_or(default.use_trader_id),
188            use_instance_id: use_instance_id.unwrap_or(default.use_instance_id),
189            streams_prefix: streams_prefix.unwrap_or(default.streams_prefix),
190            stream_per_topic: stream_per_topic.unwrap_or(default.stream_per_topic),
191            external_streams,
192            types_filter,
193            heartbeat_interval_secs,
194        };
195
196        config.validate().map_err(config_error_to_pyvalue_err)?;
197        Ok(config)
198    }
199
200    fn __repr__(&self) -> String {
201        format!("{self:?}")
202    }
203
204    fn __str__(&self) -> String {
205        format!("{self:?}")
206    }
207
208    #[getter]
209    fn encoding(&self) -> SerializationEncoding {
210        self.encoding
211    }
212
213    #[getter]
214    fn encoding_market_data(&self) -> Option<SerializationEncoding> {
215        self.encoding_market_data
216    }
217
218    #[getter]
219    fn encoding_builtin(&self) -> Option<SerializationEncoding> {
220        self.encoding_builtin
221    }
222
223    #[getter]
224    fn timestamps_as_iso8601(&self) -> bool {
225        self.timestamps_as_iso8601
226    }
227
228    #[getter]
229    fn buffer_interval_ms(&self) -> Option<u32> {
230        self.buffer_interval_ms
231    }
232
233    #[getter]
234    fn autotrim_mins(&self) -> Option<u32> {
235        self.autotrim_mins
236    }
237
238    #[getter]
239    fn autotrim_maxlen(&self) -> Option<u32> {
240        self.autotrim_maxlen
241    }
242
243    #[getter]
244    fn use_trader_prefix(&self) -> bool {
245        self.use_trader_prefix
246    }
247
248    #[getter]
249    fn use_trader_id(&self) -> bool {
250        self.use_trader_id
251    }
252
253    #[getter]
254    fn use_instance_id(&self) -> bool {
255        self.use_instance_id
256    }
257
258    #[getter]
259    fn streams_prefix(&self) -> &str {
260        &self.streams_prefix
261    }
262
263    #[getter]
264    fn stream_per_topic(&self) -> bool {
265        self.stream_per_topic
266    }
267
268    #[getter]
269    fn external_streams(&self) -> Option<Vec<String>> {
270        self.external_streams.clone()
271    }
272
273    #[getter]
274    fn types_filter(&self) -> Option<Vec<String>> {
275        self.types_filter.clone()
276    }
277
278    #[getter]
279    fn heartbeat_interval_secs(&self) -> Option<u16> {
280        self.heartbeat_interval_secs
281    }
282}
283
284/// Wraps a Python object so it can travel through the Rust Any-based message bus.
285pub struct PyMessage(pub Py<PyAny>);
286
287impl Debug for PyMessage {
288    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
289        f.debug_tuple(stringify!(PyMessage))
290            .field(&"<PyObject>")
291            .finish()
292    }
293}
294
295/// Adapts a Python callable as a [`ShareableMessageHandler`].
296///
297/// Expects messages to be [`PyMessage`] instances. Acquires the GIL and calls
298/// the Python callable with the inner Python object.
299pub struct PyCallableHandler {
300    id: Ustr,
301    callable: Py<PyAny>,
302}
303
304impl Debug for PyCallableHandler {
305    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
306        f.debug_struct(stringify!(PyCallableHandler))
307            .field("id", &self.id)
308            .finish()
309    }
310}
311
312impl PyCallableHandler {
313    /// Creates a new handler from a Python callable.
314    ///
315    /// The handler ID is derived from `repr(callable)` for stable identity
316    /// across subscribe/unsubscribe calls.
317    pub fn new(py: Python<'_>, callable: Py<PyAny>) -> PyResult<Self> {
318        let repr_str = callable.bind(py).repr()?.to_string();
319        let id = Ustr::from(&repr_str);
320        Ok(Self::with_id(id, callable))
321    }
322
323    pub(crate) fn with_id(id: Ustr, callable: Py<PyAny>) -> Self {
324        Self { id, callable }
325    }
326}
327
328impl Handler<dyn Any> for PyCallableHandler {
329    fn id(&self) -> Ustr {
330        self.id
331    }
332
333    fn handle(&self, message: &dyn Any) {
334        if let Some(py_msg) = message.downcast_ref::<PyMessage>() {
335            Python::attach(|py| {
336                if let Err(e) = self.callable.call1(py, (&py_msg.0,)) {
337                    log::error!("Python handler {id} failed: {e}", id = self.id);
338                }
339            });
340        } else {
341            log::error!(
342                "Python handler {id} received non-PyMessage type",
343                id = self.id
344            );
345        }
346    }
347}
348
349fn make_handler(py: Python<'_>, callable: Py<PyAny>) -> PyResult<ShareableMessageHandler> {
350    let handler = PyCallableHandler::new(py, callable)?;
351    Ok(TypedHandler(Rc::new(handler) as Rc<dyn Handler<dyn Any>>))
352}
353
354/// Python message bus backed by the Rust thread-local [`MessageBus`].
355///
356/// Publish, subscribe, and request/response calls from Python route through the
357/// single Rust bus. Python custom events travel through the Any-based dispatch
358/// path via [`PyMessage`] wrappers.
359#[pyclass(module = "nautilus_trader.common", name = "MessageBus", unsendable)]
360#[pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.common")]
361pub struct PyMessageBus {
362    trader_id: TraderId,
363    instance_id: UUID4,
364    name: String,
365    has_backing: bool,
366    serializer: Option<Py<PyAny>>,
367    backing: Option<Py<PyAny>>,
368    listeners: Vec<Py<PyAny>>,
369    types_filter: Option<Py<PyAny>>,
370    streaming_types: Vec<Py<PyAny>>,
371    correlation_index: AHashMap<UUID4, Py<PyAny>>,
372    sent_count: u64,
373    req_count: u64,
374    res_count: u64,
375    pub_count: u64,
376}
377
378impl Debug for PyMessageBus {
379    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
380        f.debug_struct(stringify!(PyMessageBus))
381            .field("trader_id", &self.trader_id)
382            .field("name", &self.name)
383            .finish()
384    }
385}
386
387#[pymethods]
388#[pyo3_stub_gen::derive::gen_stub_pymethods]
389impl PyMessageBus {
390    /// Creates a new `MessageBus` instance.
391    ///
392    /// This creates and registers the underlying Rust `MessageBus` as the
393    /// thread-local bus, then wraps it for Python access.
394    #[new]
395    #[pyo3(signature = (trader_id, clock=None, instance_id=None, name=None, serializer=None, backing=None, config=None))]
396    #[expect(clippy::too_many_arguments, clippy::needless_pass_by_value)]
397    fn py_new(
398        py: Python<'_>,
399        trader_id: TraderId,
400        clock: Option<Py<PyAny>>,
401        instance_id: Option<UUID4>,
402        name: Option<String>,
403        serializer: Option<Py<PyAny>>,
404        backing: Option<Py<PyAny>>,
405        config: Option<Py<PyAny>>,
406    ) -> PyResult<Self> {
407        let _ = clock;
408        let instance_id = instance_id.unwrap_or_default();
409        let bus_name = name.clone();
410        let has_backing = backing.is_some();
411
412        let msgbus = MessageBus::new(trader_id, instance_id, bus_name, None);
413        msgbus.register_message_bus();
414
415        let types_filter = if let Some(ref cfg) = config {
416            let tf = cfg.getattr(py, "types_filter")?;
417            if tf.is_none(py) {
418                None
419            } else {
420                // Convert to tuple for isinstance() checks
421                let tuple = py
422                    .import("builtins")?
423                    .call_method1("tuple", (tf,))?
424                    .unbind();
425                Some(tuple)
426            }
427        } else {
428            None
429        };
430
431        Ok(Self {
432            trader_id,
433            instance_id,
434            name: name.unwrap_or_else(|| "MessageBus".to_owned()),
435            has_backing,
436            serializer,
437            backing,
438            listeners: Vec::new(),
439            types_filter,
440            streaming_types: Vec::new(),
441            correlation_index: AHashMap::new(),
442            sent_count: 0,
443            req_count: 0,
444            res_count: 0,
445            pub_count: 0,
446        })
447    }
448
449    /// Returns the trader ID associated with the message bus.
450    #[getter]
451    #[pyo3(name = "trader_id")]
452    fn py_trader_id(&self) -> TraderId {
453        self.trader_id
454    }
455
456    /// Returns the instance ID associated with the message bus.
457    #[getter]
458    #[pyo3(name = "instance_id")]
459    fn py_instance_id(&self) -> UUID4 {
460        self.instance_id
461    }
462
463    /// Returns the name of the message bus.
464    #[getter]
465    #[pyo3(name = "name")]
466    fn py_name(&self) -> &str {
467        &self.name
468    }
469
470    /// Returns whether the message bus has an external backing.
471    #[getter]
472    #[pyo3(name = "has_backing")]
473    fn py_has_backing(&self) -> bool {
474        self.has_backing
475    }
476
477    /// Returns the count of messages sent via point-to-point.
478    #[getter]
479    #[pyo3(name = "sent_count")]
480    fn py_sent_count(&self) -> u64 {
481        self.sent_count
482    }
483
484    /// Returns the count of requests made.
485    #[getter]
486    #[pyo3(name = "req_count")]
487    fn py_req_count(&self) -> u64 {
488        self.req_count
489    }
490
491    /// Returns the count of responses handled.
492    #[getter]
493    #[pyo3(name = "res_count")]
494    fn py_res_count(&self) -> u64 {
495        self.res_count
496    }
497
498    /// Returns the count of messages published.
499    #[getter]
500    #[pyo3(name = "pub_count")]
501    fn py_pub_count(&self) -> u64 {
502        self.pub_count
503    }
504
505    /// Returns all registered endpoint addresses.
506    #[pyo3(name = "endpoints")]
507    fn py_endpoints(&self) -> Vec<String> {
508        let bus = get_message_bus();
509        let bus_ref = bus.borrow();
510        bus_ref.endpoints().into_iter().map(String::from).collect()
511    }
512
513    /// Returns all topics with active subscribers.
514    #[pyo3(name = "topics")]
515    fn py_topics(&self) -> Vec<String> {
516        let bus = get_message_bus();
517        let bus_ref = bus.borrow();
518        let mut topics: Vec<String> = bus_ref.patterns().into_iter().map(String::from).collect();
519        topics.sort();
520        topics.dedup();
521        topics
522    }
523
524    /// Returns subscriptions matching the given topic pattern.
525    #[pyo3(name = "subscriptions")]
526    #[pyo3(signature = (pattern=None))]
527    fn py_subscriptions(&self, pattern: Option<&str>) -> PyResult<Vec<String>> {
528        let filter = pattern.map(parse_pattern).transpose()?;
529
530        let bus = get_message_bus();
531        let bus_ref = bus.borrow();
532        let subs: Vec<&Subscription> = bus_ref.subscriptions();
533
534        Ok(subs
535            .into_iter()
536            .filter(|s| filter.is_none_or(|f| is_matching(s.pattern.as_bytes(), f.as_bytes())))
537            .map(|s| {
538                format!(
539                    "Subscription(topic={}, handler={})",
540                    s.pattern, s.handler_id
541                )
542            })
543            .collect())
544    }
545
546    /// Returns whether there are subscribers for the given topic pattern.
547    #[pyo3(name = "has_subscribers")]
548    #[pyo3(signature = (pattern=None))]
549    fn py_has_subscribers(&self, pattern: Option<&str>) -> PyResult<bool> {
550        let filter = pattern.map(parse_pattern).transpose()?;
551
552        let bus = get_message_bus();
553        let bus_ref = bus.borrow();
554
555        Ok(match filter {
556            Some(filter) => bus_ref
557                .subscriptions()
558                .iter()
559                .any(|s| is_matching(s.pattern.as_bytes(), filter.as_bytes())),
560            None => !bus_ref.subscriptions().is_empty(),
561        })
562    }
563
564    /// Returns whether the given topic and handler is subscribed.
565    #[pyo3(name = "is_subscribed")]
566    fn py_is_subscribed(&self, py: Python<'_>, topic: &str, handler: Py<PyAny>) -> PyResult<bool> {
567        let pattern = parse_pattern(topic)?;
568        let handler = make_handler(py, handler)?;
569        let sub = Subscription::new(pattern, handler, None);
570        Ok(get_message_bus().borrow().subscriptions.contains(&sub))
571    }
572
573    /// Returns whether the given request ID is pending a response.
574    #[pyo3(name = "is_pending_request")]
575    fn py_is_pending_request(&self, request_id: UUID4) -> bool {
576        self.correlation_index.contains_key(&request_id)
577    }
578
579    /// Returns whether the given type is registered for streaming.
580    #[pyo3(name = "is_streaming_type")]
581    #[expect(clippy::needless_pass_by_value)]
582    fn py_is_streaming_type(&self, py: Python<'_>, cls: Py<PyAny>) -> bool {
583        let cls_ref = cls.bind(py);
584        self.streaming_types.iter().any(|t| t.bind(py).is(cls_ref))
585    }
586
587    /// Returns all types registered for streaming.
588    #[pyo3(name = "streaming_types")]
589    fn py_streaming_types(&self, py: Python<'_>) -> Vec<Py<PyAny>> {
590        self.streaming_types
591            .iter()
592            .map(|t| t.clone_ref(py))
593            .collect()
594    }
595
596    /// Registers a handler at the given endpoint address.
597    #[pyo3(name = "register")]
598    fn py_register(&self, py: Python<'_>, endpoint: &str, handler: Py<PyAny>) -> PyResult<()> {
599        let endpoint = parse_endpoint(endpoint)?;
600        let handler = make_handler(py, handler)?;
601        msgbus_api::register_any(endpoint, handler);
602        Ok(())
603    }
604
605    /// Deregisters the handler from the given endpoint address.
606    #[pyo3(name = "deregister")]
607    #[pyo3(signature = (endpoint, handler=None))]
608    #[expect(clippy::needless_pass_by_value)]
609    fn py_deregister(&self, endpoint: &str, handler: Option<Py<PyAny>>) -> PyResult<()> {
610        let _ = handler;
611        let endpoint = parse_endpoint(endpoint)?;
612        msgbus_api::deregister_any(endpoint);
613        Ok(())
614    }
615
616    /// Sends a message to the given endpoint address.
617    #[pyo3(name = "send")]
618    fn py_send(&mut self, endpoint: &str, msg: Py<PyAny>) -> PyResult<()> {
619        let endpoint = parse_endpoint(endpoint)?;
620        let py_msg = PyMessage(msg);
621        msgbus_api::send_any(endpoint, &py_msg);
622        self.sent_count += 1;
623        Ok(())
624    }
625
626    /// Sends a request to the given endpoint with correlation tracking.
627    #[pyo3(name = "request")]
628    fn py_request(&mut self, py: Python<'_>, endpoint: &str, request: Py<PyAny>) -> PyResult<()> {
629        let endpoint = parse_endpoint(endpoint)?;
630        let request_ref = request.bind(py);
631
632        let request_id: UUID4 = request_ref.getattr("id")?.extract()?;
633        let callback = request_ref.getattr("callback")?;
634
635        if self.correlation_index.contains_key(&request_id) {
636            log::error!(
637                "Cannot handle request: duplicate ID {request_id} found in correlation index"
638            );
639            return Ok(());
640        }
641
642        if !callback.is_none() {
643            self.correlation_index.insert(request_id, callback.unbind());
644        }
645
646        let py_msg = PyMessage(request);
647        msgbus_api::send_any(endpoint, &py_msg);
648        self.req_count += 1;
649        Ok(())
650    }
651
652    /// Handles a response by invoking the correlated callback.
653    #[pyo3(name = "response")]
654    #[expect(clippy::needless_pass_by_value)]
655    fn py_response(&mut self, py: Python<'_>, response: Py<PyAny>) -> PyResult<()> {
656        let correlation_id: UUID4 = response.getattr(py, "correlation_id")?.extract(py)?;
657
658        if let Some(callback) = self.correlation_index.remove(&correlation_id) {
659            callback.call1(py, (&response,))?;
660        } else {
661            log::debug!("No callback for correlation_id {correlation_id}");
662        }
663
664        self.res_count += 1;
665        Ok(())
666    }
667
668    /// Subscribes to the given topic with the given handler.
669    #[pyo3(name = "subscribe")]
670    #[pyo3(signature = (topic, handler, priority=0))]
671    fn py_subscribe(
672        &self,
673        py: Python<'_>,
674        topic: &str,
675        handler: Py<PyAny>,
676        priority: u32,
677    ) -> PyResult<()> {
678        let pattern = parse_pattern(topic)?;
679        let handler = make_handler(py, handler)?;
680        msgbus_api::subscribe_any(pattern, handler, Some(priority));
681        Ok(())
682    }
683
684    /// Unsubscribes the given handler from the given topic.
685    #[pyo3(name = "unsubscribe")]
686    fn py_unsubscribe(&self, py: Python<'_>, topic: &str, handler: Py<PyAny>) -> PyResult<()> {
687        let pattern = parse_pattern(topic)?;
688        let handler = make_handler(py, handler)?;
689        msgbus_api::unsubscribe_any(pattern, &handler);
690        Ok(())
691    }
692
693    /// Publishes a message for the given topic.
694    #[pyo3(name = "publish", signature = (topic, msg, external_pub=true))]
695    #[expect(clippy::needless_pass_by_value)]
696    fn py_publish(
697        slf: &Bound<'_, Self>,
698        py: Python<'_>,
699        topic: &str,
700        msg: Py<PyAny>,
701        external_pub: bool,
702    ) -> PyResult<()> {
703        // Reject an enclosing facade borrow before publication has synchronous effects
704        drop(slf.try_borrow_mut()?);
705        let topic_mstr = MStr::<Topic>::topic(topic).map_err(to_pyruntime_err)?;
706
707        let py_msg = PyMessage(msg.clone_ref(py));
708        msgbus_api::publish_any(topic_mstr, &py_msg);
709
710        if external_pub {
711            Self::publish_external(slf, py, topic, &msg)?;
712        }
713
714        slf.try_borrow_mut()?.pub_count += 1;
715        Ok(())
716    }
717
718    /// Disposes of the message bus, clearing all state.
719    #[pyo3(name = "dispose")]
720    fn py_dispose(&mut self, py: Python<'_>) -> PyResult<()> {
721        log::debug!("Closing message bus");
722
723        get_message_bus().borrow_mut().dispose();
724
725        self.correlation_index.clear();
726        self.listeners.clear();
727        self.streaming_types.clear();
728
729        if let Some(ref backing) = self.backing {
730            let db = backing.bind(py);
731            if !db.call_method0("is_closed")?.extract::<bool>()? {
732                db.call_method0("close")?;
733            }
734        }
735
736        log::info!("Closed message bus");
737        Ok(())
738    }
739
740    /// Registers a type for external-to-internal message streaming.
741    #[pyo3(name = "add_streaming_type")]
742    fn py_add_streaming_type(&mut self, cls: Py<PyAny>) {
743        self.streaming_types.push(cls);
744    }
745
746    /// Adds a listener to the message bus.
747    #[pyo3(name = "add_listener")]
748    fn py_add_listener(&mut self, listener: Py<PyAny>) {
749        self.listeners.push(listener);
750    }
751}
752
753impl PyMessageBus {
754    fn publish_external(
755        slf: &Bound<'_, Self>,
756        py: Python<'_>,
757        topic: &str,
758        msg: &Py<PyAny>,
759    ) -> PyResult<()> {
760        let (types_filter, serializer, backing, listeners) = {
761            let bus = slf.try_borrow()?;
762            (
763                bus.types_filter.as_ref().map(|value| value.clone_ref(py)),
764                bus.serializer.as_ref().map(|value| value.clone_ref(py)),
765                bus.backing.as_ref().map(|value| value.clone_ref(py)),
766                bus.listeners
767                    .iter()
768                    .map(|value| value.clone_ref(py))
769                    .collect::<Vec<_>>(),
770            )
771        };
772
773        if let Some(ref filter) = types_filter {
774            let is_excluded = py
775                .import("builtins")?
776                .call_method1("isinstance", (msg, filter))?
777                .extract::<bool>()?;
778
779            if is_excluded {
780                return Ok(());
781            }
782        }
783
784        // Serialize: raw bytes pass through, other types need a serializer
785        let msg_ref = msg.bind(py);
786        let payload: Py<PyAny> = if msg_ref.is_instance_of::<pyo3::types::PyBytes>() {
787            msg.clone_ref(py)
788        } else if let Some(ref serializer) = serializer {
789            serializer.call_method1(py, "serialize", (msg,))?
790        } else {
791            return Ok(());
792        };
793
794        if let Some(ref backing) = backing {
795            let db = backing.bind(py);
796            if !db.call_method0("is_closed")?.extract::<bool>()? {
797                db.call_method1("publish", (topic, &payload))?;
798            }
799        }
800
801        for listener in &listeners {
802            let l = listener.bind(py);
803            if l.call_method0("is_closed")?.extract::<bool>()? {
804                continue;
805            }
806            l.call_method1("publish", (topic, &payload))?;
807        }
808
809        Ok(())
810    }
811}
812
813fn parse_endpoint(endpoint: &str) -> PyResult<MStr<Endpoint>> {
814    MStr::<Endpoint>::endpoint(endpoint).map_err(to_pyvalue_err)
815}
816
817fn parse_pattern(pattern: &str) -> PyResult<MStr<Pattern>> {
818    MStr::<Pattern>::pattern_checked(pattern).map_err(to_pyvalue_err)
819}
820
821/// Owns a component's Python subscriptions without owning the runtime bus.
822#[derive(Debug, Default)]
823pub struct PyMessageBusScope {
824    bus: RefCell<Weak<RefCell<MessageBus>>>,
825    subscriptions: RefCell<Vec<ScopedSubscription>>,
826    clearing: Cell<bool>,
827}
828
829impl PyMessageBusScope {
830    pub(crate) fn register(&self) {
831        *self.bus.borrow_mut() =
832            try_get_message_bus().map_or_else(Weak::new, |bus| Rc::downgrade(&bus));
833    }
834
835    /// Publishes the original Python object synchronously on an application topic.
836    ///
837    /// Handlers receive the same object. Nested publication finishes before the outer call
838    /// returns. Python handler exceptions are logged and do not interrupt other handlers.
839    /// This does not serialize or externally publish the object.
840    ///
841    /// # Errors
842    ///
843    /// Returns an error if the topic is invalid or the registered runtime bus is no longer active.
844    pub fn publish_message(&self, topic: &str, message: Py<PyAny>) -> PyResult<()> {
845        let topic = MStr::<Topic>::topic(topic).map_err(to_pyvalue_err)?;
846        self.active_bus()?;
847        msgbus_api::publish_any(topic, &PyMessage(message));
848        Ok(())
849    }
850
851    /// Subscribes a callable to an application topic pattern.
852    ///
853    /// Higher priorities run first. Repeating a pattern and callable within this component is
854    /// a no-op, even with a different priority. Unsubscribe first to change priority.
855    /// Python-defined bound methods are identified by their receiver and function; other callables
856    /// use object identity. Subscriptions owned by other components remain independent.
857    ///
858    /// # Errors
859    ///
860    /// Returns an error for an invalid pattern, a non-callable handler, or an inactive runtime bus.
861    pub fn subscribe_topic(
862        &self,
863        py: Python<'_>,
864        topic: &str,
865        handler: Py<PyAny>,
866        priority: u32,
867    ) -> PyResult<()> {
868        let pattern = MStr::<Pattern>::pattern_checked(topic).map_err(to_pyvalue_err)?;
869        let identity = CallableIdentity::new(handler.bind(py))?;
870        self.active_bus()?;
871
872        if self
873            .subscriptions
874            .borrow()
875            .iter()
876            .any(|sub| sub.pattern == pattern && sub.identity == identity)
877        {
878            return Ok(());
879        }
880
881        let id = format!("python-component:{}", UUID4::new()).into();
882        let handler = TypedHandler(
883            Rc::new(PyCallableHandler::with_id(id, handler)) as Rc<dyn Handler<dyn Any>>
884        );
885        msgbus_api::subscribe_any(pattern, handler.clone(), Some(priority));
886
887        self.subscriptions.borrow_mut().push(ScopedSubscription {
888            pattern,
889            identity,
890            handler,
891        });
892
893        Ok(())
894    }
895
896    /// Removes this component's exact topic pattern and callable subscription.
897    ///
898    /// An absent subscription is a no-op. A callback already snapshotted by a synchronous
899    /// publication may still run in that publication.
900    ///
901    /// # Errors
902    ///
903    /// Returns an error for an invalid pattern, a non-callable handler, or an inactive runtime bus.
904    pub fn unsubscribe_topic(&self, topic: &str, handler: &Bound<'_, PyAny>) -> PyResult<()> {
905        let pattern = MStr::<Pattern>::pattern_checked(topic).map_err(to_pyvalue_err)?;
906        let identity = CallableIdentity::new(handler)?;
907        let bus = self.active_bus()?;
908
909        let removed = {
910            let mut subscriptions = self.subscriptions.borrow_mut();
911            subscriptions
912                .iter()
913                .position(|sub| sub.pattern == pattern && sub.identity == identity)
914                .map(|index| subscriptions.remove(index))
915        };
916
917        if let Some(sub) = removed {
918            bus.borrow_mut().unsubscribe_any(sub.pattern, &sub.handler);
919        }
920
921        Ok(())
922    }
923
924    pub(crate) fn invalidate(&self) {
925        *self.bus.borrow_mut() = Weak::new();
926    }
927
928    pub(crate) fn clear(&self) {
929        if self.clearing.replace(true) {
930            return;
931        }
932
933        let subscriptions = std::mem::take(&mut *self.subscriptions.borrow_mut());
934
935        if let Some(bus) = self.bus.borrow().upgrade() {
936            let mut bus = bus.borrow_mut();
937            for sub in &subscriptions {
938                bus.unsubscribe_any(sub.pattern, &sub.handler);
939            }
940        }
941
942        // Callable finalizers may re-enter Python; release them after all bookkeeping borrows
943        drop(subscriptions);
944        self.clearing.set(false);
945    }
946
947    fn active_bus(&self) -> PyResult<Rc<RefCell<MessageBus>>> {
948        if self.clearing.get() {
949            return Err(to_pyruntime_err("Component is releasing subscriptions"));
950        }
951
952        let registered =
953            self.bus.borrow().upgrade().ok_or_else(|| {
954                to_pyruntime_err("Component's registered message bus is unavailable")
955            })?;
956        let active = try_get_message_bus()
957            .ok_or_else(|| to_pyruntime_err("No runtime message bus is active on this thread"))?;
958
959        if !Rc::ptr_eq(&registered, &active) {
960            return Err(to_pyruntime_err(
961                "Component's registered message bus has been replaced",
962            ));
963        }
964
965        Ok(registered)
966    }
967}
968
969#[derive(Debug)]
970struct ScopedSubscription {
971    pattern: MStr<Pattern>,
972    identity: CallableIdentity,
973    handler: ShareableMessageHandler,
974}
975
976#[derive(Debug, PartialEq, Eq)]
977struct CallableIdentity {
978    receiver: usize,
979    function: Option<usize>,
980}
981
982impl CallableIdentity {
983    fn new(callable: &Bound<'_, PyAny>) -> PyResult<Self> {
984        if !callable.is_callable() {
985            return Err(to_pytype_err("handler must be callable"));
986        }
987
988        let method_type = callable.py().import("types")?.getattr("MethodType")?;
989        if callable.get_type().is(&method_type) {
990            return Ok(Self {
991                receiver: callable.getattr("__self__")?.as_ptr() as usize,
992                function: Some(callable.getattr("__func__")?.as_ptr() as usize),
993            });
994        }
995
996        Ok(Self {
997            receiver: callable.as_ptr() as usize,
998            function: None,
999        })
1000    }
1001}
1002
1003#[cfg(test)]
1004mod tests {
1005    use std::any::Any;
1006
1007    use nautilus_model::{
1008        data::QuoteTick,
1009        identifiers::{ActorId, ComponentId},
1010        types::{Price, Quantity},
1011    };
1012    use pyo3::{exceptions::PyValueError, ffi::c_str, types::PyDict};
1013    use rstest::rstest;
1014
1015    use super::*;
1016    use crate::{
1017        actor::registry::with_actor_registry,
1018        cache::Cache,
1019        clock::VirtualClock,
1020        component::{release_component_subscriptions, with_component_registry},
1021        msgbus::set_message_bus,
1022        python::{actor::PyDataActor, wrappers::release_python_wrapper},
1023    };
1024
1025    #[rstest]
1026    fn test_publish_checks_facade_borrow_before_delivery() {
1027        Python::initialize();
1028        Python::attach(|py| {
1029            let bus = py
1030                .get_type::<PyMessageBus>()
1031                .call1((TraderId::from("TRADER-001"),))
1032                .unwrap()
1033                .cast_into::<PyMessageBus>()
1034                .unwrap();
1035            let locals = PyDict::new(py);
1036            locals.set_item("bus", &bus).unwrap();
1037            py.run(
1038                c_str!("seen = []\nbus.subscribe('topic', seen.append)"),
1039                Some(&locals),
1040                None,
1041            )
1042            .unwrap();
1043            let borrowed = bus.borrow_mut();
1044            py.run(
1045                c_str!(
1046                    "try:\n    bus.publish('topic', 37, external_pub=False)\nexcept BaseException as e:\n    error = (type(e).__name__, str(e))"
1047                ),
1048                Some(&locals),
1049                None,
1050            )
1051            .unwrap();
1052            drop(borrowed);
1053
1054            let seen: Vec<i64> = locals.get_item("seen").unwrap().unwrap().extract().unwrap();
1055            let error: (String, String) = locals
1056                .get_item("error")
1057                .unwrap()
1058                .unwrap()
1059                .extract()
1060                .unwrap();
1061
1062            assert_eq!(seen, Vec::<i64>::new());
1063            assert_eq!(error, ("RuntimeError".into(), "Already borrowed".into()));
1064            assert_eq!(bus.borrow().pub_count, 0);
1065        });
1066    }
1067
1068    #[rstest]
1069    fn test_message_bus_factory_registry_compatibility_constructors() {
1070        let registry = MessageBusFactoryRegistry::new();
1071        let default_registry = MessageBusFactoryRegistry::default();
1072
1073        assert_eq!(format!("{registry:?}"), format!("{default_registry:?}"));
1074        assert!(format!("{registry:?}").contains("message bus factory"));
1075    }
1076
1077    #[rstest]
1078    fn message_bus_config_py_new_maps_validate_error_to_value_error() {
1079        pyo3::Python::initialize();
1080        Python::attach(|py| {
1081            let err = MessageBusConfig::py_new(
1082                Some(SerializationEncoding::Json),
1083                None,
1084                Some(SerializationEncoding::Capnp),
1085                None,
1086                None,
1087                None,
1088                None,
1089                None,
1090                None,
1091                None,
1092                None,
1093                None,
1094                None,
1095                None,
1096                None,
1097            )
1098            .unwrap_err();
1099
1100            assert!(err.is_instance_of::<PyValueError>(py));
1101            assert_eq!(
1102                err.value(py).to_string(),
1103                format!(
1104                    "MessageBusConfig.encoding_builtin has unsupported value: {} is not supported by AccountState, OrderEventAny, PositionEvent, PortfolioSnapshot",
1105                    SerializationEncoding::Capnp
1106                )
1107            );
1108        });
1109    }
1110
1111    #[rstest]
1112    fn test_py_message_downcast() {
1113        pyo3::Python::initialize();
1114        Python::attach(|py| {
1115            let py_obj = py.eval(c_str!("42"), None, None).unwrap();
1116            let msg = PyMessage(py_obj.unbind());
1117
1118            let any_ref: &dyn Any = &msg;
1119            let downcasted = any_ref.downcast_ref::<PyMessage>();
1120            assert!(downcasted.is_some());
1121
1122            let inner = &downcasted.unwrap().0;
1123            let value: i64 = inner.extract(py).unwrap();
1124            assert_eq!(value, 42);
1125        });
1126    }
1127
1128    #[rstest]
1129    fn test_py_callable_handler_id_stability() {
1130        pyo3::Python::initialize();
1131        Python::attach(|py| {
1132            let callable = py.eval(c_str!("lambda x: x"), None, None).unwrap().unbind();
1133
1134            let handler1 = PyCallableHandler::new(py, callable.clone_ref(py)).unwrap();
1135            let handler2 = PyCallableHandler::new(py, callable).unwrap();
1136
1137            assert_eq!(handler1.id(), handler2.id());
1138        });
1139    }
1140
1141    #[rstest]
1142    fn test_py_callable_handler_dispatch() {
1143        pyo3::Python::initialize();
1144        Python::attach(|py| {
1145            let main = py.import("__main__").unwrap();
1146            let globals = main.dict();
1147            py.run(
1148                c_str!("results = []\ndef handler(x): results.append(x)"),
1149                Some(&globals),
1150                None,
1151            )
1152            .unwrap();
1153
1154            let handler_fn = globals.get_item("handler").unwrap().unwrap().unbind();
1155            let handler = PyCallableHandler::new(py, handler_fn).unwrap();
1156
1157            let py_obj = py.eval(c_str!("'hello'"), None, None).unwrap();
1158            let msg = PyMessage(py_obj.unbind());
1159
1160            let any_ref: &dyn Any = &msg;
1161            handler.handle(any_ref);
1162
1163            let results = globals.get_item("results").unwrap().unwrap();
1164            let len: usize = results.len().unwrap();
1165            assert_eq!(len, 1);
1166        });
1167    }
1168    #[rstest]
1169    fn test_py_message_bus_scope_object_identity_nested_delivery_and_subscription_ownership() {
1170        Python::initialize();
1171        Python::attach(|py| {
1172            set_message_bus(Rc::new(RefCell::new(MessageBus::default())));
1173            let first = registered_actor(py, "MESSAGE-FIRST");
1174            let second = registered_actor(py, "MESSAGE-SECOND");
1175            let locals = PyDict::new(py);
1176            locals.set_item("first", &first).unwrap();
1177            locals.set_item("second", &second).unwrap();
1178            py.run(
1179                c_str!(
1180                    r#"
1181message = {"quantity": 73, "labels": ["alpha", "beta"]}
1182seen = []
1183def high(value):
1184    seen.append(("high", value))
1185    first.publish_message("app.nested", value)
1186    seen.append(("returned", value))
1187def nested(value):
1188    seen.append(("nested", value))
1189def low(value):
1190    seen.append(("low", value))
1191first.subscribe_topic("app.outer", high, 50)
1192first.subscribe_topic("app.outer", high, 100)
1193second.subscribe_topic("app.*", low, 2)
1194second.subscribe_topic("app.nested", nested, 20)
1195first.publish_message("app.outer", message)
1196assert [name for name, _ in seen] == ["high", "nested", "low", "returned", "low"]
1197assert all(value is message for _, value in seen)
1198assert message == {"quantity": 73, "labels": ["alpha", "beta"]}
1199first.unsubscribe_topic("app.outer", high)
1200first.unsubscribe_topic("app.outer", high)
1201seen.clear()
1202first.publish_message("app.outer", message)
1203assert seen == [("low", message)]
1204first.subscribe_topic("app.shared", low)
1205second.subscribe_topic("app.shared", low)
1206first.unsubscribe_topic("app.shared", low)
1207seen.clear()
1208first.publish_message("app.shared", message)
1209assert seen == [("low", message), ("low", message)]
1210for owner, peer in [(first, second), (second, first)]:
1211    owner.unsubscribe_topic("isolated.shared", low)
1212    owner.subscribe_topic("isolated.shared", low)
1213    peer.subscribe_topic("isolated.shared", low)
1214    owner.unsubscribe_topic("isolated.shared", low)
1215    seen.clear()
1216    first.publish_message("isolated.shared", message)
1217    assert seen == [("low", message)]
1218    peer.unsubscribe_topic("isolated.shared", low)
1219    seen.clear()
1220    first.publish_message("isolated.shared", message)
1221    assert seen == []
1222"#
1223                ),
1224                Some(&locals),
1225                None,
1226            )
1227            .unwrap();
1228            first.call_method0("dispose").unwrap();
1229            second.call_method0("dispose").unwrap();
1230            locals.clear();
1231            release_actor("MESSAGE-FIRST");
1232            release_actor("MESSAGE-SECOND");
1233        });
1234    }
1235
1236    #[rstest]
1237    fn test_py_message_bus_scope_callable_identity_and_reference_release() {
1238        Python::initialize();
1239        Python::attach(|py| {
1240            set_message_bus(Rc::new(RefCell::new(MessageBus::default())));
1241            let actor = registered_actor(py, "MESSAGE-IDENTITY");
1242            let locals = PyDict::new(py);
1243            locals.set_item("actor", &actor).unwrap();
1244            py.run(
1245                c_str!(
1246                    r#"
1247import gc
1248import weakref
1249seen = []
1250class Receiver:
1251    def __init__(self, name): self.name = name
1252    def __repr__(self): raise AssertionError("repr must not run")
1253    def __eq__(self, other): raise AssertionError("equality must not run")
1254    def __call__(self, value): seen.append((self.name, value))
1255    def receive(self, value): seen.append((self.name, value))
1256left = Receiver("left")
1257right = Receiver("right")
1258actor.subscribe_topic("app.callables", left, 2)
1259actor.subscribe_topic("app.callables", right, 1)
1260actor.unsubscribe_topic("app.callables", left)
1261actor.subscribe_topic("app.methods", left.receive)
1262actor.subscribe_topic("app.methods", left.receive)
1263actor.unsubscribe_topic("app.methods", left.receive)
1264actor.publish_message("app.methods", 19)
1265actor.publish_message("app.callables", 23)
1266assert seen == [("right", 23)]
1267ref = weakref.ref(right)
1268del right
1269assert ref() is not None
1270actor.dispose()
1271gc.collect()
1272assert ref() is None
1273"#
1274                ),
1275                Some(&locals),
1276                None,
1277            )
1278            .unwrap();
1279            locals.clear();
1280            release_actor("MESSAGE-IDENTITY");
1281        });
1282    }
1283
1284    #[rstest]
1285    fn test_py_message_bus_scope_lifecycle_and_foreign_thread_errors() {
1286        Python::initialize();
1287        Python::attach(|py| {
1288            set_message_bus(Rc::new(RefCell::new(MessageBus::default())));
1289            let actor = registered_actor(py, "MESSAGE-LIFECYCLE");
1290            let locals = PyDict::new(py);
1291            locals.set_item("actor", &actor).unwrap();
1292            locals
1293                .set_item(
1294                    "unregistered",
1295                    py.get_type::<PyDataActor>().call0().unwrap(),
1296                )
1297                .unwrap();
1298            py.run(
1299                c_str!(
1300                    r#"
1301import threading
1302seen = []
1303def handler(value): seen.append(value)
1304def assert_runtime_errors(component):
1305
1306    for name, args in [("publish_message", ("app.events", 9)),
1307                       ("subscribe_topic", ("app.events", handler)),
1308                       ("unsubscribe_topic", ("app.events", handler))]:
1309        try: getattr(component, name)(*args)
1310        except RuntimeError: pass
1311        else: raise AssertionError(name)
1312assert_runtime_errors(unregistered)
1313errors = []
1314def foreign():
1315    try: assert_runtime_errors(actor)
1316    except BaseException as error: errors.append(error)
1317thread = threading.Thread(target=foreign)
1318thread.start()
1319thread.join()
1320assert errors == []
1321actor.on_start = lambda: actor.subscribe_topic("app.events", handler)
1322actor.start()
1323actor.stop()
1324actor.publish_message("app.events", 31)
1325actor.resume()
1326actor.publish_message("app.events", 37)
1327actor.stop()
1328actor.reset()
1329actor.publish_message("app.events", 41)
1330assert seen == [31, 37]
1331actor.subscribe_topic("app.events", handler)
1332actor.dispose()
1333assert_runtime_errors(actor)
1334"#
1335                ),
1336                Some(&locals),
1337                None,
1338            )
1339            .unwrap();
1340            assert_eq!(get_message_bus().borrow().subscriptions().len(), 0);
1341            locals.clear();
1342            release_actor("MESSAGE-LIFECYCLE");
1343        });
1344    }
1345
1346    #[rstest]
1347    fn test_py_message_bus_scope_replacement_rejected_and_original_cleaned() {
1348        Python::initialize();
1349        Python::attach(|py| {
1350            let original = Rc::new(RefCell::new(MessageBus::default()));
1351            set_message_bus(Rc::clone(&original));
1352            let actor = registered_actor(py, "MESSAGE-REPLACED");
1353            let handler = py.eval(c_str!("lambda value: None"), None, None).unwrap();
1354            actor
1355                .call_method1("subscribe_topic", ("app.events", &handler))
1356                .unwrap();
1357            let replacement = Rc::new(RefCell::new(MessageBus::default()));
1358            set_message_bus(Rc::clone(&replacement));
1359            let error = actor
1360                .call_method1("publish_message", ("app.events", 43))
1361                .unwrap_err();
1362            assert!(error.is_instance_of::<pyo3::exceptions::PyRuntimeError>(py));
1363            assert_eq!(original.borrow().subscriptions().len(), 1);
1364            actor.call_method0("dispose").unwrap();
1365            assert_eq!(original.borrow().subscriptions().len(), 0);
1366            assert!(Rc::ptr_eq(&replacement, &get_message_bus()));
1367            release_actor("MESSAGE-REPLACED");
1368        });
1369    }
1370
1371    #[rstest]
1372    fn test_py_message_bus_scope_missing_runtime_does_not_create_one() {
1373        Python::initialize();
1374        Python::attach(|py| {
1375            assert!(try_get_message_bus().is_none());
1376            let actor = registered_actor(py, "MESSAGE-MISSING");
1377            let error = actor
1378                .call_method1("publish_message", ("app.events", 47))
1379                .unwrap_err();
1380            assert!(error.is_instance_of::<pyo3::exceptions::PyRuntimeError>(py));
1381            assert_eq!(
1382                error.value(py).str().unwrap().to_str().unwrap(),
1383                "Component's registered message bus is unavailable"
1384            );
1385            assert!(try_get_message_bus().is_none());
1386            release_actor("MESSAGE-MISSING");
1387        });
1388    }
1389
1390    #[rstest]
1391    fn test_py_message_bus_scope_input_errors_and_existing_routes() {
1392        Python::initialize();
1393        Python::attach(|py| {
1394            let bus = Rc::new(RefCell::new(MessageBus::default()));
1395            set_message_bus(Rc::clone(&bus));
1396            let endpoint_received = Rc::new(RefCell::new(Vec::<i32>::new()));
1397            let received = Rc::clone(&endpoint_received);
1398            let endpoint_handler = ShareableMessageHandler::from_typed(move |value: &i32| {
1399                received.borrow_mut().push(*value);
1400            });
1401            msgbus_api::register_any("app.endpoint".into(), endpoint_handler);
1402            let quotes_received = Rc::new(RefCell::new(Vec::new()));
1403            let received = Rc::clone(&quotes_received);
1404            let quote_handler =
1405                TypedHandler::from(move |quote: &QuoteTick| received.borrow_mut().push(*quote));
1406            msgbus_api::subscribe_quotes("app.quotes".into(), quote_handler.clone(), None);
1407            let actor = registered_actor(py, "MESSAGE-ERRORS");
1408            let locals = PyDict::new(py);
1409            locals.set_item("actor", &actor).unwrap();
1410            py.run(
1411                c_str!(
1412                    r#"
1413def handler(value): pass
1414
1415for name, args, expected in [
1416    ("publish_message", ("", 53), ValueError),
1417    ("publish_message", ("app.*", 59), ValueError),
1418    ("subscribe_topic", ("", handler), ValueError),
1419    ("unsubscribe_topic", ("", handler), ValueError),
1420    ("subscribe_topic", ("app.events", 61), TypeError),
1421    ("unsubscribe_topic", ("app.events", 67), TypeError),
1422    ("subscribe_topic", ("app.events", handler, -1), OverflowError),
1423    ("subscribe_topic", ("app.events", handler, 2**32), OverflowError),
1424]:
1425    try: getattr(actor, name)(*args)
1426    except expected: pass
1427    else: raise AssertionError((name, args))
1428actor.subscribe_topic("app.events", handler)
1429actor.publish_message("app.events", 71)
1430actor.unsubscribe_topic("app.events", handler)
1431actor.dispose()
1432"#
1433                ),
1434                Some(&locals),
1435                None,
1436            )
1437            .unwrap();
1438            msgbus_api::send_any("app.endpoint".into(), &73_i32);
1439            let quote = QuoteTick::new(
1440                "BTCUSDT.BINANCE".into(),
1441                Price::from("101.25"),
1442                Price::from("102.50"),
1443                Quantity::from("3.75"),
1444                Quantity::from("4.50"),
1445                123.into(),
1446                456.into(),
1447            );
1448            msgbus_api::publish_quote("app.quotes".into(), &quote);
1449            assert_eq!(*endpoint_received.borrow(), vec![73]);
1450            assert_eq!(*quotes_received.borrow(), vec![quote]);
1451            assert!(Rc::ptr_eq(&bus, &get_message_bus()));
1452            msgbus_api::unsubscribe_quotes("app.quotes".into(), &quote_handler);
1453            msgbus_api::deregister_any("app.endpoint".into());
1454            locals.clear();
1455            release_actor("MESSAGE-ERRORS");
1456        });
1457    }
1458
1459    #[rstest]
1460    fn test_py_message_bus_scope_failed_disposal_and_finalizer_reentry() {
1461        Python::initialize();
1462        Python::attach(|py| {
1463            set_message_bus(Rc::new(RefCell::new(MessageBus::default())));
1464            let actor = registered_actor(py, "MESSAGE-FINALIZER");
1465            let locals = PyDict::new(py);
1466            locals.set_item("actor", &actor).unwrap();
1467            py.run(
1468                c_str!(
1469                    r#"
1470seen = []
1471finalized = []
1472def handler(value): seen.append(value)
1473def fail(): raise ValueError("disposal refused")
1474actor.subscribe_topic("app.events", handler)
1475actor.on_dispose = fail
1476try: actor.dispose()
1477except RuntimeError as error: assert "disposal refused" in str(error)
1478else: raise AssertionError("disposal must fail")
1479actor.publish_message("app.events", 79)
1480assert seen == [79]
1481class Finalizer:
1482    def __call__(self, value): pass
1483    def __del__(self):
1484        try: actor.subscribe_topic("app.events", handler)
1485        except RuntimeError: finalized.append("rejected")
1486        else: finalized.append("subscribed")
1487actor.subscribe_topic("app.finalize", Finalizer())
1488
1489"#
1490                ),
1491                Some(&locals),
1492                None,
1493            )
1494            .unwrap();
1495            release_component_subscriptions(&"MESSAGE-FINALIZER".into()).unwrap();
1496            assert_eq!(
1497                locals
1498                    .get_item("finalized")
1499                    .unwrap()
1500                    .unwrap()
1501                    .extract::<Vec<String>>()
1502                    .unwrap(),
1503                vec!["rejected"]
1504            );
1505            assert_eq!(get_message_bus().borrow().subscriptions().len(), 0);
1506            locals.clear();
1507            release_actor("MESSAGE-FINALIZER");
1508        });
1509    }
1510
1511    #[rstest]
1512    fn test_py_message_bus_scope_snapshot_delivery_survives_unsubscription_and_error() {
1513        Python::initialize();
1514        Python::attach(|py| {
1515            set_message_bus(Rc::new(RefCell::new(MessageBus::default())));
1516            let actor = registered_actor(py, "MESSAGE-SNAPSHOT");
1517            let locals = PyDict::new(py);
1518            locals.set_item("actor", &actor).unwrap();
1519            py.run(
1520                c_str!(
1521                    r#"
1522seen = []
1523def high(value):
1524    actor.unsubscribe_topic("app.events", low)
1525    raise ValueError("handler failed")
1526def low(value): seen.append(value)
1527actor.subscribe_topic("app.events", high, 2)
1528actor.subscribe_topic("app.events", low, 1)
1529actor.publish_message("app.events", 83)
1530actor.publish_message("app.events", 89)
1531assert seen == [83]
1532actor.dispose()
1533"#
1534                ),
1535                Some(&locals),
1536                None,
1537            )
1538            .unwrap();
1539            locals.clear();
1540            release_actor("MESSAGE-SNAPSHOT");
1541        });
1542    }
1543
1544    fn release_actor(id: &str) {
1545        release_python_wrapper(ComponentId::from(id));
1546        with_actor_registry(|registry| registry.remove(&id.into()));
1547        with_component_registry(|registry| registry.remove(&id.into()));
1548    }
1549
1550    fn registered_actor<'py>(py: Python<'py>, id: &str) -> Bound<'py, PyDataActor> {
1551        let locals = PyDict::new(py);
1552        locals
1553            .set_item("DataActor", py.get_type::<PyDataActor>())
1554            .unwrap();
1555        py.run(
1556            c_str!(
1557                r#"
1558class Actor(DataActor):
1559    def on_start(self): pass
1560    def on_stop(self): pass
1561    def on_resume(self): pass
1562    def on_reset(self): pass
1563    def on_dispose(self): pass
1564"#
1565            ),
1566            Some(&locals),
1567            None,
1568        )
1569        .unwrap();
1570        let actor = locals
1571            .get_item("Actor")
1572            .unwrap()
1573            .unwrap()
1574            .call0()
1575            .unwrap()
1576            .cast_into::<PyDataActor>()
1577            .unwrap();
1578        {
1579            let mut borrowed = actor.borrow_mut();
1580            borrowed.set_actor_id(ActorId::from(id));
1581            borrowed.set_python_instance(actor.as_any()).unwrap();
1582            borrowed
1583                .register(
1584                    TraderId::from("TRADER-001"),
1585                    Rc::new(RefCell::new(VirtualClock::new())),
1586                    Rc::new(RefCell::new(Cache::default())),
1587                )
1588                .unwrap();
1589            borrowed.register_in_global_registries().unwrap();
1590        }
1591        actor
1592    }
1593}
nautilus_common/python/msgbus.rs

nautilus_common/python/
msgbus.rs
