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 the topic is invalid or a supported payload cannot be decoded. Unsupported
109/// type/encoding pairs are skipped with a warning.
110pub fn republish_external_message(message: &BusMessage) -> anyhow::Result<()> {
111    let topic =
112        MStr::<Topic>::topic_from_ustr(message.topic).context("invalid external message topic")?;
113
114    if !is_registered_streaming_type(message) {
115        return Ok(());
116    }
117
118    let _guard = SuppressExternalGuard::new();
119
120    match message.payload_type {
121        BusPayloadType::Custom(_) => {
122            handle_custom_data(
123                topic,
124                message.payload_type,
125                message.encoding,
126                &message.payload,
127            )?;
128        }
129        BusPayloadType::Instrument => {
130            handle_json_msgpack(
131                topic,
132                message.payload_type,
133                message.encoding,
134                &message.payload,
135                publish_instrument,
136            )?;
137        }
138        BusPayloadType::OrderBookDeltas => handle_market_data(
139            topic,
140            message.encoding,
141            &message.payload,
142            codec::deserialize_order_book_deltas,
143            publish_deltas,
144        )?,
145        BusPayloadType::OrderBookDepth10 => handle_market_data(
146            topic,
147            message.encoding,
148            &message.payload,
149            codec::deserialize_order_book_depth10,
150            publish_depth10,
151        )?,
152        BusPayloadType::QuoteTick => handle_market_data(
153            topic,
154            message.encoding,
155            &message.payload,
156            codec::deserialize_quote,
157            publish_quote,
158        )?,
159        BusPayloadType::TradeTick => handle_market_data(
160            topic,
161            message.encoding,
162            &message.payload,
163            codec::deserialize_trade,
164            publish_trade,
165        )?,
166        BusPayloadType::Bar => handle_market_data(
167            topic,
168            message.encoding,
169            &message.payload,
170            codec::deserialize_bar,
171            publish_bar,
172        )?,
173        BusPayloadType::MarkPriceUpdate => handle_market_data(
174            topic,
175            message.encoding,
176            &message.payload,
177            codec::deserialize_mark_price,
178            publish_mark_price,
179        )?,
180        BusPayloadType::IndexPriceUpdate => handle_market_data(
181            topic,
182            message.encoding,
183            &message.payload,
184            codec::deserialize_index_price,
185            publish_index_price,
186        )?,
187        BusPayloadType::FundingRateUpdate => handle_market_data(
188            topic,
189            message.encoding,
190            &message.payload,
191            codec::deserialize_funding_rate,
192            publish_funding_rate,
193        )?,
194        BusPayloadType::OptionGreeks => {
195            handle_market_data(
196                topic,
197                message.encoding,
198                &message.payload,
199                codec::deserialize_option_greeks,
200                publish_option_greeks,
201            )?;
202        }
203        BusPayloadType::AccountState => {
204            handle_json_msgpack(
205                topic,
206                message.payload_type,
207                message.encoding,
208                &message.payload,
209                publish_account_state,
210            )?;
211        }
212        BusPayloadType::OrderEvent => {
213            handle_json_msgpack(
214                topic,
215                message.payload_type,
216                message.encoding,
217                &message.payload,
218                publish_order_event,
219            )?;
220        }
221        BusPayloadType::PositionEvent => {
222            handle_json_msgpack(
223                topic,
224                message.payload_type,
225                message.encoding,
226                &message.payload,
227                publish_position_event,
228            )?;
229        }
230        BusPayloadType::PortfolioSnapshot => {
231            handle_json_msgpack(
232                topic,
233                message.payload_type,
234                message.encoding,
235                &message.payload,
236                publish_portfolio_snapshot,
237            )?;
238        }
239        #[cfg(feature = "defi")]
240        BusPayloadType::Block
241        | BusPayloadType::Pool
242        | BusPayloadType::PoolLiquidityUpdate
243        | BusPayloadType::PoolFeeCollect
244        | BusPayloadType::PoolFlash => {
245            crate::defi::msgbus::republish_external_message(
246                topic,
247                message.payload_type,
248                message.encoding,
249                &message.payload,
250            )?;
251        }
252    }
253
254    Ok(())
255}
256
257fn is_registered_streaming_type(message: &BusMessage) -> bool {
258    if get_message_bus()
259        .borrow()
260        .is_streaming_type(message.payload_type)
261    {
262        return true;
263    }
264
265    let type_name = message.payload_type.as_str();
266    if type_name.is_empty() {
267        log::debug!(
268            "Skipping external message on topic '{}' with no payload type for inbound republishing",
269            message.topic
270        );
271    } else {
272        log::debug!(
273            "Skipping external {type_name} message on topic '{}' because the type is not registered for streaming",
274            message.topic
275        );
276    }
277
278    false
279}
280
281pub(crate) fn handle_json_msgpack<T>(
282    topic: MStr<Topic>,
283    payload_type: BusPayloadType,
284    encoding: SerializationEncoding,
285    payload: &[u8],
286    publish: impl FnOnce(MStr<Topic>, &T),
287) -> anyhow::Result<()>
288where
289    T: DeserializeOwned,
290{
291    let Some(value) = codec::deserialize_json_msgpack_payload(payload_type, encoding, payload)?
292    else {
293        return Ok(());
294    };
295
296    publish(topic, &value);
297    Ok(())
298}
299
300fn handle_market_data<T>(
301    topic: MStr<Topic>,
302    encoding: SerializationEncoding,
303    payload: &[u8],
304    deserialize: fn(SerializationEncoding, &[u8]) -> anyhow::Result<Option<T>>,
305    publish: impl FnOnce(MStr<Topic>, &T),
306) -> anyhow::Result<()> {
307    let Some(value) = deserialize(encoding, payload)? else {
308        return Ok(());
309    };
310
311    publish(topic, &value);
312    Ok(())
313}
314
315fn handle_custom_data(
316    topic: MStr<Topic>,
317    payload_type: BusPayloadType,
318    encoding: SerializationEncoding,
319    payload: &[u8],
320) -> anyhow::Result<()> {
321    let Some(custom) = decode_custom_data_payload(payload_type, encoding, payload)? else {
322        return Ok(());
323    };
324
325    publish_any(topic, &custom);
326    Ok(())
327}
328
329fn decode_custom_data_payload(
330    payload_type: BusPayloadType,
331    encoding: SerializationEncoding,
332    payload: &[u8],
333) -> anyhow::Result<Option<CustomData>> {
334    let BusPayloadType::Custom(custom_type_name) = payload_type else {
335        unreachable!("custom data payload decoding requires a custom payload type");
336    };
337
338    if custom_type_name.is_empty() {
339        log::warn!("External payload has no type for inbound republishing");
340        return Ok(None);
341    } else if !payload_type.supports(encoding) {
342        codec::warn_unsupported_inbound(payload_type, encoding);
343        return Ok(None);
344    }
345
346    match encoding {
347        SerializationEncoding::Json => {
348            let value =
349                codec::deserialize_json_payload::<serde_json::Value>(payload, "CustomData")?;
350            decode_custom_data_value(custom_type_name, &value)
351                .context("failed to decode JSON CustomData")
352        }
353        SerializationEncoding::MsgPack => {
354            let value =
355                codec::deserialize_msgpack_payload::<serde_json::Value>(payload, "CustomData")?;
356            decode_custom_data_value(custom_type_name, &value)
357                .context("failed to decode MsgPack CustomData")
358        }
359        SerializationEncoding::Sbe | SerializationEncoding::Capnp => {
360            codec::warn_unsupported_inbound(payload_type, encoding);
361            Ok(None)
362        }
363    }
364}
365
366fn decode_custom_data_value(
367    custom_type_name: Ustr,
368    value: &serde_json::Value,
369) -> anyhow::Result<Option<CustomData>> {
370    let Some(data) = deserialize_custom_from_json(custom_type_name.as_str(), value)? else {
371        log::warn!(
372            "External custom payload type '{custom_type_name}' is not registered for inbound republishing"
373        );
374        return Ok(None);
375    };
376
377    let envelope_type_name = value
378        .get("type")
379        .and_then(serde_json::Value::as_str)
380        .context("CustomData JSON missing 'type' field")?;
381    anyhow::ensure!(
382        envelope_type_name == custom_type_name.as_str(),
383        "CustomData envelope type '{envelope_type_name}' does not match message type '{custom_type_name}'"
384    );
385
386    let Data::Custom(custom) = data else {
387        anyhow::bail!("CustomData registry returned non-custom data");
388    };
389
390    Ok(Some(custom))
391}
nautilus_common/msgbus/external/mod.rs

nautilus_common/msgbus/external/
mod.rs
