nautilus_common/msgbus/
message.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
16use std::fmt::Display;
17
18use bytes::Bytes;
19use nautilus_core::UUID4;
20use nautilus_model::identifiers::TraderId;
21use serde::{Deserialize, Serialize, de::Error as _};
22use ustr::Ustr;
23
24use super::switchboard::CLOSE_TOPIC;
25use crate::enums::SerializationEncoding;
26
27/// External message bus payload category used to select category-level encodings.
28#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
29pub enum BusPayloadCategory {
30    MarketData,
31    BuiltIn,
32    Other,
33}
34
35/// Identifies the publisher and position of an external stream delivery.
36#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
37pub struct BusMessageSource {
38    /// Stable identity of the external publisher.
39    pub publisher_id: TraderId,
40    /// Identity of the publisher process lifetime.
41    pub instance_id: UUID4,
42    /// Publisher-owned stream carrying the message.
43    pub stream: Ustr,
44    /// Position within that stream and publisher lifetime.
45    pub sequence: u64,
46}
47
48/// The payload type carried by a [`BusMessage`].
49///
50/// The fixed variants cover every type the bus publishes externally; [`BusPayloadType::Custom`]
51/// carries the user-defined type name for arbitrary custom data. Untagged variants serialize as
52/// flat name strings such as `"QuoteTick"`. Typed message variants serialize as
53/// `{"typed":"<name>"}` so they remain distinct from custom payloads with the same name. Redis
54/// carries the discriminator as `payload_kind=typed` next to its flat `type` field. Without it,
55/// [`BusPayloadType::from_name`] resolves a typed name as [`BusPayloadType::Custom`].
56#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
57pub enum BusPayloadType {
58    /// User-defined custom data, identified by its type name.
59    Custom(Ustr),
60    Instrument,
61    OrderBookDeltas,
62    OrderBookDepth,
63    QuoteTick,
64    TradeTick,
65    Bar,
66    MarkPriceUpdate,
67    IndexPriceUpdate,
68    FundingRateUpdate,
69    OptionGreeks,
70    AccountState,
71    OrderEvent,
72    PositionEvent,
73    PortfolioSnapshot,
74    SubscribeCommand,
75    UnsubscribeCommand,
76    RequestCommand,
77    DataResponse,
78    DataResponseEnvelope,
79    TradingCommand,
80    GenerateExecutionMassStatus,
81    OrderStatusReport,
82    FillReport,
83    PositionStatusReport,
84    ExecutionMassStatus,
85    #[cfg(feature = "defi")]
86    Block,
87    #[cfg(feature = "defi")]
88    Pool,
89    #[cfg(feature = "defi")]
90    PoolLiquidityUpdate,
91    #[cfg(feature = "defi")]
92    PoolFeeCollect,
93    #[cfg(feature = "defi")]
94    PoolFlash,
95}
96
97impl BusPayloadType {
98    pub(crate) const PUBLISHED_TYPES: &'static [Self] = &[
99        Self::Instrument,
100        Self::OrderBookDeltas,
101        Self::OrderBookDepth,
102        Self::QuoteTick,
103        Self::TradeTick,
104        Self::Bar,
105        Self::MarkPriceUpdate,
106        Self::IndexPriceUpdate,
107        Self::FundingRateUpdate,
108        Self::OptionGreeks,
109        Self::AccountState,
110        Self::OrderEvent,
111        Self::PositionEvent,
112        Self::PortfolioSnapshot,
113        Self::SubscribeCommand,
114        Self::UnsubscribeCommand,
115        Self::RequestCommand,
116        Self::DataResponse,
117        Self::DataResponseEnvelope,
118        Self::TradingCommand,
119        Self::GenerateExecutionMassStatus,
120        Self::OrderStatusReport,
121        Self::FillReport,
122        Self::PositionStatusReport,
123        Self::ExecutionMassStatus,
124        #[cfg(feature = "defi")]
125        Self::Block,
126        #[cfg(feature = "defi")]
127        Self::Pool,
128        #[cfg(feature = "defi")]
129        Self::PoolLiquidityUpdate,
130        #[cfg(feature = "defi")]
131        Self::PoolFeeCollect,
132        #[cfg(feature = "defi")]
133        Self::PoolFlash,
134    ];
135
136    /// Returns the canonical type name for this payload type.
137    ///
138    /// The name alone does not identify variants for which [`Self::is_typed_message`] returns
139    /// `true`.
140    #[must_use]
141    pub fn as_str(&self) -> &str {
142        match self {
143            Self::Custom(type_name) => type_name.as_str(),
144            Self::Instrument => "InstrumentAny",
145            Self::OrderBookDeltas => "OrderBookDeltas",
146            Self::OrderBookDepth => "OrderBookDepth",
147            Self::QuoteTick => "QuoteTick",
148            Self::TradeTick => "TradeTick",
149            Self::Bar => "Bar",
150            Self::MarkPriceUpdate => "MarkPriceUpdate",
151            Self::IndexPriceUpdate => "IndexPriceUpdate",
152            Self::FundingRateUpdate => "FundingRateUpdate",
153            Self::OptionGreeks => "OptionGreeks",
154            Self::AccountState => "AccountState",
155            Self::OrderEvent => "OrderEventAny",
156            Self::PositionEvent => "PositionEvent",
157            Self::PortfolioSnapshot => "PortfolioSnapshot",
158            Self::SubscribeCommand => "SubscribeCommand",
159            Self::UnsubscribeCommand => "UnsubscribeCommand",
160            Self::RequestCommand => "RequestCommand",
161            Self::DataResponse => "DataResponse",
162            Self::DataResponseEnvelope => "DataResponseEnvelope",
163            Self::TradingCommand => "TradingCommand",
164            Self::GenerateExecutionMassStatus => "GenerateExecutionMassStatus",
165            Self::OrderStatusReport => "OrderStatusReport",
166            Self::FillReport => "FillReport",
167            Self::PositionStatusReport => "PositionStatusReport",
168            Self::ExecutionMassStatus => "ExecutionMassStatus",
169            #[cfg(feature = "defi")]
170            Self::Block => "Block",
171            #[cfg(feature = "defi")]
172            Self::Pool => "Pool",
173            #[cfg(feature = "defi")]
174            Self::PoolLiquidityUpdate => "PoolLiquidityUpdate",
175            #[cfg(feature = "defi")]
176            Self::PoolFeeCollect => "PoolFeeCollect",
177            #[cfg(feature = "defi")]
178            Self::PoolFlash => "PoolFlash",
179        }
180    }
181
182    /// Resolves an untagged canonical type name to a [`BusPayloadType`].
183    ///
184    /// Typed message and unknown names resolve to [`BusPayloadType::Custom`]. Use
185    /// [`BusPayloadType::from_typed_name`] when a separate discriminator identifies a typed
186    /// payload.
187    #[must_use]
188    pub fn from_name(name: &str) -> Self {
189        match name {
190            "InstrumentAny" => Self::Instrument,
191            "OrderBookDeltas" => Self::OrderBookDeltas,
192            "OrderBookDepth" => Self::OrderBookDepth,
193            "QuoteTick" => Self::QuoteTick,
194            "TradeTick" => Self::TradeTick,
195            "Bar" => Self::Bar,
196            "MarkPriceUpdate" => Self::MarkPriceUpdate,
197            "IndexPriceUpdate" => Self::IndexPriceUpdate,
198            "FundingRateUpdate" => Self::FundingRateUpdate,
199            "OptionGreeks" => Self::OptionGreeks,
200            "AccountState" => Self::AccountState,
201            "OrderEventAny" => Self::OrderEvent,
202            "PositionEvent" => Self::PositionEvent,
203            "PortfolioSnapshot" => Self::PortfolioSnapshot,
204            #[cfg(feature = "defi")]
205            "Block" => Self::Block,
206            #[cfg(feature = "defi")]
207            "Pool" => Self::Pool,
208            #[cfg(feature = "defi")]
209            "PoolLiquidityUpdate" => Self::PoolLiquidityUpdate,
210            #[cfg(feature = "defi")]
211            "PoolFeeCollect" => Self::PoolFeeCollect,
212            #[cfg(feature = "defi")]
213            "PoolFlash" => Self::PoolFlash,
214            other => Self::Custom(Ustr::from(other)),
215        }
216    }
217
218    /// Resolves a discriminated typed name to its fixed payload type.
219    ///
220    /// Returns `None` when the name does not identify a typed message variant.
221    #[must_use]
222    pub fn from_typed_name(name: &str) -> Option<Self> {
223        match name {
224            "SubscribeCommand" => Some(Self::SubscribeCommand),
225            "UnsubscribeCommand" => Some(Self::UnsubscribeCommand),
226            "RequestCommand" => Some(Self::RequestCommand),
227            "DataResponse" => Some(Self::DataResponse),
228            "DataResponseEnvelope" => Some(Self::DataResponseEnvelope),
229            "TradingCommand" => Some(Self::TradingCommand),
230            "GenerateExecutionMassStatus" => Some(Self::GenerateExecutionMassStatus),
231            "OrderStatusReport" => Some(Self::OrderStatusReport),
232            "FillReport" => Some(Self::FillReport),
233            "PositionStatusReport" => Some(Self::PositionStatusReport),
234            "ExecutionMassStatus" => Some(Self::ExecutionMassStatus),
235            _ => None,
236        }
237    }
238
239    /// Returns whether this payload type uses the typed discriminator.
240    #[must_use]
241    pub const fn is_typed_message(self) -> bool {
242        matches!(
243            self,
244            Self::SubscribeCommand
245                | Self::UnsubscribeCommand
246                | Self::RequestCommand
247                | Self::DataResponse
248                | Self::DataResponseEnvelope
249                | Self::TradingCommand
250                | Self::GenerateExecutionMassStatus
251                | Self::OrderStatusReport
252                | Self::FillReport
253                | Self::PositionStatusReport
254                | Self::ExecutionMassStatus
255        )
256    }
257
258    /// Returns the encoding policy category for this payload type.
259    #[must_use]
260    pub fn category(&self) -> BusPayloadCategory {
261        if self.has_bus_binary_schema() {
262            BusPayloadCategory::MarketData
263        } else {
264            match self {
265                Self::AccountState
266                | Self::OrderEvent
267                | Self::PositionEvent
268                | Self::PortfolioSnapshot => BusPayloadCategory::BuiltIn,
269                _ => BusPayloadCategory::Other,
270            }
271        }
272    }
273
274    /// Returns whether this payload type supports the given bus serialization encoding.
275    #[must_use]
276    pub fn supports(&self, encoding: SerializationEncoding) -> bool {
277        match encoding {
278            SerializationEncoding::Json | SerializationEncoding::MsgPack => true,
279            SerializationEncoding::Sbe => cfg!(feature = "sbe") && self.has_bus_binary_schema(),
280        }
281    }
282
283    fn has_bus_binary_schema(&self) -> bool {
284        matches!(
285            self,
286            Self::OrderBookDeltas
287                | Self::OrderBookDepth
288                | Self::QuoteTick
289                | Self::TradeTick
290                | Self::Bar
291                | Self::MarkPriceUpdate
292                | Self::IndexPriceUpdate
293                | Self::FundingRateUpdate
294                | Self::OptionGreeks
295        )
296    }
297}
298
299impl Display for BusPayloadType {
300    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
301        f.write_str(self.as_str())
302    }
303}
304
305impl Serialize for BusPayloadType {
306    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
307        if self.is_typed_message() {
308            TypedPayloadTypeRef {
309                typed: self.as_str(),
310            }
311            .serialize(serializer)
312        } else {
313            serializer.serialize_str(self.as_str())
314        }
315    }
316}
317
318impl<'de> Deserialize<'de> for BusPayloadType {
319    fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
320        match BusPayloadTypeRepr::deserialize(deserializer)? {
321            BusPayloadTypeRepr::Name(name) => Ok(Self::from_name(&name)),
322            BusPayloadTypeRepr::Typed(value) => Self::from_typed_name(&value.typed)
323                .ok_or_else(|| D::Error::custom(format!("unknown typed payload: {}", value.typed))),
324        }
325    }
326}
327
328#[derive(Serialize)]
329struct TypedPayloadTypeRef<'a> {
330    typed: &'a str,
331}
332
333#[derive(Deserialize)]
334#[serde(deny_unknown_fields)]
335struct TypedPayloadType {
336    typed: String,
337}
338
339#[derive(Deserialize)]
340#[serde(untagged)]
341enum BusPayloadTypeRepr {
342    Name(String),
343    Typed(TypedPayloadType),
344}
345
346/// Represents a bus message including a topic and serialized payload.
347///
348/// Control messages (such as `CLOSE`) that carry no typed payload use an empty
349/// [`BusPayloadType::Custom`].
350#[derive(Clone, Debug, Serialize, Deserialize)]
351#[cfg_attr(
352    feature = "python",
353    pyo3::pyclass(module = "nautilus_trader.common", from_py_object)
354)]
355#[cfg_attr(
356    feature = "python",
357    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.common")
358)]
359pub struct BusMessage {
360    /// The topic to publish the message on.
361    pub topic: Ustr,
362    /// The payload type, carried out-of-band so the receiver can dispatch the serialized payload
363    /// without parsing the topic or inspecting the bytes.
364    pub payload_type: BusPayloadType,
365    /// The serialized payload for the message.
366    pub payload: Bytes,
367    /// The encoding the `payload` is serialized with, so the receiver can decode it without
368    /// relying on its own configuration (mirrors a wire `content-type`).
369    pub encoding: SerializationEncoding,
370    /// Publisher provenance supplied by the external backing, when available.
371    #[serde(default, skip_serializing_if = "Option::is_none")]
372    pub source: Option<BusMessageSource>,
373}
374
375impl BusMessage {
376    /// Creates a new [`BusMessage`] instance.
377    pub fn new(
378        topic: Ustr,
379        payload_type: BusPayloadType,
380        payload: Bytes,
381        encoding: SerializationEncoding,
382    ) -> Self {
383        debug_assert!(!topic.is_empty());
384
385        Self {
386            topic,
387            payload_type,
388            payload,
389            encoding,
390            source: None,
391        }
392    }
393
394    /// Creates a new [`BusMessage`] instance with a string-like topic.
395    ///
396    /// This is a convenience constructor that converts any string-like type
397    /// (implementing `AsRef<str>`) into the required `Ustr` type.
398    pub fn with_str_topic<T: AsRef<str>>(
399        topic: T,
400        payload_type: BusPayloadType,
401        payload: Bytes,
402        encoding: SerializationEncoding,
403    ) -> Self {
404        Self::new(Ustr::from(topic.as_ref()), payload_type, payload, encoding)
405    }
406
407    /// Creates a new [`BusMessage`] instance with the `CLOSE` topic and empty payload.
408    pub fn new_close() -> Self {
409        Self::with_str_topic(
410            CLOSE_TOPIC,
411            BusPayloadType::Custom(Ustr::default()),
412            Bytes::new(),
413            SerializationEncoding::default(),
414        )
415    }
416}
417
418impl Display for BusMessage {
419    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
420        write!(
421            f,
422            "[{}] {} {} {}",
423            self.topic,
424            self.payload_type.as_str(),
425            String::from_utf8_lossy(&self.payload),
426            self.encoding
427        )
428    }
429}
430
431#[cfg(test)]
432mod tests {
433    use bytes::Bytes;
434    use rstest::rstest;
435
436    use super::*;
437
438    #[rstest]
439    #[case(false)]
440    #[case(true)]
441    fn test_source_round_trip_and_legacy_message_compatibility(#[case] with_source: bool) {
442        let mut message = BusMessage::with_str_topic(
443            "external.responses",
444            BusPayloadType::DataResponse,
445            Bytes::from_static(b"reply"),
446            SerializationEncoding::MsgPack,
447        );
448
449        if with_source {
450            message.source = Some(BusMessageSource {
451                publisher_id: "PUBLISHER-017".into(),
452                instance_id: UUID4::new(),
453                stream: "snapshot".into(),
454                sequence: 37,
455            });
456        }
457
458        let encoded = serde_json::to_value(&message).unwrap();
459        let decoded: BusMessage = serde_json::from_value(encoded.clone()).unwrap();
460
461        assert_eq!(encoded.get("source").is_some(), with_source);
462        assert_eq!(decoded.topic, message.topic);
463        assert_eq!(decoded.payload_type, message.payload_type);
464        assert_eq!(decoded.payload, message.payload);
465        assert_eq!(decoded.encoding, message.encoding);
466        assert_eq!(decoded.source, message.source);
467    }
468
469    #[rstest]
470    #[case("test/topic", "payload data")]
471    #[case("events/trading", "Another payload")]
472    fn test_with_str_topic_str(#[case] topic: &str, #[case] payload_str: &str) {
473        let payload = Bytes::from(payload_str.to_string());
474
475        let message = BusMessage::with_str_topic(
476            topic,
477            BusPayloadType::QuoteTick,
478            payload.clone(),
479            SerializationEncoding::Json,
480        );
481
482        assert_eq!(message.topic, topic);
483        assert_eq!(message.payload_type, BusPayloadType::QuoteTick);
484        assert_eq!(message.encoding, SerializationEncoding::Json);
485        assert_eq!(message.payload, payload);
486    }
487
488    #[rstest]
489    fn test_with_str_topic_string() {
490        let topic_string = String::from("orders/new");
491        let payload = Bytes::from("order payload data");
492
493        let message = BusMessage::with_str_topic(
494            topic_string.clone(),
495            BusPayloadType::OrderEvent,
496            payload.clone(),
497            SerializationEncoding::MsgPack,
498        );
499
500        assert_eq!(message.topic, topic_string);
501        assert_eq!(message.payload_type, BusPayloadType::OrderEvent);
502        assert_eq!(message.encoding, SerializationEncoding::MsgPack);
503        assert_eq!(message.payload, payload);
504    }
505
506    #[rstest]
507    fn test_new_close() {
508        let message = BusMessage::new_close();
509
510        assert_eq!(message.topic, "CLOSE");
511        assert!(message.payload.is_empty());
512    }
513
514    #[rstest]
515    #[case(BusPayloadType::QuoteTick, BusPayloadCategory::MarketData)]
516    #[case(BusPayloadType::OrderBookDeltas, BusPayloadCategory::MarketData)]
517    #[case(BusPayloadType::AccountState, BusPayloadCategory::BuiltIn)]
518    #[case(BusPayloadType::OrderEvent, BusPayloadCategory::BuiltIn)]
519    #[case(BusPayloadType::Instrument, BusPayloadCategory::Other)]
520    #[case(BusPayloadType::OptionGreeks, BusPayloadCategory::MarketData)]
521    #[case(
522        BusPayloadType::Custom(Ustr::from("CustomPayload")),
523        BusPayloadCategory::Other
524    )]
525    fn bus_payload_type_category(
526        #[case] payload_type: BusPayloadType,
527        #[case] expected: BusPayloadCategory,
528    ) {
529        assert_eq!(payload_type.category(), expected);
530    }
531
532    #[rstest]
533    #[case(BusPayloadType::QuoteTick, SerializationEncoding::Json, true)]
534    #[case(BusPayloadType::QuoteTick, SerializationEncoding::MsgPack, true)]
535    #[cfg_attr(
536        feature = "sbe",
537        case(BusPayloadType::QuoteTick, SerializationEncoding::Sbe, true)
538    )]
539    #[cfg_attr(
540        not(feature = "sbe"),
541        case(BusPayloadType::QuoteTick, SerializationEncoding::Sbe, false)
542    )]
543    #[case(BusPayloadType::AccountState, SerializationEncoding::Json, true)]
544    #[case(BusPayloadType::Instrument, SerializationEncoding::Sbe, false)]
545    #[cfg_attr(
546        feature = "sbe",
547        case(BusPayloadType::OptionGreeks, SerializationEncoding::Sbe, true)
548    )]
549    #[cfg_attr(
550        not(feature = "sbe"),
551        case(BusPayloadType::OptionGreeks, SerializationEncoding::Sbe, false)
552    )]
553    #[case(
554        BusPayloadType::Custom(Ustr::from("CustomPayload")),
555        SerializationEncoding::MsgPack,
556        true
557    )]
558    #[case(
559        BusPayloadType::Custom(Ustr::from("CustomPayload")),
560        SerializationEncoding::Sbe,
561        false
562    )]
563    fn bus_payload_type_supports_encoding(
564        #[case] payload_type: BusPayloadType,
565        #[case] encoding: SerializationEncoding,
566        #[case] expected: bool,
567    ) {
568        assert_eq!(payload_type.supports(encoding), expected);
569    }
570
571    #[rstest]
572    fn bus_payload_type_accepts_canonical_depth_name() {
573        assert_eq!(
574            BusPayloadType::from_name("OrderBookDepth"),
575            BusPayloadType::OrderBookDepth
576        );
577        assert_eq!(
578            BusPayloadType::from_name("OrderBookDepth10"),
579            BusPayloadType::Custom(Ustr::from("OrderBookDepth10")),
580        );
581    }
582
583    #[rstest]
584    fn typed_payload_types_use_json_msgpack_policy() {
585        let payload_types = [
586            (BusPayloadType::SubscribeCommand, "SubscribeCommand"),
587            (BusPayloadType::UnsubscribeCommand, "UnsubscribeCommand"),
588            (BusPayloadType::RequestCommand, "RequestCommand"),
589            (BusPayloadType::DataResponse, "DataResponse"),
590            (BusPayloadType::TradingCommand, "TradingCommand"),
591            (
592                BusPayloadType::GenerateExecutionMassStatus,
593                "GenerateExecutionMassStatus",
594            ),
595            (BusPayloadType::OrderStatusReport, "OrderStatusReport"),
596            (BusPayloadType::FillReport, "FillReport"),
597            (BusPayloadType::PositionStatusReport, "PositionStatusReport"),
598            (BusPayloadType::ExecutionMassStatus, "ExecutionMassStatus"),
599        ];
600
601        for (payload_type, name) in payload_types {
602            let custom_payload_type = BusPayloadType::Custom(Ustr::from(name));
603            let typed_json =
604                serde_json::to_value(payload_type).expect("typed payload type must serialize");
605            let custom_json = serde_json::to_value(custom_payload_type)
606                .expect("custom payload type must serialize");
607
608            assert_eq!(payload_type.as_str(), name);
609            assert_eq!(BusPayloadType::from_name(name), custom_payload_type);
610            assert_eq!(BusPayloadType::from_typed_name(name), Some(payload_type));
611            assert_eq!(typed_json, serde_json::json!({ "typed": name }));
612            assert_eq!(custom_json, serde_json::json!(name));
613            assert_eq!(
614                serde_json::from_value::<BusPayloadType>(typed_json)
615                    .expect("typed payload type must deserialize"),
616                payload_type
617            );
618            assert_eq!(
619                serde_json::from_value::<BusPayloadType>(custom_json)
620                    .expect("custom payload type must deserialize"),
621                custom_payload_type
622            );
623            assert!(payload_type.is_typed_message());
624            assert!(!custom_payload_type.is_typed_message());
625            assert_eq!(payload_type.category(), BusPayloadCategory::Other);
626            assert!(payload_type.supports(SerializationEncoding::Json));
627            assert!(payload_type.supports(SerializationEncoding::MsgPack));
628            assert!(!payload_type.supports(SerializationEncoding::Sbe));
629        }
630    }
631}
nautilus_common/msgbus/message.rs

nautilus_common/msgbus/
message.rs
