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, BusMessageSource, BusPayloadType, HAS_EXTERNAL_EGRESS, MessagingSwitchboard,
33    SUPPRESS_EXTERNAL_DEPTH, SuppressExternalGuard,
34    api::{
35        publish_any, publish_bar, publish_deltas, publish_depth, publish_funding_rate,
36        publish_index_price, publish_instrument, publish_mark_price, publish_option_greeks,
37        publish_order_event, publish_portfolio_snapshot, publish_position_event, publish_quote,
38        publish_trade, send_account_state,
39    },
40    get_message_bus,
41    mstr::{MStr, Topic},
42};
43use crate::{
44    enums::SerializationEncoding,
45    messages::{
46        data::{
47            DataResponse, DataResponseEnvelope, RequestCommand, SubscribeCommand,
48            UnsubscribeCommand,
49        },
50        execution::{GenerateExecutionMassStatus, TradingCommand},
51    },
52};
53
54#[inline(always)]
55pub(super) fn forward_to_external_egress<T>(
56    topic: MStr<Topic>,
57    payload_type: BusPayloadType,
58    message: &T,
59) where
60    T: serde::Serialize + Any,
61{
62    if !HAS_EXTERNAL_EGRESS.with(Cell::get) {
63        return;
64    }
65
66    forward_external_message(topic, payload_type, message);
67}
68
69#[inline]
70pub(super) fn forward_any_to_external_egress(topic: MStr<Topic>, message: &dyn Any) {
71    if !HAS_EXTERNAL_EGRESS.with(Cell::get) {
72        return;
73    }
74
75    if let Some(custom) = message.downcast_ref::<CustomData>() {
76        forward_external_message(
77            topic,
78            BusPayloadType::Custom(Ustr::from(custom.data.type_name())),
79            custom,
80        );
81        return;
82    }
83
84    if forward_downcast::<SubscribeCommand>(topic, BusPayloadType::SubscribeCommand, message)
85        || forward_downcast::<UnsubscribeCommand>(
86            topic,
87            BusPayloadType::UnsubscribeCommand,
88            message,
89        )
90        || forward_downcast::<DataResponse>(topic, BusPayloadType::DataResponse, message)
91        || forward_downcast::<DataResponseEnvelope>(
92            topic,
93            BusPayloadType::DataResponseEnvelope,
94            message,
95        )
96        || forward_downcast::<RequestCommand>(topic, BusPayloadType::RequestCommand, message)
97        || forward_downcast::<TradingCommand>(topic, BusPayloadType::TradingCommand, message)
98        || forward_downcast::<GenerateExecutionMassStatus>(
99            topic,
100            BusPayloadType::GenerateExecutionMassStatus,
101            message,
102        )
103        || forward_downcast::<OrderStatusReport>(topic, BusPayloadType::OrderStatusReport, message)
104        || forward_downcast::<FillReport>(topic, BusPayloadType::FillReport, message)
105        || forward_downcast::<PositionStatusReport>(
106            topic,
107            BusPayloadType::PositionStatusReport,
108            message,
109        )
110    {
111        return;
112    }
113
114    forward_downcast::<ExecutionMassStatus>(topic, BusPayloadType::ExecutionMassStatus, message);
115}
116
117fn forward_downcast<T>(topic: MStr<Topic>, payload_type: BusPayloadType, message: &dyn Any) -> bool
118where
119    T: serde::Serialize + Any,
120{
121    let Some(message) = message.downcast_ref::<T>() else {
122        return false;
123    };
124
125    forward_external_message(topic, payload_type, message);
126    true
127}
128
129#[cold]
130#[inline(never)]
131fn forward_external_message<T>(topic: MStr<Topic>, payload_type: BusPayloadType, message: &T)
132where
133    T: serde::Serialize + Any,
134{
135    if SUPPRESS_EXTERNAL_DEPTH.with(Cell::get) > 0 {
136        return;
137    }
138
139    let (external_egress, encoding) = {
140        let bus_rc = get_message_bus();
141        let bus = bus_rc.borrow();
142
143        let Some(external_egress) = bus.external_egress() else {
144            return;
145        };
146
147        if payload_type.is_typed_message() && !bus.has_external_streams() {
148            return;
149        }
150
151        if bus.types_filter().contains(&payload_type) {
152            return;
153        }
154
155        (external_egress, bus.encoding_for(payload_type))
156    };
157
158    if external_egress.borrow().is_closed() {
159        return;
160    }
161
162    let payload = match codec::serialize_payload(encoding, payload_type, message) {
163        Ok(payload) => payload,
164        Err(PayloadCodecError::Dropped(e)) => {
165            log::debug!("{e}");
166            return;
167        }
168        Err(PayloadCodecError::Failed(e)) => {
169            log::error!("{e}");
170            return;
171        }
172    };
173
174    let external_egress = external_egress.borrow();
175    if external_egress.is_closed() {
176        return;
177    }
178
179    // Serialization can close the original destination or replace it on the bus
180    external_egress.publish(BusMessage::new(*topic, payload_type, payload, encoding));
181}
182
183thread_local! {
184    static EXTERNAL_SOURCE: Cell<Option<BusMessageSource>> = const { Cell::new(None) };
185}
186
187/// Returns the source of the external message currently being dispatched on this thread.
188#[must_use]
189pub fn current_external_source() -> Option<BusMessageSource> {
190    EXTERNAL_SOURCE.get()
191}
192
193struct ExternalSourceGuard(Option<BusMessageSource>);
194
195impl ExternalSourceGuard {
196    fn new(source: Option<BusMessageSource>) -> Self {
197        Self(EXTERNAL_SOURCE.replace(source))
198    }
199}
200
201impl Drop for ExternalSourceGuard {
202    fn drop(&mut self) {
203        EXTERNAL_SOURCE.set(self.0);
204    }
205}
206
207/// Republishes a native data reply only when its correlation handler is pending.
208///
209/// Returns `false` without dispatching when the message is not a streaming-registered
210/// `DataResponse` or `DataResponseEnvelope`, or no handler is pending for its correlation ID.
211/// `LiveNode` uses this during data-client connection, before actors and strategies subscribe.
212/// The reply dispatches through application topic handlers, with external provenance and egress
213/// suppression active during dispatch. Startup replies bypass stream processors and can overtake
214/// buffered messages from the same ingress.
215///
216/// # Errors
217///
218/// Returns an error if a native data reply cannot be decoded or its topic is invalid.
219pub fn republish_pending_external_response(message: &BusMessage) -> anyhow::Result<bool> {
220    if !matches!(
221        message.payload_type,
222        BusPayloadType::DataResponse | BusPayloadType::DataResponseEnvelope
223    ) || !is_registered_streaming_type(message)
224    {
225        return Ok(false);
226    }
227
228    match message.payload_type {
229        BusPayloadType::DataResponseEnvelope => republish_pending_response::<DataResponseEnvelope>(
230            message,
231            DataResponseEnvelope::correlation_id,
232        ),
233        _ => republish_pending_response::<DataResponse>(message, DataResponse::correlation_id),
234    }
235}
236
237fn republish_pending_response<T: serde::de::DeserializeOwned + 'static>(
238    message: &BusMessage,
239    correlation_id: fn(&T) -> &nautilus_core::UUID4,
240) -> anyhow::Result<bool> {
241    let Some(response) =
242        codec::deserialize_payload::<T>(message.payload_type, message.encoding, &message.payload)?
243    else {
244        return Ok(false);
245    };
246
247    if get_message_bus()
248        .borrow()
249        .get_response_handler(correlation_id(&response))
250        .is_none()
251    {
252        return Ok(false);
253    }
254
255    let topic =
256        MStr::<Topic>::topic_from_ustr(message.topic).context("invalid external message topic")?;
257    let _guard = SuppressExternalGuard::new();
258    let _source = ExternalSourceGuard::new(message.source);
259    publish_any(topic, &response);
260    Ok(true)
261}
262
263/// Decodes an externally-received [`BusMessage`] and republishes it onto the internal bus.
264///
265/// The message `payload_type` header selects the concrete type and the message `encoding` selects
266/// the wire codec, so the message is decoded with the producer's encoding rather than the local
267/// configuration. Republishing runs under a [`SuppressExternalGuard`] so the message is not
268/// forwarded straight back out through external egress, which would create an echo loop on a node
269/// that has both external ingress and egress.
270///
271/// Account states are sent to Portfolio's account endpoint for application before Portfolio
272/// publishes them on `events.account.{account_id}`. Portfolio must be registered to process them.
273///
274/// # Errors
275///
276/// Returns an error if the topic is invalid or a supported payload cannot be decoded. Unsupported
277/// encodings for native data requests and responses also return an error. Other unsupported
278/// type/encoding pairs are skipped with a warning.
279pub fn republish_external_message(message: &BusMessage) -> anyhow::Result<()> {
280    let topic =
281        MStr::<Topic>::topic_from_ustr(message.topic).context("invalid external message topic")?;
282
283    if !is_registered_streaming_type(message) {
284        return Ok(());
285    }
286
287    let _guard = SuppressExternalGuard::new();
288    let _source = ExternalSourceGuard::new(message.source);
289
290    match message.payload_type {
291        BusPayloadType::Custom(_) => {
292            handle_custom_data(
293                topic,
294                message.payload_type,
295                message.encoding,
296                &message.payload,
297            )?;
298        }
299        BusPayloadType::Instrument => {
300            handle_payload(
301                topic,
302                message.payload_type,
303                message.encoding,
304                &message.payload,
305                publish_instrument,
306            )?;
307        }
308        BusPayloadType::OrderBookDeltas => handle_market_data(
309            topic,
310            message.encoding,
311            &message.payload,
312            codec::deserialize_order_book_deltas,
313            publish_deltas,
314        )?,
315        BusPayloadType::OrderBookDepth => handle_market_data(
316            topic,
317            message.encoding,
318            &message.payload,
319            codec::deserialize_order_book_depth,
320            publish_depth,
321        )?,
322        BusPayloadType::QuoteTick => handle_market_data(
323            topic,
324            message.encoding,
325            &message.payload,
326            codec::deserialize_quote,
327            publish_quote,
328        )?,
329        BusPayloadType::TradeTick => handle_market_data(
330            topic,
331            message.encoding,
332            &message.payload,
333            codec::deserialize_trade,
334            publish_trade,
335        )?,
336        BusPayloadType::Bar => handle_market_data(
337            topic,
338            message.encoding,
339            &message.payload,
340            codec::deserialize_bar,
341            publish_bar,
342        )?,
343        BusPayloadType::MarkPriceUpdate => handle_market_data(
344            topic,
345            message.encoding,
346            &message.payload,
347            codec::deserialize_mark_price,
348            publish_mark_price,
349        )?,
350        BusPayloadType::IndexPriceUpdate => handle_market_data(
351            topic,
352            message.encoding,
353            &message.payload,
354            codec::deserialize_index_price,
355            publish_index_price,
356        )?,
357        BusPayloadType::FundingRateUpdate => handle_market_data(
358            topic,
359            message.encoding,
360            &message.payload,
361            codec::deserialize_funding_rate,
362            publish_funding_rate,
363        )?,
364        BusPayloadType::OptionGreeks => {
365            handle_market_data(
366                topic,
367                message.encoding,
368                &message.payload,
369                codec::deserialize_option_greeks,
370                publish_option_greeks,
371            )?;
372        }
373        BusPayloadType::AccountState => {
374            handle_payload(
375                topic,
376                message.payload_type,
377                message.encoding,
378                &message.payload,
379                |_, state| {
380                    send_account_state(MessagingSwitchboard::portfolio_update_account(), state);
381                },
382            )?;
383        }
384        BusPayloadType::OrderEvent => {
385            handle_payload(
386                topic,
387                message.payload_type,
388                message.encoding,
389                &message.payload,
390                publish_order_event,
391            )?;
392        }
393        BusPayloadType::PositionEvent => {
394            handle_payload(
395                topic,
396                message.payload_type,
397                message.encoding,
398                &message.payload,
399                publish_position_event,
400            )?;
401        }
402        BusPayloadType::PortfolioSnapshot => {
403            handle_payload(
404                topic,
405                message.payload_type,
406                message.encoding,
407                &message.payload,
408                publish_portfolio_snapshot,
409            )?;
410        }
411        BusPayloadType::SubscribeCommand => {
412            handle_payload_any::<SubscribeCommand>(topic, message)?;
413        }
414        BusPayloadType::UnsubscribeCommand => {
415            handle_payload_any::<UnsubscribeCommand>(topic, message)?;
416        }
417        BusPayloadType::DataResponse => {
418            handle_payload_any::<DataResponse>(topic, message)?;
419        }
420        BusPayloadType::DataResponseEnvelope => {
421            handle_payload_any::<DataResponseEnvelope>(topic, message)?;
422        }
423        BusPayloadType::RequestCommand => {
424            handle_payload_any::<RequestCommand>(topic, message)?;
425        }
426        BusPayloadType::TradingCommand => {
427            handle_payload_any::<TradingCommand>(topic, message)?;
428        }
429        BusPayloadType::GenerateExecutionMassStatus => {
430            handle_payload_any::<GenerateExecutionMassStatus>(topic, message)?;
431        }
432        BusPayloadType::OrderStatusReport => {
433            handle_payload_any::<OrderStatusReport>(topic, message)?;
434        }
435        BusPayloadType::FillReport => {
436            handle_payload_any::<FillReport>(topic, message)?;
437        }
438        BusPayloadType::PositionStatusReport => {
439            handle_payload_any::<PositionStatusReport>(topic, message)?;
440        }
441        BusPayloadType::ExecutionMassStatus => {
442            handle_payload_any::<ExecutionMassStatus>(topic, message)?;
443        }
444        #[cfg(feature = "defi")]
445        BusPayloadType::Block
446        | BusPayloadType::Pool
447        | BusPayloadType::PoolLiquidityUpdate
448        | BusPayloadType::PoolFeeCollect
449        | BusPayloadType::PoolFlash => {
450            crate::defi::msgbus::republish_external_message(
451                topic,
452                message.payload_type,
453                message.encoding,
454                &message.payload,
455            )?;
456        }
457    }
458
459    Ok(())
460}
461
462/// Decodes a supported typed payload and passes it to `processor` before applying normal inbound
463/// republishing rules.
464///
465/// Supported typed JSON or MessagePack payloads reach `processor` regardless of internal streaming
466/// registration. Internal republishing still requires registration. Native requests and replies
467/// use the shared JSON/MessagePack codecs; unsupported encodings return an error.
468/// Other unsupported type/encoding pairs are skipped with a warning. The processor mapping
469/// includes a `payload_type` field containing the [`BusPayloadType`] name. External egress is
470/// suppressed for the duration of processing, including processor callbacks. Other payloads bypass
471/// the processor and follow normal inbound republishing. A processor error stops processing and
472/// skips internal republishing.
473///
474/// # Errors
475///
476/// Returns an error if:
477/// - The topic is invalid.
478/// - A native data request or response uses an unsupported encoding.
479/// - The typed payload cannot be decoded or mapped to an object.
480/// - The mapping already contains the reserved `payload_type` field.
481/// - The processor fails.
482/// - Normal inbound republishing fails.
483pub fn process_external_typed_message(
484    message: &BusMessage,
485    processor: &mut dyn FnMut(&dyn Any, &serde_json::Value) -> anyhow::Result<()>,
486) -> anyhow::Result<()> {
487    let topic =
488        MStr::<Topic>::topic_from_ustr(message.topic).context("invalid external message topic")?;
489    let _guard = SuppressExternalGuard::new();
490    let _source = ExternalSourceGuard::new(message.source);
491
492    #[rustfmt::skip]
493    let result = match message.payload_type {
494        BusPayloadType::SubscribeCommand => process_typed_payload::<SubscribeCommand>(topic, message, processor),
495        BusPayloadType::UnsubscribeCommand => process_typed_payload::<UnsubscribeCommand>(topic, message, processor),
496        BusPayloadType::DataResponse => process_typed_payload::<DataResponse>(topic, message, processor),
497        BusPayloadType::DataResponseEnvelope => process_typed_payload::<DataResponseEnvelope>(topic, message, processor),
498        BusPayloadType::RequestCommand => process_typed_payload::<RequestCommand>(topic, message, processor),
499        BusPayloadType::TradingCommand => process_typed_payload::<TradingCommand>(topic, message, processor),
500        BusPayloadType::GenerateExecutionMassStatus => process_typed_payload::<GenerateExecutionMassStatus>(topic, message, processor),
501        BusPayloadType::OrderStatusReport => process_typed_payload::<OrderStatusReport>(topic, message, processor),
502        BusPayloadType::FillReport => process_typed_payload::<FillReport>(topic, message, processor),
503        BusPayloadType::PositionStatusReport => process_typed_payload::<PositionStatusReport>(topic, message, processor),
504        BusPayloadType::ExecutionMassStatus => process_typed_payload::<ExecutionMassStatus>(topic, message, processor),
505        _ => republish_external_message(message),
506    };
507    result
508}
509
510fn handle_payload_any<T>(topic: MStr<Topic>, message: &BusMessage) -> anyhow::Result<()>
511where
512    T: DeserializeOwned + Any,
513{
514    handle_payload(
515        topic,
516        message.payload_type,
517        message.encoding,
518        &message.payload,
519        |topic, value: &T| publish_any(topic, value),
520    )
521}
522
523fn process_typed_payload<T>(
524    topic: MStr<Topic>,
525    message: &BusMessage,
526    processor: &mut dyn FnMut(&dyn Any, &serde_json::Value) -> anyhow::Result<()>,
527) -> anyhow::Result<()>
528where
529    T: DeserializeOwned + serde::Serialize + Any,
530{
531    let Some(value) =
532        codec::deserialize_payload::<T>(message.payload_type, message.encoding, &message.payload)?
533    else {
534        return Ok(());
535    };
536
537    let mut mapping = serde_json::to_value(&value).with_context(|| {
538        format!(
539            "failed to map decoded {} stream payload",
540            message.payload_type
541        )
542    })?;
543
544    let mapping_object = mapping.as_object_mut().with_context(|| {
545        format!(
546            "decoded {} stream payload did not map to an object",
547            message.payload_type
548        )
549    })?;
550
551    anyhow::ensure!(
552        !mapping_object.contains_key("payload_type"),
553        "decoded {} stream payload contains reserved payload_type field",
554        message.payload_type
555    );
556    mapping_object.insert(
557        "payload_type".to_string(),
558        serde_json::Value::String(message.payload_type.as_str().to_string()),
559    );
560
561    processor(&value, &mapping)?;
562    if is_registered_streaming_type(message) {
563        publish_any(topic, &value);
564    }
565
566    Ok(())
567}
568
569fn is_registered_streaming_type(message: &BusMessage) -> bool {
570    if get_message_bus()
571        .borrow()
572        .is_streaming_type(message.payload_type)
573    {
574        return true;
575    }
576
577    let type_name = message.payload_type.as_str();
578    if type_name.is_empty() {
579        log::debug!(
580            "Skipping external message on topic '{}' with no payload type for inbound republishing",
581            message.topic
582        );
583    } else {
584        log::debug!(
585            "Skipping external {type_name} message on topic '{}' because the type is not registered for streaming",
586            message.topic
587        );
588    }
589
590    false
591}
592
593pub(crate) fn handle_payload<T>(
594    topic: MStr<Topic>,
595    payload_type: BusPayloadType,
596    encoding: SerializationEncoding,
597    payload: &[u8],
598    publish: impl FnOnce(MStr<Topic>, &T),
599) -> anyhow::Result<()>
600where
601    T: DeserializeOwned + 'static,
602{
603    let Some(value) = codec::deserialize_payload(payload_type, encoding, payload)? else {
604        return Ok(());
605    };
606
607    publish(topic, &value);
608    Ok(())
609}
610
611fn handle_market_data<T>(
612    topic: MStr<Topic>,
613    encoding: SerializationEncoding,
614    payload: &[u8],
615    deserialize: fn(SerializationEncoding, &[u8]) -> anyhow::Result<Option<T>>,
616    publish: impl FnOnce(MStr<Topic>, &T),
617) -> anyhow::Result<()> {
618    let Some(value) = deserialize(encoding, payload)? else {
619        return Ok(());
620    };
621
622    publish(topic, &value);
623    Ok(())
624}
625
626fn handle_custom_data(
627    topic: MStr<Topic>,
628    payload_type: BusPayloadType,
629    encoding: SerializationEncoding,
630    payload: &[u8],
631) -> anyhow::Result<()> {
632    let Some(custom) = decode_custom_data_payload(payload_type, encoding, payload)? else {
633        return Ok(());
634    };
635
636    publish_any(topic, &custom);
637    Ok(())
638}
639
640fn decode_custom_data_payload(
641    payload_type: BusPayloadType,
642    encoding: SerializationEncoding,
643    payload: &[u8],
644) -> anyhow::Result<Option<CustomData>> {
645    let BusPayloadType::Custom(custom_type_name) = payload_type else {
646        unreachable!("custom data payload decoding requires a custom payload type");
647    };
648
649    if custom_type_name.is_empty() {
650        log::warn!("External payload has no type for inbound republishing");
651        return Ok(None);
652    } else if !payload_type.supports(encoding) {
653        codec::warn_unsupported_inbound(payload_type, encoding);
654        return Ok(None);
655    }
656
657    match encoding {
658        SerializationEncoding::Json => {
659            let value =
660                codec::deserialize_json_payload::<serde_json::Value>(payload, "CustomData")?;
661            decode_custom_data_value(custom_type_name, &value)
662                .context("failed to decode JSON CustomData")
663        }
664        SerializationEncoding::MsgPack => {
665            let value =
666                codec::deserialize_msgpack_payload::<serde_json::Value>(payload, "CustomData")?;
667            decode_custom_data_value(custom_type_name, &value)
668                .context("failed to decode MsgPack CustomData")
669        }
670        SerializationEncoding::Sbe => {
671            codec::warn_unsupported_inbound(payload_type, encoding);
672            Ok(None)
673        }
674    }
675}
676
677fn decode_custom_data_value(
678    custom_type_name: Ustr,
679    value: &serde_json::Value,
680) -> anyhow::Result<Option<CustomData>> {
681    let Some(data) = deserialize_custom_from_json(custom_type_name.as_str(), value)? else {
682        log::warn!(
683            "External custom payload type '{custom_type_name}' is not registered for inbound republishing"
684        );
685        return Ok(None);
686    };
687
688    let envelope_type_name = value
689        .get("type")
690        .and_then(serde_json::Value::as_str)
691        .context("CustomData JSON missing 'type' field")?;
692    anyhow::ensure!(
693        envelope_type_name == custom_type_name.as_str(),
694        "CustomData envelope type '{envelope_type_name}' does not match message type '{custom_type_name}'"
695    );
696
697    let Data::Custom(custom) = data else {
698        anyhow::bail!("CustomData registry returned non-custom data");
699    };
700
701    Ok(Some(custom))
702}
703
704#[cfg(test)]
705mod tests {
706    use std::{cell::RefCell, rc::Rc};
707
708    use bytes::Bytes;
709    use rstest::rstest;
710    use serde::{Serialize, Serializer};
711
712    use super::*;
713    use crate::msgbus::{
714        MessageBus, TypedHandler, backing::MessageBusExternalEgress, set_message_bus,
715    };
716
717    struct RecordingEgress {
718        messages: Rc<RefCell<Vec<BusMessage>>>,
719        closed: bool,
720    }
721
722    impl MessageBusExternalEgress for RecordingEgress {
723        fn is_closed(&self) -> bool {
724            self.closed
725        }
726
727        fn publish(&self, message: BusMessage) {
728            self.messages.borrow_mut().push(message);
729        }
730
731        fn close(&mut self) {
732            self.closed = true;
733        }
734    }
735
736    struct ReentrantPayload<F>(F);
737
738    impl<F: Fn()> Serialize for ReentrantPayload<F> {
739        fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
740            (self.0)();
741            serializer.serialize_u64(37)
742        }
743    }
744
745    #[rstest]
746    fn serialization_can_publish_before_original_egress_delivery() {
747        let messages = install_egress();
748
749        let payload = ReentrantPayload(|| {
750            forward_external_message("nested".into(), BusPayloadType::QuoteTick, &19_u64);
751        });
752
753        forward_external_message("outer".into(), BusPayloadType::QuoteTick, &payload);
754
755        let messages = messages.borrow();
756        assert_eq!(messages.len(), 2);
757        assert_message(&messages[0], "nested", b"19");
758        assert_message(&messages[1], "outer", b"37");
759    }
760
761    #[rstest]
762    #[case::close(false)]
763    #[case::dispose(true)]
764    fn serialization_can_close_original_egress(#[case] dispose: bool) {
765        let messages = install_egress();
766
767        let payload = ReentrantPayload(move || {
768            let bus = get_message_bus();
769            let mut bus = bus.borrow_mut();
770            if dispose {
771                bus.dispose();
772            } else {
773                bus.close().unwrap();
774            }
775        });
776
777        forward_external_message("outer".into(), BusPayloadType::QuoteTick, &payload);
778
779        assert!(messages.borrow().is_empty());
780        assert!(!get_message_bus().borrow().has_external_egress());
781    }
782
783    #[rstest]
784    fn serialization_keeps_original_destination_when_egress_is_replaced() {
785        let original = install_egress();
786        let replacement = Rc::new(RefCell::new(Vec::new()));
787        let replacement_clone = replacement.clone();
788
789        let payload = ReentrantPayload(move || {
790            get_message_bus().borrow_mut().set_external_egress(
791                Box::new(RecordingEgress {
792                    messages: replacement_clone.clone(),
793                    closed: false,
794                }),
795                SerializationEncoding::Json,
796            );
797        });
798
799        forward_external_message("outer".into(), BusPayloadType::QuoteTick, &payload);
800        forward_external_message("later".into(), BusPayloadType::QuoteTick, &23_u64);
801
802        let original = original.borrow();
803        let replacement = replacement.borrow();
804        assert_eq!(original.len(), 1);
805        assert_message(&original[0], "outer", b"37");
806        assert_eq!(replacement.len(), 1);
807        assert_message(&replacement[0], "later", b"23");
808    }
809
810    #[rstest]
811    #[case(SerializationEncoding::Json)]
812    #[case(SerializationEncoding::MsgPack)]
813    fn test_native_requests_reach_processors_without_stream_registration(
814        #[case] encoding: SerializationEncoding,
815    ) {
816        use nautilus_core::UUID4;
817        use nautilus_model::{data::DataType, identifiers::ClientId};
818
819        use crate::messages::data::RequestCustomData;
820
821        let publications = install_egress();
822        get_message_bus()
823            .borrow_mut()
824            .set_external_egress_config(
825                Box::new(RecordingEgress {
826                    messages: publications.clone(),
827                    closed: false,
828                }),
829                &crate::msgbus::MessageBusConfig {
830                    encoding,
831                    external_streams: Some(vec!["external.test".into()]),
832                    ..Default::default()
833                },
834            )
835            .unwrap();
836
837        let request = RequestCommand::Data(RequestCustomData::new(
838            ClientId::from("FEED"),
839            DataType::new("DistinctRequest", None, None),
840            None,
841            None,
842            None,
843            UUID4::new(),
844            71.into(),
845            None,
846        ));
847
848        let payload_type = BusPayloadType::RequestCommand;
849        publish_any("commands.data.FEED".into(), &request);
850
851        let message = publications.borrow()[0].clone();
852        let mut received = None;
853        process_external_typed_message(&message, &mut |value, mapping| {
854            let decoded =
855                serde_json::to_value(value.downcast_ref::<RequestCommand>().unwrap()).unwrap();
856
857            received = Some((decoded, mapping.clone()));
858            Ok(())
859        })
860        .unwrap();
861
862        let expected_value = serde_json::to_value(&request).unwrap();
863
864        let mut expected_mapping = expected_value.clone();
865        expected_mapping
866            .as_object_mut()
867            .unwrap()
868            .insert("payload_type".into(), payload_type.as_str().into());
869        assert_eq!(publications.borrow().len(), 1);
870        assert_eq!(message.topic.as_str(), "commands.data.FEED");
871        assert_eq!(message.payload_type, payload_type);
872        assert_eq!(message.encoding, encoding);
873        assert_eq!(received, Some((expected_value, expected_mapping)));
874    }
875
876    #[rstest]
877    #[case(BusPayloadType::RequestCommand, "data request")]
878    #[case(BusPayloadType::DataResponse, "data response")]
879    fn test_native_messages_reject_unsupported_encoding_without_dispatch(
880        #[case] payload_type: BusPayloadType,
881        #[case] name: &str,
882        #[values(SerializationEncoding::Sbe)] encoding: SerializationEncoding,
883    ) {
884        set_message_bus(Rc::new(RefCell::new(MessageBus::default())));
885        get_message_bus()
886            .borrow_mut()
887            .add_streaming_type(payload_type);
888        let message = BusMessage::with_str_topic(
889            "external.test",
890            payload_type,
891            Bytes::from_static(b"unsupported"),
892            encoding,
893        );
894        let expected = format!("{name} encoding is not supported: {encoding}");
895        let mut calls = 0;
896        let error = process_external_typed_message(&message, &mut |_, _| {
897            calls += 1;
898            Ok(())
899        })
900        .unwrap_err();
901
902        assert_eq!(error.to_string(), expected);
903        assert_eq!(
904            republish_external_message(&message)
905                .unwrap_err()
906                .to_string(),
907            expected
908        );
909        assert_eq!(calls, 0);
910    }
911
912    #[rstest]
913    #[case(false)]
914    #[case(true)]
915    fn test_external_source_restored_after_nested_dispatch_and_processor_failure(
916        #[case] panic: bool,
917    ) {
918        use std::panic::{AssertUnwindSafe, catch_unwind};
919
920        use nautilus_core::UUID4;
921        use nautilus_model::{data::DataType, identifiers::ClientId};
922
923        use crate::{
924            messages::data::RequestCustomData,
925            msgbus::{ShareableMessageHandler, subscribe_any},
926        };
927
928        set_message_bus(Rc::new(RefCell::new(MessageBus::default())));
929        get_message_bus()
930            .borrow_mut()
931            .add_streaming_type(BusPayloadType::RequestCommand);
932        let request = RequestCommand::Data(RequestCustomData::new(
933            ClientId::from("FEED"),
934            DataType::new("DistinctRequest", None, None),
935            None,
936            None,
937            None,
938            UUID4::new(),
939            71.into(),
940            None,
941        ));
942        let payload = codec::serialize_payload(
943            SerializationEncoding::Json,
944            BusPayloadType::RequestCommand,
945            &request,
946        )
947        .unwrap();
948        let mut message = BusMessage::with_str_topic(
949            "external.outer",
950            BusPayloadType::RequestCommand,
951            payload,
952            SerializationEncoding::Json,
953        );
954
955        let source = BusMessageSource {
956            publisher_id: "PUBLISHER-017".into(),
957            instance_id: UUID4::new(),
958            stream: "snapshot".into(),
959            sequence: 37,
960        };
961
962        message.source = Some(source);
963        let mut nested = message.clone();
964        nested.topic = "external.nested".into();
965        nested.source = None;
966        let received = Rc::new(RefCell::new(Vec::new()));
967        let shared = received.clone();
968        subscribe_any(
969            nested.topic.as_str().into(),
970            ShareableMessageHandler::from_typed::<RequestCommand, _>(move |_| {
971                shared.borrow_mut().push(current_external_source());
972            }),
973            None,
974        );
975
976        let result = catch_unwind(AssertUnwindSafe(|| {
977            process_external_typed_message(&message, &mut |_, _| {
978                assert_eq!(current_external_source(), Some(source));
979                republish_external_message(&nested)?;
980                assert_eq!(current_external_source(), Some(source));
981                assert_eq!(*received.borrow(), vec![None]);
982
983                assert!(!panic, "processor panicked");
984
985                anyhow::bail!("processor rejected payload")
986            })
987        }));
988
989        if panic {
990            assert_eq!(
991                result.unwrap_err().downcast_ref::<&str>(),
992                Some(&"processor panicked")
993            );
994        } else {
995            assert_eq!(
996                result.unwrap().unwrap_err().to_string(),
997                "processor rejected payload"
998            );
999        }
1000
1001        assert_eq!(current_external_source(), None);
1002    }
1003
1004    #[rstest]
1005    #[case(false, false)]
1006    #[case(true, false)]
1007    #[case(false, true)]
1008    #[case(true, true)]
1009    fn test_native_response_processors_precede_streaming_filters(
1010        #[case] streaming: bool,
1011        #[case] rejected: bool,
1012    ) {
1013        use nautilus_core::UUID4;
1014        use nautilus_model::{data::DataType, identifiers::ClientId};
1015
1016        use crate::{
1017            messages::data::CustomDataResponse,
1018            msgbus::{self, ShareableMessageHandler},
1019        };
1020
1021        set_message_bus(Rc::new(RefCell::new(MessageBus::default())));
1022
1023        if streaming {
1024            get_message_bus()
1025                .borrow_mut()
1026                .add_streaming_type(BusPayloadType::DataResponse);
1027        }
1028
1029        let response = CustomDataResponse::new(
1030            UUID4::new(),
1031            ClientId::from("FEED"),
1032            None,
1033            DataType::new("DistinctReply", None, None),
1034            Bytes::from_static(b"native-bytes"),
1035            Some(67.into()),
1036            Some(71.into()),
1037            73.into(),
1038            None,
1039        );
1040        let publications = Rc::new(RefCell::new(Vec::new()));
1041        get_message_bus()
1042            .borrow_mut()
1043            .set_external_egress_config(
1044                Box::new(RecordingEgress {
1045                    messages: publications.clone(),
1046                    closed: false,
1047                }),
1048                &crate::msgbus::MessageBusConfig {
1049                    encoding: SerializationEncoding::MsgPack,
1050                    external_streams: Some(vec!["external.test".into()]),
1051                    ..Default::default()
1052                },
1053            )
1054            .unwrap();
1055
1056        publish_any(
1057            "responses.data.FEED".into(),
1058            &DataResponse::Data(response.clone()),
1059        );
1060        let message = publications.borrow()[0].clone();
1061        let published = Rc::new(RefCell::new(0));
1062        let shared = published.clone();
1063        msgbus::subscribe_any(
1064            message.topic.as_str().into(),
1065            ShareableMessageHandler::from_typed::<DataResponse, _>(move |_| {
1066                *shared.borrow_mut() += 1;
1067            }),
1068            None,
1069        );
1070
1071        let mut calls = 0;
1072        let result = process_external_typed_message(&message, &mut |value, mapping| {
1073            calls += 1;
1074
1075            let DataResponse::Data(actual) = value.downcast_ref::<DataResponse>().unwrap() else {
1076                panic!("Expected native custom response");
1077            };
1078
1079            assert_eq!(actual.correlation_id, response.correlation_id);
1080            assert_eq!(
1081                actual.data.downcast_ref::<Bytes>(),
1082                response.data.downcast_ref::<Bytes>()
1083            );
1084            assert_eq!(
1085                mapping,
1086                &serde_json::json!({
1087                    "payload_type": "DataResponse",
1088                    "Data": {
1089                        "correlation_id": response.correlation_id, "client_id": "FEED",
1090                        "start": 67, "end": 71, "ts_init": 73, "params": null,
1091                        "venue": null, "data_type": response.data_type,
1092                        "data": {"Buffer": b"native-bytes".to_vec()},
1093                    }
1094                })
1095            );
1096
1097            if rejected {
1098                anyhow::bail!("processor rejected reply");
1099            }
1100
1101            Ok(())
1102        });
1103
1104        assert_eq!(calls, 1);
1105        assert_eq!(publications.borrow().len(), 1);
1106        assert_eq!(message.topic.as_str(), "responses.data.FEED");
1107        assert_eq!(message.payload_type, BusPayloadType::DataResponse);
1108        assert_eq!(message.encoding, SerializationEncoding::MsgPack);
1109        assert_eq!(
1110            result.err().map(|e| e.to_string()),
1111            rejected.then(|| "processor rejected reply".to_string())
1112        );
1113        assert_eq!(*published.borrow(), usize::from(streaming && !rejected));
1114    }
1115
1116    #[rstest]
1117    fn test_startup_dispatches_only_registered_pending_data_responses() {
1118        use nautilus_core::UUID4;
1119        use nautilus_model::{data::DataType, identifiers::ClientId};
1120
1121        use crate::{
1122            messages::data::CustomDataResponse,
1123            msgbus::{self, ShareableMessageHandler},
1124        };
1125
1126        let bus = Rc::new(RefCell::new(MessageBus::default()));
1127        set_message_bus(bus.clone());
1128        bus.borrow_mut()
1129            .add_streaming_type(BusPayloadType::DataResponse);
1130        let id = UUID4::new();
1131        let response = DataResponse::Data(CustomDataResponse::new(
1132            id,
1133            ClientId::from("FEED"),
1134            None,
1135            DataType::new("DistinctReply", None, None),
1136            Bytes::from_static(b"native-bytes"),
1137            None,
1138            None,
1139            73.into(),
1140            None,
1141        ));
1142        let message = BusMessage::with_str_topic(
1143            "responses.data.FEED",
1144            BusPayloadType::DataResponse,
1145            codec::serialize_payload(
1146                SerializationEncoding::MsgPack,
1147                BusPayloadType::DataResponse,
1148                &response,
1149            )
1150            .unwrap(),
1151            SerializationEncoding::MsgPack,
1152        );
1153
1154        let topic_handler = ShareableMessageHandler::from_typed::<DataResponse, _>(|reply| {
1155            msgbus::send_response(reply.correlation_id(), reply);
1156        });
1157
1158        msgbus::subscribe_any(message.topic.as_str().into(), topic_handler, None);
1159        let received = Rc::new(RefCell::new(Vec::new()));
1160        let shared = received.clone();
1161        assert!(!republish_pending_external_response(&message).unwrap());
1162        msgbus::register_response_handler(
1163            &id,
1164            TypedHandler::from(move |envelope: &DataResponseEnvelope| {
1165                let Some(DataResponse::Data(reply)) = envelope.response() else {
1166                    panic!("expected custom reply");
1167                };
1168
1169                shared.borrow_mut().push((
1170                    reply.correlation_id,
1171                    reply.data.downcast_ref::<Bytes>().unwrap().clone(),
1172                    reply.ts_init,
1173                ));
1174            }),
1175        );
1176
1177        assert!(republish_pending_external_response(&message).unwrap());
1178        assert!(!republish_pending_external_response(&message).unwrap());
1179        assert_eq!(
1180            *received.borrow(),
1181            vec![(id, Bytes::from_static(b"native-bytes"), 73.into())]
1182        );
1183    }
1184
1185    #[rstest]
1186    #[case(SerializationEncoding::Json)]
1187    #[case(SerializationEncoding::MsgPack)]
1188    fn test_startup_dispatches_payload_free_completion(#[case] encoding: SerializationEncoding) {
1189        use crate::msgbus::{self, ShareableMessageHandler};
1190
1191        let bus = Rc::new(RefCell::new(MessageBus::default()));
1192        set_message_bus(bus.clone());
1193        bus.borrow_mut()
1194            .add_streaming_type(BusPayloadType::DataResponseEnvelope);
1195        let id = nautilus_core::UUID4::new();
1196        let envelope = DataResponseEnvelope::terminal(id, Some("Source unavailable".to_owned()));
1197        let payload =
1198            codec::serialize_payload(encoding, BusPayloadType::DataResponseEnvelope, &envelope)
1199                .unwrap();
1200        let message = BusMessage::with_str_topic(
1201            "responses.data.FEED",
1202            BusPayloadType::DataResponseEnvelope,
1203            payload,
1204            encoding,
1205        );
1206        let received = Rc::new(RefCell::new(Vec::new()));
1207        let shared = received.clone();
1208        msgbus::subscribe_any(
1209            message.topic.as_str().into(),
1210            ShareableMessageHandler::from_typed::<DataResponseEnvelope, _>(
1211                msgbus::send_response_envelope,
1212            ),
1213            None,
1214        );
1215        msgbus::register_response_handler(
1216            &id,
1217            TypedHandler::from(move |response: &DataResponseEnvelope| {
1218                shared.borrow_mut().push(response.clone());
1219            }),
1220        );
1221
1222        assert!(republish_pending_external_response(&message).unwrap());
1223
1224        let received = received.borrow();
1225        assert_eq!(received.len(), 1);
1226        assert_eq!(received[0].correlation_id(), &id);
1227        assert!(received[0].completed());
1228        assert!(received[0].response().is_none());
1229        assert_eq!(received[0].sub_correlation_id(), None);
1230        assert_eq!(received[0].failure_reason(), Some("Source unavailable"));
1231        assert!(bus.borrow().get_response_handler(&id).is_none());
1232        assert!(!republish_pending_external_response(&message).unwrap());
1233    }
1234
1235    fn install_egress() -> Rc<RefCell<Vec<BusMessage>>> {
1236        let messages = Rc::new(RefCell::new(Vec::new()));
1237        let mut bus = MessageBus::default();
1238        bus.set_external_egress(
1239            Box::new(RecordingEgress {
1240                messages: messages.clone(),
1241                closed: false,
1242            }),
1243            SerializationEncoding::Json,
1244        );
1245
1246        set_message_bus(Rc::new(RefCell::new(bus)));
1247        messages
1248    }
1249
1250    fn assert_message(message: &BusMessage, topic: &str, payload: &[u8]) {
1251        assert_eq!(message.topic.as_str(), topic);
1252        assert_eq!(message.payload_type, BusPayloadType::QuoteTick);
1253        assert_eq!(message.encoding, SerializationEncoding::Json);
1254        assert_eq!(message.payload.as_ref(), payload);
1255    }
1256}
nautilus_common/msgbus/external/mod.rs

nautilus_common/msgbus/external/
mod.rs
