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::data::{CustomData, Data, deserialize_custom_from_json};
22use serde::de::DeserializeOwned;
23use ustr::Ustr;
24
25pub(crate) mod codec;
26
27use self::codec::PayloadCodecError;
28use super::{
29    BusMessage, BusPayloadType, HAS_EXTERNAL_EGRESS, SUPPRESS_EXTERNAL_DEPTH,
30    SuppressExternalGuard,
31    api::{
32        publish_account_state, publish_any, publish_bar, publish_deltas, publish_depth10,
33        publish_funding_rate, publish_index_price, publish_instrument, publish_mark_price,
34        publish_option_greeks, publish_order_event, publish_portfolio_snapshot,
35        publish_position_event, publish_quote, publish_trade,
36    },
37    get_message_bus,
38    mstr::{MStr, Topic},
39};
40use crate::enums::SerializationEncoding;
41
42#[inline(always)]
43pub(super) fn forward_to_external_egress<T>(
44    topic: MStr<Topic>,
45    payload_type: BusPayloadType,
46    message: &T,
47) where
48    T: serde::Serialize + Any,
49{
50    if !HAS_EXTERNAL_EGRESS.with(Cell::get) {
51        return;
52    }
53
54    forward_external_message(topic, payload_type, message);
55}
56
57#[cold]
58#[inline(never)]
59fn forward_external_message<T>(topic: MStr<Topic>, payload_type: BusPayloadType, message: &T)
60where
61    T: serde::Serialize + Any,
62{
63    if SUPPRESS_EXTERNAL_DEPTH.with(Cell::get) > 0 {
64        return;
65    }
66
67    let bus_rc = get_message_bus();
68    let bus = bus_rc.borrow();
69    let Some(external_egress) = bus
70        .external_egress()
71        .filter(|external_egress| !external_egress.is_closed())
72    else {
73        return;
74    };
75
76    if bus.types_filter().contains(&payload_type) {
77        return;
78    }
79
80    let encoding = bus.encoding_for(payload_type);
81
82    let payload = match codec::serialize_payload(encoding, payload_type, message) {
83        Ok(payload) => payload,
84        Err(PayloadCodecError::Dropped(e)) => {
85            log::debug!("{e}");
86            return;
87        }
88        Err(PayloadCodecError::Failed(e)) => {
89            log::error!("{e}");
90            return;
91        }
92    };
93
94    // Build after drop checks to avoid allocating discarded external messages
95    external_egress.publish(BusMessage::new(*topic, payload_type, payload, encoding));
96}
97
98/// Decodes an externally-received [`BusMessage`] and republishes it onto the internal bus.
99///
100/// The message `payload_type` header selects the concrete type and the message `encoding` selects
101/// the wire codec, so the message is decoded with the producer's encoding rather than the local
102/// configuration. Republishing runs under a [`SuppressExternalGuard`] so the message is not
103/// forwarded straight back out through external egress, which would create an echo loop on a node
104/// that has both external ingress and egress.
105///
106/// # Errors
107///
108/// Returns an error if a supported payload cannot be decoded. Unsupported type/encoding pairs are
109/// skipped with a warning.
110pub fn republish_external_message(message: &BusMessage) -> anyhow::Result<()> {
111    if !is_registered_streaming_type(message) {
112        return Ok(());
113    }
114
115    let _guard = SuppressExternalGuard::new();
116    let topic: MStr<Topic> = message.topic.into();
117
118    match message.payload_type {
119        BusPayloadType::Custom(_) => {
120            handle_custom_data(
121                topic,
122                message.payload_type,
123                message.encoding,
124                &message.payload,
125            )?;
126        }
127        BusPayloadType::Instrument => {
128            handle_json_msgpack(
129                topic,
130                message.payload_type,
131                message.encoding,
132                &message.payload,
133                publish_instrument,
134            )?;
135        }
136        BusPayloadType::OrderBookDeltas => handle_market_data(
137            topic,
138            message.encoding,
139            &message.payload,
140            codec::deserialize_order_book_deltas,
141            publish_deltas,
142        )?,
143        BusPayloadType::OrderBookDepth10 => handle_market_data(
144            topic,
145            message.encoding,
146            &message.payload,
147            codec::deserialize_order_book_depth10,
148            publish_depth10,
149        )?,
150        BusPayloadType::QuoteTick => handle_market_data(
151            topic,
152            message.encoding,
153            &message.payload,
154            codec::deserialize_quote,
155            publish_quote,
156        )?,
157        BusPayloadType::TradeTick => handle_market_data(
158            topic,
159            message.encoding,
160            &message.payload,
161            codec::deserialize_trade,
162            publish_trade,
163        )?,
164        BusPayloadType::Bar => handle_market_data(
165            topic,
166            message.encoding,
167            &message.payload,
168            codec::deserialize_bar,
169            publish_bar,
170        )?,
171        BusPayloadType::MarkPriceUpdate => handle_market_data(
172            topic,
173            message.encoding,
174            &message.payload,
175            codec::deserialize_mark_price,
176            publish_mark_price,
177        )?,
178        BusPayloadType::IndexPriceUpdate => handle_market_data(
179            topic,
180            message.encoding,
181            &message.payload,
182            codec::deserialize_index_price,
183            publish_index_price,
184        )?,
185        BusPayloadType::FundingRateUpdate => handle_market_data(
186            topic,
187            message.encoding,
188            &message.payload,
189            codec::deserialize_funding_rate,
190            publish_funding_rate,
191        )?,
192        BusPayloadType::OptionGreeks => {
193            handle_market_data(
194                topic,
195                message.encoding,
196                &message.payload,
197                codec::deserialize_option_greeks,
198                publish_option_greeks,
199            )?;
200        }
201        BusPayloadType::AccountState => {
202            handle_json_msgpack(
203                topic,
204                message.payload_type,
205                message.encoding,
206                &message.payload,
207                publish_account_state,
208            )?;
209        }
210        BusPayloadType::OrderEvent => {
211            handle_json_msgpack(
212                topic,
213                message.payload_type,
214                message.encoding,
215                &message.payload,
216                publish_order_event,
217            )?;
218        }
219        BusPayloadType::PositionEvent => {
220            handle_json_msgpack(
221                topic,
222                message.payload_type,
223                message.encoding,
224                &message.payload,
225                publish_position_event,
226            )?;
227        }
228        BusPayloadType::PortfolioSnapshot => {
229            handle_json_msgpack(
230                topic,
231                message.payload_type,
232                message.encoding,
233                &message.payload,
234                publish_portfolio_snapshot,
235            )?;
236        }
237        #[cfg(feature = "defi")]
238        BusPayloadType::Block
239        | BusPayloadType::Pool
240        | BusPayloadType::PoolLiquidityUpdate
241        | BusPayloadType::PoolFeeCollect
242        | BusPayloadType::PoolFlash => {
243            crate::defi::msgbus::republish_external_message(
244                topic,
245                message.payload_type,
246                message.encoding,
247                &message.payload,
248            )?;
249        }
250    }
251
252    Ok(())
253}
254
255fn is_registered_streaming_type(message: &BusMessage) -> bool {
256    if get_message_bus()
257        .borrow()
258        .is_streaming_type(message.payload_type)
259    {
260        return true;
261    }
262
263    let type_name = message.payload_type.as_str();
264    if type_name.is_empty() {
265        log::debug!(
266            "Skipping external message on topic '{}' with no payload type for inbound republishing",
267            message.topic
268        );
269    } else {
270        log::debug!(
271            "Skipping external {type_name} message on topic '{}' because the type is not registered for streaming",
272            message.topic
273        );
274    }
275
276    false
277}
278
279pub(crate) fn handle_json_msgpack<T>(
280    topic: MStr<Topic>,
281    payload_type: BusPayloadType,
282    encoding: SerializationEncoding,
283    payload: &[u8],
284    publish: impl FnOnce(MStr<Topic>, &T),
285) -> anyhow::Result<()>
286where
287    T: DeserializeOwned,
288{
289    let Some(value) = codec::deserialize_json_msgpack_payload(payload_type, encoding, payload)?
290    else {
291        return Ok(());
292    };
293
294    publish(topic, &value);
295    Ok(())
296}
297
298fn handle_market_data<T>(
299    topic: MStr<Topic>,
300    encoding: SerializationEncoding,
301    payload: &[u8],
302    deserialize: fn(SerializationEncoding, &[u8]) -> anyhow::Result<Option<T>>,
303    publish: impl FnOnce(MStr<Topic>, &T),
304) -> anyhow::Result<()> {
305    let Some(value) = deserialize(encoding, payload)? else {
306        return Ok(());
307    };
308
309    publish(topic, &value);
310    Ok(())
311}
312
313fn handle_custom_data(
314    topic: MStr<Topic>,
315    payload_type: BusPayloadType,
316    encoding: SerializationEncoding,
317    payload: &[u8],
318) -> anyhow::Result<()> {
319    let Some(custom) = decode_custom_data_payload(payload_type, encoding, payload)? else {
320        return Ok(());
321    };
322
323    publish_any(topic, &custom);
324    Ok(())
325}
326
327fn decode_custom_data_payload(
328    payload_type: BusPayloadType,
329    encoding: SerializationEncoding,
330    payload: &[u8],
331) -> anyhow::Result<Option<CustomData>> {
332    let BusPayloadType::Custom(custom_type_name) = payload_type else {
333        unreachable!("custom data payload decoding requires a custom payload type");
334    };
335
336    if custom_type_name.is_empty() {
337        log::warn!("External payload has no type for inbound republishing");
338        return Ok(None);
339    } else if !payload_type.supports(encoding) {
340        codec::warn_unsupported_inbound(payload_type, encoding);
341        return Ok(None);
342    }
343
344    match encoding {
345        SerializationEncoding::Json => {
346            let value =
347                codec::deserialize_json_payload::<serde_json::Value>(payload, "CustomData")?;
348            decode_custom_data_value(custom_type_name, &value)
349                .context("failed to decode JSON CustomData")
350        }
351        SerializationEncoding::MsgPack => {
352            let value =
353                codec::deserialize_msgpack_payload::<serde_json::Value>(payload, "CustomData")?;
354            decode_custom_data_value(custom_type_name, &value)
355                .context("failed to decode MsgPack CustomData")
356        }
357        SerializationEncoding::Sbe | SerializationEncoding::Capnp => {
358            codec::warn_unsupported_inbound(payload_type, encoding);
359            Ok(None)
360        }
361    }
362}
363
364fn decode_custom_data_value(
365    custom_type_name: Ustr,
366    value: &serde_json::Value,
367) -> anyhow::Result<Option<CustomData>> {
368    let Some(data) = deserialize_custom_from_json(custom_type_name.as_str(), value)? else {
369        log::warn!(
370            "External custom payload type '{custom_type_name}' is not registered for inbound republishing"
371        );
372        return Ok(None);
373    };
374
375    let envelope_type_name = value
376        .get("type")
377        .and_then(serde_json::Value::as_str)
378        .context("CustomData JSON missing 'type' field")?;
379    anyhow::ensure!(
380        envelope_type_name == custom_type_name.as_str(),
381        "CustomData envelope type '{envelope_type_name}' does not match message type '{custom_type_name}'"
382    );
383
384    let Data::Custom(custom) = data else {
385        anyhow::bail!("CustomData registry returned non-custom data");
386    };
387
388    Ok(Some(custom))
389}
nautilus_common/msgbus/external/mod.rs

nautilus_common/msgbus/external/
mod.rs
