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 if the operation encounters an invalid pattern, a non-callable handler, or
861    /// an inactive runtime bus.
862    pub fn subscribe_topic(
863        &self,
864        py: Python<'_>,
865        topic: &str,
866        handler: Py<PyAny>,
867        priority: u32,
868    ) -> PyResult<()> {
869        let pattern = MStr::<Pattern>::pattern_checked(topic).map_err(to_pyvalue_err)?;
870        let identity = CallableIdentity::new(handler.bind(py))?;
871        self.active_bus()?;
872
873        if self
874            .subscriptions
875            .borrow()
876            .iter()
877            .any(|sub| sub.pattern == pattern && sub.identity == identity)
878        {
879            return Ok(());
880        }
881
882        let id = format!("python-component:{}", UUID4::new()).into();
883        let handler = TypedHandler(
884            Rc::new(PyCallableHandler::with_id(id, handler)) as Rc<dyn Handler<dyn Any>>
885        );
886        msgbus_api::subscribe_any(pattern, handler.clone(), Some(priority));
887
888        self.subscriptions.borrow_mut().push(ScopedSubscription {
889            pattern,
890            identity,
891            handler,
892        });
893
894        Ok(())
895    }
896
897    /// Removes this component's exact topic pattern and callable subscription.
898    ///
899    /// An absent subscription is a no-op. A callback already snapshotted by a synchronous
900    /// publication may still run in that publication.
901    ///
902    /// # Errors
903    ///
904    /// Returns an error if the operation encounters an invalid pattern, a non-callable handler, or
905    /// an inactive runtime bus.
906    pub fn unsubscribe_topic(&self, topic: &str, handler: &Bound<'_, PyAny>) -> PyResult<()> {
907        let pattern = MStr::<Pattern>::pattern_checked(topic).map_err(to_pyvalue_err)?;
908        let identity = CallableIdentity::new(handler)?;
909        let bus = self.active_bus()?;
910
911        let removed = {
912            let mut subscriptions = self.subscriptions.borrow_mut();
913            subscriptions
914                .iter()
915                .position(|sub| sub.pattern == pattern && sub.identity == identity)
916                .map(|index| subscriptions.remove(index))
917        };
918
919        if let Some(sub) = removed {
920            bus.borrow_mut().unsubscribe_any(sub.pattern, &sub.handler);
921        }
922
923        Ok(())
924    }
925
926    pub(crate) fn invalidate(&self) {
927        *self.bus.borrow_mut() = Weak::new();
928    }
929
930    pub(crate) fn clear(&self) {
931        if self.clearing.replace(true) {
932            return;
933        }
934
935        let subscriptions = std::mem::take(&mut *self.subscriptions.borrow_mut());
936
937        if let Some(bus) = self.bus.borrow().upgrade() {
938            let mut bus = bus.borrow_mut();
939            for sub in &subscriptions {
940                bus.unsubscribe_any(sub.pattern, &sub.handler);
941            }
942        }
943
944        // Callable finalizers may re-enter Python; release them after all bookkeeping borrows
945        drop(subscriptions);
946        self.clearing.set(false);
947    }
948
949    fn active_bus(&self) -> PyResult<Rc<RefCell<MessageBus>>> {
950        if self.clearing.get() {
951            return Err(to_pyruntime_err("Component is releasing subscriptions"));
952        }
953
954        let registered =
955            self.bus.borrow().upgrade().ok_or_else(|| {
956                to_pyruntime_err("Component's registered message bus is unavailable")
957            })?;
958        let active = try_get_message_bus()
959            .ok_or_else(|| to_pyruntime_err("No runtime message bus is active on this thread"))?;
960
961        if !Rc::ptr_eq(&registered, &active) {
962            return Err(to_pyruntime_err(
963                "Component's registered message bus has been replaced",
964            ));
965        }
966
967        Ok(registered)
968    }
969}
970
971#[derive(Debug)]
972struct ScopedSubscription {
973    pattern: MStr<Pattern>,
974    identity: CallableIdentity,
975    handler: ShareableMessageHandler,
976}
977
978#[derive(Debug, PartialEq, Eq)]
979struct CallableIdentity {
980    receiver: usize,
981    function: Option<usize>,
982}
983
984impl CallableIdentity {
985    fn new(callable: &Bound<'_, PyAny>) -> PyResult<Self> {
986        if !callable.is_callable() {
987            return Err(to_pytype_err("handler must be callable"));
988        }
989
990        let method_type = callable.py().import("types")?.getattr("MethodType")?;
991        if callable.get_type().is(&method_type) {
992            return Ok(Self {
993                receiver: callable.getattr("__self__")?.as_ptr() as usize,
994                function: Some(callable.getattr("__func__")?.as_ptr() as usize),
995            });
996        }
997
998        Ok(Self {
999            receiver: callable.as_ptr() as usize,
1000            function: None,
1001        })
1002    }
1003}
1004
1005#[cfg(test)]
1006mod tests {
1007    use std::any::Any;
1008
1009    use nautilus_model::{
1010        data::QuoteTick,
1011        identifiers::{ActorId, ComponentId},
1012        types::{Price, Quantity},
1013    };
1014    use pyo3::{exceptions::PyValueError, ffi::c_str, types::PyDict};
1015    use rstest::rstest;
1016
1017    use super::*;
1018    use crate::{
1019        actor::registry::with_actor_registry,
1020        cache::Cache,
1021        clock::VirtualClock,
1022        component::{release_component_subscriptions, with_component_registry},
1023        msgbus::set_message_bus,
1024        python::{actor::PyDataActor, wrappers::release_python_wrapper},
1025    };
1026
1027    #[rstest]
1028    fn test_publish_checks_facade_borrow_before_delivery() {
1029        Python::initialize();
1030        Python::attach(|py| {
1031            let bus = py
1032                .get_type::<PyMessageBus>()
1033                .call1((TraderId::from("TRADER-001"),))
1034                .unwrap()
1035                .cast_into::<PyMessageBus>()
1036                .unwrap();
1037            let locals = PyDict::new(py);
1038            locals.set_item("bus", &bus).unwrap();
1039            py.run(
1040                c_str!("seen = []\nbus.subscribe('topic', seen.append)"),
1041                Some(&locals),
1042                None,
1043            )
1044            .unwrap();
1045            let borrowed = bus.borrow_mut();
1046            py.run(
1047                c_str!(
1048                    "try:\n    bus.publish('topic', 37, external_pub=False)\nexcept BaseException as e:\n    error = (type(e).__name__, str(e))"
1049                ),
1050                Some(&locals),
1051                None,
1052            )
1053            .unwrap();
1054            drop(borrowed);
1055
1056            let seen: Vec<i64> = locals.get_item("seen").unwrap().unwrap().extract().unwrap();
1057            let error: (String, String) = locals
1058                .get_item("error")
1059                .unwrap()
1060                .unwrap()
1061                .extract()
1062                .unwrap();
1063
1064            assert_eq!(seen, Vec::<i64>::new());
1065            assert_eq!(error, ("RuntimeError".into(), "Already borrowed".into()));
1066            assert_eq!(bus.borrow().pub_count, 0);
1067        });
1068    }
1069
1070    #[rstest]
1071    fn test_message_bus_factory_registry_compatibility_constructors() {
1072        let registry = MessageBusFactoryRegistry::new();
1073        let default_registry = MessageBusFactoryRegistry::default();
1074
1075        assert_eq!(format!("{registry:?}"), format!("{default_registry:?}"));
1076        assert!(format!("{registry:?}").contains("message bus factory"));
1077    }
1078
1079    #[rstest]
1080    fn message_bus_config_py_new_maps_validate_error_to_value_error() {
1081        pyo3::Python::initialize();
1082        Python::attach(|py| {
1083            let err = MessageBusConfig::py_new(
1084                Some(SerializationEncoding::Json),
1085                None,
1086                Some(SerializationEncoding::Capnp),
1087                None,
1088                None,
1089                None,
1090                None,
1091                None,
1092                None,
1093                None,
1094                None,
1095                None,
1096                None,
1097                None,
1098                None,
1099            )
1100            .unwrap_err();
1101
1102            assert!(err.is_instance_of::<PyValueError>(py));
1103            assert_eq!(
1104                err.value(py).to_string(),
1105                format!(
1106                    "MessageBusConfig.encoding_builtin has unsupported value: {} is not supported by AccountState, OrderEventAny, PositionEvent, PortfolioSnapshot",
1107                    SerializationEncoding::Capnp
1108                )
1109            );
1110        });
1111    }
1112
1113    #[rstest]
1114    fn test_py_message_downcast() {
1115        pyo3::Python::initialize();
1116        Python::attach(|py| {
1117            let py_obj = py.eval(c_str!("42"), None, None).unwrap();
1118            let msg = PyMessage(py_obj.unbind());
1119
1120            let any_ref: &dyn Any = &msg;
1121            let downcasted = any_ref.downcast_ref::<PyMessage>();
1122            assert!(downcasted.is_some());
1123
1124            let inner = &downcasted.unwrap().0;
1125            let value: i64 = inner.extract(py).unwrap();
1126            assert_eq!(value, 42);
1127        });
1128    }
1129
1130    #[rstest]
1131    fn test_py_callable_handler_id_stability() {
1132        pyo3::Python::initialize();
1133        Python::attach(|py| {
1134            let callable = py.eval(c_str!("lambda x: x"), None, None).unwrap().unbind();
1135
1136            let handler1 = PyCallableHandler::new(py, callable.clone_ref(py)).unwrap();
1137            let handler2 = PyCallableHandler::new(py, callable).unwrap();
1138
1139            assert_eq!(handler1.id(), handler2.id());
1140        });
1141    }
1142
1143    #[rstest]
1144    fn test_py_callable_handler_dispatch() {
1145        pyo3::Python::initialize();
1146        Python::attach(|py| {
1147            let main = py.import("__main__").unwrap();
1148            let globals = main.dict();
1149            py.run(
1150                c_str!("results = []\ndef handler(x): results.append(x)"),
1151                Some(&globals),
1152                None,
1153            )
1154            .unwrap();
1155
1156            let handler_fn = globals.get_item("handler").unwrap().unwrap().unbind();
1157            let handler = PyCallableHandler::new(py, handler_fn).unwrap();
1158
1159            let py_obj = py.eval(c_str!("'hello'"), None, None).unwrap();
1160            let msg = PyMessage(py_obj.unbind());
1161
1162            let any_ref: &dyn Any = &msg;
1163            handler.handle(any_ref);
1164
1165            let results = globals.get_item("results").unwrap().unwrap();
1166            let len: usize = results.len().unwrap();
1167            assert_eq!(len, 1);
1168        });
1169    }
1170    #[rstest]
1171    fn test_py_message_bus_scope_object_identity_nested_delivery_and_subscription_ownership() {
1172        Python::initialize();
1173        Python::attach(|py| {
1174            set_message_bus(Rc::new(RefCell::new(MessageBus::default())));
1175            let first = registered_actor(py, "MESSAGE-FIRST");
1176            let second = registered_actor(py, "MESSAGE-SECOND");
1177            let locals = PyDict::new(py);
1178            locals.set_item("first", &first).unwrap();
1179            locals.set_item("second", &second).unwrap();
1180            py.run(
1181                c_str!(
1182                    r#"
1183message = {"quantity": 73, "labels": ["alpha", "beta"]}
1184seen = []
1185def high(value):
1186    seen.append(("high", value))
1187    first.publish_message("app.nested", value)
1188    seen.append(("returned", value))
1189def nested(value):
1190    seen.append(("nested", value))
1191def low(value):
1192    seen.append(("low", value))
1193first.subscribe_topic("app.outer", high, 50)
1194first.subscribe_topic("app.outer", high, 100)
1195second.subscribe_topic("app.*", low, 2)
1196second.subscribe_topic("app.nested", nested, 20)
1197first.publish_message("app.outer", message)
1198assert [name for name, _ in seen] == ["high", "nested", "low", "returned", "low"]
1199assert all(value is message for _, value in seen)
1200assert message == {"quantity": 73, "labels": ["alpha", "beta"]}
1201first.unsubscribe_topic("app.outer", high)
1202first.unsubscribe_topic("app.outer", high)
1203seen.clear()
1204first.publish_message("app.outer", message)
1205assert seen == [("low", message)]
1206first.subscribe_topic("app.shared", low)
1207second.subscribe_topic("app.shared", low)
1208first.unsubscribe_topic("app.shared", low)
1209seen.clear()
1210first.publish_message("app.shared", message)
1211assert seen == [("low", message), ("low", message)]
1212for owner, peer in [(first, second), (second, first)]:
1213    owner.unsubscribe_topic("isolated.shared", low)
1214    owner.subscribe_topic("isolated.shared", low)
1215    peer.subscribe_topic("isolated.shared", low)
1216    owner.unsubscribe_topic("isolated.shared", low)
1217    seen.clear()
1218    first.publish_message("isolated.shared", message)
1219    assert seen == [("low", message)]
1220    peer.unsubscribe_topic("isolated.shared", low)
1221    seen.clear()
1222    first.publish_message("isolated.shared", message)
1223    assert seen == []
1224"#
1225                ),
1226                Some(&locals),
1227                None,
1228            )
1229            .unwrap();
1230            first.call_method0("dispose").unwrap();
1231            second.call_method0("dispose").unwrap();
1232            locals.clear();
1233            release_actor("MESSAGE-FIRST");
1234            release_actor("MESSAGE-SECOND");
1235        });
1236    }
1237
1238    #[rstest]
1239    fn test_py_message_bus_scope_callable_identity_and_reference_release() {
1240        Python::initialize();
1241        Python::attach(|py| {
1242            set_message_bus(Rc::new(RefCell::new(MessageBus::default())));
1243            let actor = registered_actor(py, "MESSAGE-IDENTITY");
1244            let locals = PyDict::new(py);
1245            locals.set_item("actor", &actor).unwrap();
1246            py.run(
1247                c_str!(
1248                    r#"
1249import gc
1250import weakref
1251seen = []
1252class Receiver:
1253    def __init__(self, name): self.name = name
1254    def __repr__(self): raise AssertionError("repr must not run")
1255    def __eq__(self, other): raise AssertionError("equality must not run")
1256    def __call__(self, value): seen.append((self.name, value))
1257    def receive(self, value): seen.append((self.name, value))
1258left = Receiver("left")
1259right = Receiver("right")
1260actor.subscribe_topic("app.callables", left, 2)
1261actor.subscribe_topic("app.callables", right, 1)
1262actor.unsubscribe_topic("app.callables", left)
1263actor.subscribe_topic("app.methods", left.receive)
1264actor.subscribe_topic("app.methods", left.receive)
1265actor.unsubscribe_topic("app.methods", left.receive)
1266actor.publish_message("app.methods", 19)
1267actor.publish_message("app.callables", 23)
1268assert seen == [("right", 23)]
1269ref = weakref.ref(right)
1270del right
1271assert ref() is not None
1272actor.dispose()
1273gc.collect()
1274assert ref() is None
1275"#
1276                ),
1277                Some(&locals),
1278                None,
1279            )
1280            .unwrap();
1281            locals.clear();
1282            release_actor("MESSAGE-IDENTITY");
1283        });
1284    }
1285
1286    #[rstest]
1287    fn test_py_message_bus_scope_lifecycle_and_foreign_thread_errors() {
1288        Python::initialize();
1289        Python::attach(|py| {
1290            set_message_bus(Rc::new(RefCell::new(MessageBus::default())));
1291            let actor = registered_actor(py, "MESSAGE-LIFECYCLE");
1292            let locals = PyDict::new(py);
1293            locals.set_item("actor", &actor).unwrap();
1294            locals
1295                .set_item(
1296                    "unregistered",
1297                    py.get_type::<PyDataActor>().call0().unwrap(),
1298                )
1299                .unwrap();
1300            py.run(
1301                c_str!(
1302                    r#"
1303import threading
1304seen = []
1305def handler(value): seen.append(value)
1306def assert_runtime_errors(component):
1307
1308    for name, args in [("publish_message", ("app.events", 9)),
1309                       ("subscribe_topic", ("app.events", handler)),
1310                       ("unsubscribe_topic", ("app.events", handler))]:
1311        try: getattr(component, name)(*args)
1312        except RuntimeError: pass
1313        else: raise AssertionError(name)
1314assert_runtime_errors(unregistered)
1315errors = []
1316def foreign():
1317    try: assert_runtime_errors(actor)
1318    except BaseException as error: errors.append(error)
1319thread = threading.Thread(target=foreign)
1320thread.start()
1321thread.join()
1322assert errors == []
1323actor.on_start = lambda: actor.subscribe_topic("app.events", handler)
1324actor.start()
1325actor.stop()
1326actor.publish_message("app.events", 31)
1327actor.resume()
1328actor.publish_message("app.events", 37)
1329actor.stop()
1330actor.reset()
1331actor.publish_message("app.events", 41)
1332assert seen == [31, 37]
1333actor.subscribe_topic("app.events", handler)
1334actor.dispose()
1335assert_runtime_errors(actor)
1336"#
1337                ),
1338                Some(&locals),
1339                None,
1340            )
1341            .unwrap();
1342            assert_eq!(get_message_bus().borrow().subscriptions().len(), 0);
1343            locals.clear();
1344            release_actor("MESSAGE-LIFECYCLE");
1345        });
1346    }
1347
1348    #[rstest]
1349    fn test_py_message_bus_scope_replacement_rejected_and_original_cleaned() {
1350        Python::initialize();
1351        Python::attach(|py| {
1352            let original = Rc::new(RefCell::new(MessageBus::default()));
1353            set_message_bus(Rc::clone(&original));
1354            let actor = registered_actor(py, "MESSAGE-REPLACED");
1355            let handler = py.eval(c_str!("lambda value: None"), None, None).unwrap();
1356            actor
1357                .call_method1("subscribe_topic", ("app.events", &handler))
1358                .unwrap();
1359            let replacement = Rc::new(RefCell::new(MessageBus::default()));
1360            set_message_bus(Rc::clone(&replacement));
1361            let error = actor
1362                .call_method1("publish_message", ("app.events", 43))
1363                .unwrap_err();
1364            assert!(error.is_instance_of::<pyo3::exceptions::PyRuntimeError>(py));
1365            assert_eq!(original.borrow().subscriptions().len(), 1);
1366            actor.call_method0("dispose").unwrap();
1367            assert_eq!(original.borrow().subscriptions().len(), 0);
1368            assert!(Rc::ptr_eq(&replacement, &get_message_bus()));
1369            release_actor("MESSAGE-REPLACED");
1370        });
1371    }
1372
1373    #[rstest]
1374    fn test_py_message_bus_scope_missing_runtime_does_not_create_one() {
1375        Python::initialize();
1376        Python::attach(|py| {
1377            assert!(try_get_message_bus().is_none());
1378            let actor = registered_actor(py, "MESSAGE-MISSING");
1379            let error = actor
1380                .call_method1("publish_message", ("app.events", 47))
1381                .unwrap_err();
1382            assert!(error.is_instance_of::<pyo3::exceptions::PyRuntimeError>(py));
1383            assert_eq!(
1384                error.value(py).str().unwrap().to_str().unwrap(),
1385                "Component's registered message bus is unavailable"
1386            );
1387            assert!(try_get_message_bus().is_none());
1388            release_actor("MESSAGE-MISSING");
1389        });
1390    }
1391
1392    #[rstest]
1393    fn test_py_message_bus_scope_input_errors_and_existing_routes() {
1394        Python::initialize();
1395        Python::attach(|py| {
1396            let bus = Rc::new(RefCell::new(MessageBus::default()));
1397            set_message_bus(Rc::clone(&bus));
1398            let endpoint_received = Rc::new(RefCell::new(Vec::<i32>::new()));
1399            let received = Rc::clone(&endpoint_received);
1400            let endpoint_handler = ShareableMessageHandler::from_typed(move |value: &i32| {
1401                received.borrow_mut().push(*value);
1402            });
1403            msgbus_api::register_any("app.endpoint".into(), endpoint_handler);
1404            let quotes_received = Rc::new(RefCell::new(Vec::new()));
1405            let received = Rc::clone(&quotes_received);
1406            let quote_handler =
1407                TypedHandler::from(move |quote: &QuoteTick| received.borrow_mut().push(*quote));
1408            msgbus_api::subscribe_quotes("app.quotes".into(), quote_handler.clone(), None);
1409            let actor = registered_actor(py, "MESSAGE-ERRORS");
1410            let locals = PyDict::new(py);
1411            locals.set_item("actor", &actor).unwrap();
1412            py.run(
1413                c_str!(
1414                    r#"
1415def handler(value): pass
1416
1417for name, args, expected in [
1418    ("publish_message", ("", 53), ValueError),
1419    ("publish_message", ("app.*", 59), ValueError),
1420    ("subscribe_topic", ("", handler), ValueError),
1421    ("unsubscribe_topic", ("", handler), ValueError),
1422    ("subscribe_topic", ("app.events", 61), TypeError),
1423    ("unsubscribe_topic", ("app.events", 67), TypeError),
1424    ("subscribe_topic", ("app.events", handler, -1), OverflowError),
1425    ("subscribe_topic", ("app.events", handler, 2**32), OverflowError),
1426]:
1427    try: getattr(actor, name)(*args)
1428    except expected: pass
1429    else: raise AssertionError((name, args))
1430actor.subscribe_topic("app.events", handler)
1431actor.publish_message("app.events", 71)
1432actor.unsubscribe_topic("app.events", handler)
1433actor.dispose()
1434"#
1435                ),
1436                Some(&locals),
1437                None,
1438            )
1439            .unwrap();
1440            msgbus_api::send_any("app.endpoint".into(), &73_i32);
1441            let quote = QuoteTick::new(
1442                "BTCUSDT.BINANCE".into(),
1443                Price::from("101.25"),
1444                Price::from("102.50"),
1445                Quantity::from("3.75"),
1446                Quantity::from("4.50"),
1447                123.into(),
1448                456.into(),
1449            );
1450            msgbus_api::publish_quote("app.quotes".into(), &quote);
1451            assert_eq!(*endpoint_received.borrow(), vec![73]);
1452            assert_eq!(*quotes_received.borrow(), vec![quote]);
1453            assert!(Rc::ptr_eq(&bus, &get_message_bus()));
1454            msgbus_api::unsubscribe_quotes("app.quotes".into(), &quote_handler);
1455            msgbus_api::deregister_any("app.endpoint".into());
1456            locals.clear();
1457            release_actor("MESSAGE-ERRORS");
1458        });
1459    }
1460
1461    #[rstest]
1462    fn test_py_message_bus_scope_failed_disposal_and_finalizer_reentry() {
1463        Python::initialize();
1464        Python::attach(|py| {
1465            set_message_bus(Rc::new(RefCell::new(MessageBus::default())));
1466            let actor = registered_actor(py, "MESSAGE-FINALIZER");
1467            let locals = PyDict::new(py);
1468            locals.set_item("actor", &actor).unwrap();
1469            py.run(
1470                c_str!(
1471                    r#"
1472seen = []
1473finalized = []
1474def handler(value): seen.append(value)
1475def fail(): raise ValueError("disposal refused")
1476actor.subscribe_topic("app.events", handler)
1477actor.on_dispose = fail
1478try: actor.dispose()
1479except RuntimeError as error: assert "disposal refused" in str(error)
1480else: raise AssertionError("disposal must fail")
1481actor.publish_message("app.events", 79)
1482assert seen == [79]
1483class Finalizer:
1484    def __call__(self, value): pass
1485    def __del__(self):
1486        try: actor.subscribe_topic("app.events", handler)
1487        except RuntimeError: finalized.append("rejected")
1488        else: finalized.append("subscribed")
1489actor.subscribe_topic("app.finalize", Finalizer())
1490
1491"#
1492                ),
1493                Some(&locals),
1494                None,
1495            )
1496            .unwrap();
1497            release_component_subscriptions(&"MESSAGE-FINALIZER".into()).unwrap();
1498            assert_eq!(
1499                locals
1500                    .get_item("finalized")
1501                    .unwrap()
1502                    .unwrap()
1503                    .extract::<Vec<String>>()
1504                    .unwrap(),
1505                vec!["rejected"]
1506            );
1507            assert_eq!(get_message_bus().borrow().subscriptions().len(), 0);
1508            locals.clear();
1509            release_actor("MESSAGE-FINALIZER");
1510        });
1511    }
1512
1513    #[rstest]
1514    fn test_py_message_bus_scope_snapshot_delivery_survives_unsubscription_and_error() {
1515        Python::initialize();
1516        Python::attach(|py| {
1517            set_message_bus(Rc::new(RefCell::new(MessageBus::default())));
1518            let actor = registered_actor(py, "MESSAGE-SNAPSHOT");
1519            let locals = PyDict::new(py);
1520            locals.set_item("actor", &actor).unwrap();
1521            py.run(
1522                c_str!(
1523                    r#"
1524seen = []
1525def high(value):
1526    actor.unsubscribe_topic("app.events", low)
1527    raise ValueError("handler failed")
1528def low(value): seen.append(value)
1529actor.subscribe_topic("app.events", high, 2)
1530actor.subscribe_topic("app.events", low, 1)
1531actor.publish_message("app.events", 83)
1532actor.publish_message("app.events", 89)
1533assert seen == [83]
1534actor.dispose()
1535"#
1536                ),
1537                Some(&locals),
1538                None,
1539            )
1540            .unwrap();
1541            locals.clear();
1542            release_actor("MESSAGE-SNAPSHOT");
1543        });
1544    }
1545
1546    fn release_actor(id: &str) {
1547        release_python_wrapper(ComponentId::from(id));
1548        with_actor_registry(|registry| registry.remove(&id.into()));
1549        with_component_registry(|registry| registry.remove(&id.into()));
1550    }
1551
1552    fn registered_actor<'py>(py: Python<'py>, id: &str) -> Bound<'py, PyDataActor> {
1553        let locals = PyDict::new(py);
1554        locals
1555            .set_item("DataActor", py.get_type::<PyDataActor>())
1556            .unwrap();
1557        py.run(
1558            c_str!(
1559                r#"
1560class Actor(DataActor):
1561    def on_start(self): pass
1562    def on_stop(self): pass
1563    def on_resume(self): pass
1564    def on_reset(self): pass
1565    def on_dispose(self): pass
1566"#
1567            ),
1568            Some(&locals),
1569            None,
1570        )
1571        .unwrap();
1572        let actor = locals
1573            .get_item("Actor")
1574            .unwrap()
1575            .unwrap()
1576            .call0()
1577            .unwrap()
1578            .cast_into::<PyDataActor>()
1579            .unwrap();
1580        {
1581            let mut borrowed = actor.borrow_mut();
1582            borrowed.set_actor_id(ActorId::from(id));
1583            borrowed.set_python_instance(actor.as_any()).unwrap();
1584            borrowed
1585                .register(
1586                    TraderId::from("TRADER-001"),
1587                    Rc::new(RefCell::new(VirtualClock::new())),
1588                    Rc::new(RefCell::new(Cache::default())),
1589                )
1590                .unwrap();
1591            borrowed.register_in_global_registries().unwrap();
1592        }
1593        actor
1594    }
1595}
nautilus_common/python/msgbus.rs

nautilus_common/python/
msgbus.rs
