nautilus_common/msgbus/external/
mod.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! External message bus stream encoding and republishing.
17
18use std::{any::Any, cell::Cell};
19
20use anyhow::Context;
21use nautilus_model::{
22    data::{CustomData, Data, deserialize_custom_from_json},
23    reports::{ExecutionMassStatus, FillReport, OrderStatusReport, PositionStatusReport},
24};
25use serde::de::DeserializeOwned;
26use ustr::Ustr;
27
28pub(crate) mod codec;
29
30use self::codec::PayloadCodecError;
31use super::{
32    BusMessage, BusPayloadType, HAS_EXTERNAL_EGRESS, SUPPRESS_EXTERNAL_DEPTH,
33    SuppressExternalGuard,
34    api::{
35        publish_account_state, publish_any, publish_bar, publish_deltas, publish_depth,
36        publish_funding_rate, publish_index_price, publish_instrument, publish_mark_price,
37        publish_option_greeks, publish_order_event, publish_portfolio_snapshot,
38        publish_position_event, publish_quote, publish_trade,
39    },
40    get_message_bus,
41    mstr::{MStr, Topic},
42};
43use crate::{
44    enums::SerializationEncoding,
45    messages::{
46        data::{SubscribeCommand, UnsubscribeCommand},
47        execution::{GenerateExecutionMassStatus, TradingCommand},
48    },
49};
50
51#[inline(always)]
52pub(super) fn forward_to_external_egress<T>(
53    topic: MStr<Topic>,
54    payload_type: BusPayloadType,
55    message: &T,
56) where
57    T: serde::Serialize + Any,
58{
59    if !HAS_EXTERNAL_EGRESS.with(Cell::get) {
60        return;
61    }
62
63    forward_external_message(topic, payload_type, message);
64}
65
66#[inline]
67pub(super) fn forward_any_to_external_egress(topic: MStr<Topic>, message: &dyn Any) {
68    if !HAS_EXTERNAL_EGRESS.with(Cell::get) {
69        return;
70    }
71
72    if let Some(custom) = message.downcast_ref::<CustomData>() {
73        forward_external_message(
74            topic,
75            BusPayloadType::Custom(Ustr::from(custom.data.type_name())),
76            custom,
77        );
78        return;
79    }
80
81    if forward_downcast::<SubscribeCommand>(topic, BusPayloadType::SubscribeCommand, message)
82        || forward_downcast::<UnsubscribeCommand>(
83            topic,
84            BusPayloadType::UnsubscribeCommand,
85            message,
86        )
87        || forward_downcast::<TradingCommand>(topic, BusPayloadType::TradingCommand, message)
88        || forward_downcast::<GenerateExecutionMassStatus>(
89            topic,
90            BusPayloadType::GenerateExecutionMassStatus,
91            message,
92        )
93        || forward_downcast::<OrderStatusReport>(topic, BusPayloadType::OrderStatusReport, message)
94        || forward_downcast::<FillReport>(topic, BusPayloadType::FillReport, message)
95        || forward_downcast::<PositionStatusReport>(
96            topic,
97            BusPayloadType::PositionStatusReport,
98            message,
99        )
100    {
101        return;
102    }
103
104    forward_downcast::<ExecutionMassStatus>(topic, BusPayloadType::ExecutionMassStatus, message);
105}
106
107fn forward_downcast<T>(topic: MStr<Topic>, payload_type: BusPayloadType, message: &dyn Any) -> bool
108where
109    T: serde::Serialize + Any,
110{
111    let Some(message) = message.downcast_ref::<T>() else {
112        return false;
113    };
114
115    forward_external_message(topic, payload_type, message);
116    true
117}
118
119#[cold]
120#[inline(never)]
121fn forward_external_message<T>(topic: MStr<Topic>, payload_type: BusPayloadType, message: &T)
122where
123    T: serde::Serialize + Any,
124{
125    if SUPPRESS_EXTERNAL_DEPTH.with(Cell::get) > 0 {
126        return;
127    }
128
129    let (external_egress, encoding) = {
130        let bus_rc = get_message_bus();
131        let bus = bus_rc.borrow();
132
133        let Some(external_egress) = bus.external_egress() else {
134            return;
135        };
136
137        if payload_type.is_typed_message() && !bus.has_external_streams() {
138            return;
139        }
140
141        if bus.types_filter().contains(&payload_type) {
142            return;
143        }
144
145        (external_egress, bus.encoding_for(payload_type))
146    };
147
148    if external_egress.borrow().is_closed() {
149        return;
150    }
151
152    let payload = match codec::serialize_payload(encoding, payload_type, message) {
153        Ok(payload) => payload,
154        Err(PayloadCodecError::Dropped(e)) => {
155            log::debug!("{e}");
156            return;
157        }
158        Err(PayloadCodecError::Failed(e)) => {
159            log::error!("{e}");
160            return;
161        }
162    };
163
164    let external_egress = external_egress.borrow();
165    if external_egress.is_closed() {
166        return;
167    }
168
169    // Serialization can close the original destination or replace it on the bus
170    external_egress.publish(BusMessage::new(*topic, payload_type, payload, encoding));
171}
172
173/// Decodes an externally-received [`BusMessage`] and republishes it onto the internal bus.
174///
175/// The message `payload_type` header selects the concrete type and the message `encoding` selects
176/// the wire codec, so the message is decoded with the producer's encoding rather than the local
177/// configuration. Republishing runs under a [`SuppressExternalGuard`] so the message is not
178/// forwarded straight back out through external egress, which would create an echo loop on a node
179/// that has both external ingress and egress.
180///
181/// # Errors
182///
183/// Returns an error if the topic is invalid or a supported payload cannot be decoded. Unsupported
184/// type/encoding pairs are skipped with a warning.
185pub fn republish_external_message(message: &BusMessage) -> anyhow::Result<()> {
186    let topic =
187        MStr::<Topic>::topic_from_ustr(message.topic).context("invalid external message topic")?;
188
189    if !is_registered_streaming_type(message) {
190        return Ok(());
191    }
192
193    let _guard = SuppressExternalGuard::new();
194
195    match message.payload_type {
196        BusPayloadType::Custom(_) => {
197            handle_custom_data(
198                topic,
199                message.payload_type,
200                message.encoding,
201                &message.payload,
202            )?;
203        }
204        BusPayloadType::Instrument => {
205            handle_json_msgpack(
206                topic,
207                message.payload_type,
208                message.encoding,
209                &message.payload,
210                publish_instrument,
211            )?;
212        }
213        BusPayloadType::OrderBookDeltas => handle_market_data(
214            topic,
215            message.encoding,
216            &message.payload,
217            codec::deserialize_order_book_deltas,
218            publish_deltas,
219        )?,
220        BusPayloadType::OrderBookDepth => handle_market_data(
221            topic,
222            message.encoding,
223            &message.payload,
224            codec::deserialize_order_book_depth,
225            publish_depth,
226        )?,
227        BusPayloadType::QuoteTick => handle_market_data(
228            topic,
229            message.encoding,
230            &message.payload,
231            codec::deserialize_quote,
232            publish_quote,
233        )?,
234        BusPayloadType::TradeTick => handle_market_data(
235            topic,
236            message.encoding,
237            &message.payload,
238            codec::deserialize_trade,
239            publish_trade,
240        )?,
241        BusPayloadType::Bar => handle_market_data(
242            topic,
243            message.encoding,
244            &message.payload,
245            codec::deserialize_bar,
246            publish_bar,
247        )?,
248        BusPayloadType::MarkPriceUpdate => handle_market_data(
249            topic,
250            message.encoding,
251            &message.payload,
252            codec::deserialize_mark_price,
253            publish_mark_price,
254        )?,
255        BusPayloadType::IndexPriceUpdate => handle_market_data(
256            topic,
257            message.encoding,
258            &message.payload,
259            codec::deserialize_index_price,
260            publish_index_price,
261        )?,
262        BusPayloadType::FundingRateUpdate => handle_market_data(
263            topic,
264            message.encoding,
265            &message.payload,
266            codec::deserialize_funding_rate,
267            publish_funding_rate,
268        )?,
269        BusPayloadType::OptionGreeks => {
270            handle_market_data(
271                topic,
272                message.encoding,
273                &message.payload,
274                codec::deserialize_option_greeks,
275                publish_option_greeks,
276            )?;
277        }
278        BusPayloadType::AccountState => {
279            handle_json_msgpack(
280                topic,
281                message.payload_type,
282                message.encoding,
283                &message.payload,
284                publish_account_state,
285            )?;
286        }
287        BusPayloadType::OrderEvent => {
288            handle_json_msgpack(
289                topic,
290                message.payload_type,
291                message.encoding,
292                &message.payload,
293                publish_order_event,
294            )?;
295        }
296        BusPayloadType::PositionEvent => {
297            handle_json_msgpack(
298                topic,
299                message.payload_type,
300                message.encoding,
301                &message.payload,
302                publish_position_event,
303            )?;
304        }
305        BusPayloadType::PortfolioSnapshot => {
306            handle_json_msgpack(
307                topic,
308                message.payload_type,
309                message.encoding,
310                &message.payload,
311                publish_portfolio_snapshot,
312            )?;
313        }
314        BusPayloadType::SubscribeCommand => {
315            handle_json_msgpack_any::<SubscribeCommand>(topic, message)?;
316        }
317        BusPayloadType::UnsubscribeCommand => {
318            handle_json_msgpack_any::<UnsubscribeCommand>(topic, message)?;
319        }
320        BusPayloadType::TradingCommand => {
321            handle_json_msgpack_any::<TradingCommand>(topic, message)?;
322        }
323        BusPayloadType::GenerateExecutionMassStatus => {
324            handle_json_msgpack_any::<GenerateExecutionMassStatus>(topic, message)?;
325        }
326        BusPayloadType::OrderStatusReport => {
327            handle_json_msgpack_any::<OrderStatusReport>(topic, message)?;
328        }
329        BusPayloadType::FillReport => {
330            handle_json_msgpack_any::<FillReport>(topic, message)?;
331        }
332        BusPayloadType::PositionStatusReport => {
333            handle_json_msgpack_any::<PositionStatusReport>(topic, message)?;
334        }
335        BusPayloadType::ExecutionMassStatus => {
336            handle_json_msgpack_any::<ExecutionMassStatus>(topic, message)?;
337        }
338        #[cfg(feature = "defi")]
339        BusPayloadType::Block
340        | BusPayloadType::Pool
341        | BusPayloadType::PoolLiquidityUpdate
342        | BusPayloadType::PoolFeeCollect
343        | BusPayloadType::PoolFlash => {
344            crate::defi::msgbus::republish_external_message(
345                topic,
346                message.payload_type,
347                message.encoding,
348                &message.payload,
349            )?;
350        }
351    }
352
353    Ok(())
354}
355
356/// Decodes a supported typed payload and passes it to `processor` before applying normal inbound
357/// republishing rules.
358///
359/// JSON or MessagePack payloads identified by [`BusPayloadType::is_typed_message`] reach
360/// `processor` regardless of internal streaming registration. Internal republishing still requires
361/// registration. Unsupported type/encoding pairs are skipped with a warning. The processor mapping
362/// includes a `payload_type` field containing the [`BusPayloadType`] name. External egress is
363/// suppressed for the duration of processing, including processor callbacks. Other payloads bypass
364/// the processor and follow normal inbound republishing. A processor error stops processing and
365/// skips internal republishing.
366///
367/// # Errors
368///
369/// Returns an error if:
370/// - The topic is invalid.
371/// - The typed payload cannot be decoded or mapped to an object.
372/// - The mapping already contains the reserved `payload_type` field.
373/// - The processor fails.
374/// - Normal inbound republishing fails.
375pub fn process_external_typed_message(
376    message: &BusMessage,
377    processor: &mut dyn FnMut(&dyn Any, &serde_json::Value) -> anyhow::Result<()>,
378) -> anyhow::Result<()> {
379    let topic =
380        MStr::<Topic>::topic_from_ustr(message.topic).context("invalid external message topic")?;
381    let _guard = SuppressExternalGuard::new();
382
383    #[rustfmt::skip]
384    let result = match message.payload_type {
385        BusPayloadType::SubscribeCommand => process_typed_payload::<SubscribeCommand>(topic, message, processor),
386        BusPayloadType::UnsubscribeCommand => process_typed_payload::<UnsubscribeCommand>(topic, message, processor),
387        BusPayloadType::TradingCommand => process_typed_payload::<TradingCommand>(topic, message, processor),
388        BusPayloadType::GenerateExecutionMassStatus => process_typed_payload::<GenerateExecutionMassStatus>(topic, message, processor),
389        BusPayloadType::OrderStatusReport => process_typed_payload::<OrderStatusReport>(topic, message, processor),
390        BusPayloadType::FillReport => process_typed_payload::<FillReport>(topic, message, processor),
391        BusPayloadType::PositionStatusReport => process_typed_payload::<PositionStatusReport>(topic, message, processor),
392        BusPayloadType::ExecutionMassStatus => process_typed_payload::<ExecutionMassStatus>(topic, message, processor),
393        _ => republish_external_message(message),
394    };
395    result
396}
397
398fn handle_json_msgpack_any<T>(topic: MStr<Topic>, message: &BusMessage) -> anyhow::Result<()>
399where
400    T: DeserializeOwned + Any,
401{
402    handle_json_msgpack(
403        topic,
404        message.payload_type,
405        message.encoding,
406        &message.payload,
407        |topic, value: &T| publish_any(topic, value),
408    )
409}
410
411fn process_typed_payload<T>(
412    topic: MStr<Topic>,
413    message: &BusMessage,
414    processor: &mut dyn FnMut(&dyn Any, &serde_json::Value) -> anyhow::Result<()>,
415) -> anyhow::Result<()>
416where
417    T: DeserializeOwned + serde::Serialize + Any,
418{
419    let Some(value) = codec::deserialize_json_msgpack_payload::<T>(
420        message.payload_type,
421        message.encoding,
422        &message.payload,
423    )?
424    else {
425        return Ok(());
426    };
427    let mut mapping = serde_json::to_value(&value).with_context(|| {
428        format!(
429            "failed to map decoded {} stream payload",
430            message.payload_type
431        )
432    })?;
433    let mapping_object = mapping.as_object_mut().with_context(|| {
434        format!(
435            "decoded {} stream payload did not map to an object",
436            message.payload_type
437        )
438    })?;
439    anyhow::ensure!(
440        !mapping_object.contains_key("payload_type"),
441        "decoded {} stream payload contains reserved payload_type field",
442        message.payload_type
443    );
444    mapping_object.insert(
445        "payload_type".to_string(),
446        serde_json::Value::String(message.payload_type.as_str().to_string()),
447    );
448
449    processor(&value, &mapping)?;
450    if is_registered_streaming_type(message) {
451        publish_any(topic, &value);
452    }
453    Ok(())
454}
455
456fn is_registered_streaming_type(message: &BusMessage) -> bool {
457    if get_message_bus()
458        .borrow()
459        .is_streaming_type(message.payload_type)
460    {
461        return true;
462    }
463
464    let type_name = message.payload_type.as_str();
465    if type_name.is_empty() {
466        log::debug!(
467            "Skipping external message on topic '{}' with no payload type for inbound republishing",
468            message.topic
469        );
470    } else {
471        log::debug!(
472            "Skipping external {type_name} message on topic '{}' because the type is not registered for streaming",
473            message.topic
474        );
475    }
476
477    false
478}
479
480pub(crate) fn handle_json_msgpack<T>(
481    topic: MStr<Topic>,
482    payload_type: BusPayloadType,
483    encoding: SerializationEncoding,
484    payload: &[u8],
485    publish: impl FnOnce(MStr<Topic>, &T),
486) -> anyhow::Result<()>
487where
488    T: DeserializeOwned,
489{
490    let Some(value) = codec::deserialize_json_msgpack_payload(payload_type, encoding, payload)?
491    else {
492        return Ok(());
493    };
494
495    publish(topic, &value);
496    Ok(())
497}
498
499fn handle_market_data<T>(
500    topic: MStr<Topic>,
501    encoding: SerializationEncoding,
502    payload: &[u8],
503    deserialize: fn(SerializationEncoding, &[u8]) -> anyhow::Result<Option<T>>,
504    publish: impl FnOnce(MStr<Topic>, &T),
505) -> anyhow::Result<()> {
506    let Some(value) = deserialize(encoding, payload)? else {
507        return Ok(());
508    };
509
510    publish(topic, &value);
511    Ok(())
512}
513
514fn handle_custom_data(
515    topic: MStr<Topic>,
516    payload_type: BusPayloadType,
517    encoding: SerializationEncoding,
518    payload: &[u8],
519) -> anyhow::Result<()> {
520    let Some(custom) = decode_custom_data_payload(payload_type, encoding, payload)? else {
521        return Ok(());
522    };
523
524    publish_any(topic, &custom);
525    Ok(())
526}
527
528fn decode_custom_data_payload(
529    payload_type: BusPayloadType,
530    encoding: SerializationEncoding,
531    payload: &[u8],
532) -> anyhow::Result<Option<CustomData>> {
533    let BusPayloadType::Custom(custom_type_name) = payload_type else {
534        unreachable!("custom data payload decoding requires a custom payload type");
535    };
536
537    if custom_type_name.is_empty() {
538        log::warn!("External payload has no type for inbound republishing");
539        return Ok(None);
540    } else if !payload_type.supports(encoding) {
541        codec::warn_unsupported_inbound(payload_type, encoding);
542        return Ok(None);
543    }
544
545    match encoding {
546        SerializationEncoding::Json => {
547            let value =
548                codec::deserialize_json_payload::<serde_json::Value>(payload, "CustomData")?;
549            decode_custom_data_value(custom_type_name, &value)
550                .context("failed to decode JSON CustomData")
551        }
552        SerializationEncoding::MsgPack => {
553            let value =
554                codec::deserialize_msgpack_payload::<serde_json::Value>(payload, "CustomData")?;
555            decode_custom_data_value(custom_type_name, &value)
556                .context("failed to decode MsgPack CustomData")
557        }
558        SerializationEncoding::Sbe | SerializationEncoding::Capnp => {
559            codec::warn_unsupported_inbound(payload_type, encoding);
560            Ok(None)
561        }
562    }
563}
564
565fn decode_custom_data_value(
566    custom_type_name: Ustr,
567    value: &serde_json::Value,
568) -> anyhow::Result<Option<CustomData>> {
569    let Some(data) = deserialize_custom_from_json(custom_type_name.as_str(), value)? else {
570        log::warn!(
571            "External custom payload type '{custom_type_name}' is not registered for inbound republishing"
572        );
573        return Ok(None);
574    };
575
576    let envelope_type_name = value
577        .get("type")
578        .and_then(serde_json::Value::as_str)
579        .context("CustomData JSON missing 'type' field")?;
580    anyhow::ensure!(
581        envelope_type_name == custom_type_name.as_str(),
582        "CustomData envelope type '{envelope_type_name}' does not match message type '{custom_type_name}'"
583    );
584
585    let Data::Custom(custom) = data else {
586        anyhow::bail!("CustomData registry returned non-custom data");
587    };
588
589    Ok(Some(custom))
590}
591
592#[cfg(test)]
593mod tests {
594    use std::{cell::RefCell, rc::Rc};
595
596    use rstest::rstest;
597    use serde::{Serialize, Serializer};
598
599    use super::*;
600    use crate::msgbus::{MessageBus, backing::MessageBusExternalEgress, set_message_bus};
601
602    struct RecordingEgress {
603        messages: Rc<RefCell<Vec<BusMessage>>>,
604        closed: bool,
605    }
606
607    impl MessageBusExternalEgress for RecordingEgress {
608        fn is_closed(&self) -> bool {
609            self.closed
610        }
611
612        fn publish(&self, message: BusMessage) {
613            self.messages.borrow_mut().push(message);
614        }
615
616        fn close(&mut self) {
617            self.closed = true;
618        }
619    }
620
621    struct ReentrantPayload<F>(F);
622
623    impl<F: Fn()> Serialize for ReentrantPayload<F> {
624        fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
625            (self.0)();
626            serializer.serialize_u64(37)
627        }
628    }
629
630    #[rstest]
631    fn serialization_can_publish_before_original_egress_delivery() {
632        let messages = install_egress();
633
634        let payload = ReentrantPayload(|| {
635            forward_external_message("nested".into(), BusPayloadType::QuoteTick, &19_u64);
636        });
637
638        forward_external_message("outer".into(), BusPayloadType::QuoteTick, &payload);
639
640        let messages = messages.borrow();
641        assert_eq!(messages.len(), 2);
642        assert_message(&messages[0], "nested", b"19");
643        assert_message(&messages[1], "outer", b"37");
644    }
645
646    #[rstest]
647    #[case::close(false)]
648    #[case::dispose(true)]
649    fn serialization_can_close_original_egress(#[case] dispose: bool) {
650        let messages = install_egress();
651
652        let payload = ReentrantPayload(move || {
653            let bus = get_message_bus();
654            let mut bus = bus.borrow_mut();
655
656            if dispose {
657                bus.dispose();
658            } else {
659                bus.close().unwrap();
660            }
661        });
662
663        forward_external_message("outer".into(), BusPayloadType::QuoteTick, &payload);
664
665        assert!(messages.borrow().is_empty());
666        assert!(!get_message_bus().borrow().has_external_egress());
667    }
668
669    #[rstest]
670    fn serialization_keeps_original_destination_when_egress_is_replaced() {
671        let original = install_egress();
672        let replacement = Rc::new(RefCell::new(Vec::new()));
673        let replacement_clone = replacement.clone();
674
675        let payload = ReentrantPayload(move || {
676            get_message_bus().borrow_mut().set_external_egress(
677                Box::new(RecordingEgress {
678                    messages: replacement_clone.clone(),
679                    closed: false,
680                }),
681                SerializationEncoding::Json,
682            );
683        });
684
685        forward_external_message("outer".into(), BusPayloadType::QuoteTick, &payload);
686        forward_external_message("later".into(), BusPayloadType::QuoteTick, &23_u64);
687
688        let original = original.borrow();
689        let replacement = replacement.borrow();
690        assert_eq!(original.len(), 1);
691        assert_message(&original[0], "outer", b"37");
692        assert_eq!(replacement.len(), 1);
693        assert_message(&replacement[0], "later", b"23");
694    }
695
696    fn install_egress() -> Rc<RefCell<Vec<BusMessage>>> {
697        let messages = Rc::new(RefCell::new(Vec::new()));
698        let mut bus = MessageBus::default();
699        bus.set_external_egress(
700            Box::new(RecordingEgress {
701                messages: messages.clone(),
702                closed: false,
703            }),
704            SerializationEncoding::Json,
705        );
706
707        set_message_bus(Rc::new(RefCell::new(bus)));
708        messages
709    }
710
711    fn assert_message(message: &BusMessage, topic: &str, payload: &[u8]) {
712        assert_eq!(message.topic.as_str(), topic);
713        assert_eq!(message.payload_type, BusPayloadType::QuoteTick);
714        assert_eq!(message.encoding, SerializationEncoding::Json);
715        assert_eq!(message.payload.as_ref(), payload);
716    }
717}
nautilus_common/msgbus/external/mod.rs

nautilus_common/msgbus/external/
mod.rs
