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

nautilus_common/msgbus/
message.rs
