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};
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. Serializes as its flat name
37/// string (e.g. `"QuoteTick"`), matching the wire `type` field.
38#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
39pub enum BusPayloadType {
40    /// User-defined custom data, identified by its type name.
41    Custom(Ustr),
42    Instrument,
43    OrderBookDeltas,
44    OrderBookDepth10,
45    QuoteTick,
46    TradeTick,
47    Bar,
48    MarkPriceUpdate,
49    IndexPriceUpdate,
50    FundingRateUpdate,
51    OptionGreeks,
52    AccountState,
53    OrderEvent,
54    PositionEvent,
55    PortfolioSnapshot,
56    #[cfg(feature = "defi")]
57    Block,
58    #[cfg(feature = "defi")]
59    Pool,
60    #[cfg(feature = "defi")]
61    PoolLiquidityUpdate,
62    #[cfg(feature = "defi")]
63    PoolFeeCollect,
64    #[cfg(feature = "defi")]
65    PoolFlash,
66}
67
68impl BusPayloadType {
69    pub(crate) const PUBLISHED_TYPES: &'static [Self] = &[
70        Self::Instrument,
71        Self::OrderBookDeltas,
72        Self::OrderBookDepth10,
73        Self::QuoteTick,
74        Self::TradeTick,
75        Self::Bar,
76        Self::MarkPriceUpdate,
77        Self::IndexPriceUpdate,
78        Self::FundingRateUpdate,
79        Self::OptionGreeks,
80        Self::AccountState,
81        Self::OrderEvent,
82        Self::PositionEvent,
83        Self::PortfolioSnapshot,
84        #[cfg(feature = "defi")]
85        Self::Block,
86        #[cfg(feature = "defi")]
87        Self::Pool,
88        #[cfg(feature = "defi")]
89        Self::PoolLiquidityUpdate,
90        #[cfg(feature = "defi")]
91        Self::PoolFeeCollect,
92        #[cfg(feature = "defi")]
93        Self::PoolFlash,
94    ];
95
96    /// Returns the canonical type name for this payload type.
97    #[must_use]
98    pub fn as_str(&self) -> &str {
99        match self {
100            Self::Custom(type_name) => type_name.as_str(),
101            Self::Instrument => "InstrumentAny",
102            Self::OrderBookDeltas => "OrderBookDeltas",
103            Self::OrderBookDepth10 => "OrderBookDepth10",
104            Self::QuoteTick => "QuoteTick",
105            Self::TradeTick => "TradeTick",
106            Self::Bar => "Bar",
107            Self::MarkPriceUpdate => "MarkPriceUpdate",
108            Self::IndexPriceUpdate => "IndexPriceUpdate",
109            Self::FundingRateUpdate => "FundingRateUpdate",
110            Self::OptionGreeks => "OptionGreeks",
111            Self::AccountState => "AccountState",
112            Self::OrderEvent => "OrderEventAny",
113            Self::PositionEvent => "PositionEvent",
114            Self::PortfolioSnapshot => "PortfolioSnapshot",
115            #[cfg(feature = "defi")]
116            Self::Block => "Block",
117            #[cfg(feature = "defi")]
118            Self::Pool => "Pool",
119            #[cfg(feature = "defi")]
120            Self::PoolLiquidityUpdate => "PoolLiquidityUpdate",
121            #[cfg(feature = "defi")]
122            Self::PoolFeeCollect => "PoolFeeCollect",
123            #[cfg(feature = "defi")]
124            Self::PoolFlash => "PoolFlash",
125        }
126    }
127
128    /// Resolves a canonical type name to a [`BusPayloadType`].
129    ///
130    /// Unknown names resolve to [`BusPayloadType::Custom`].
131    #[must_use]
132    pub fn from_name(name: &str) -> Self {
133        match name {
134            "InstrumentAny" => Self::Instrument,
135            "OrderBookDeltas" => Self::OrderBookDeltas,
136            "OrderBookDepth10" => Self::OrderBookDepth10,
137            "QuoteTick" => Self::QuoteTick,
138            "TradeTick" => Self::TradeTick,
139            "Bar" => Self::Bar,
140            "MarkPriceUpdate" => Self::MarkPriceUpdate,
141            "IndexPriceUpdate" => Self::IndexPriceUpdate,
142            "FundingRateUpdate" => Self::FundingRateUpdate,
143            "OptionGreeks" => Self::OptionGreeks,
144            "AccountState" => Self::AccountState,
145            "OrderEventAny" => Self::OrderEvent,
146            "PositionEvent" => Self::PositionEvent,
147            "PortfolioSnapshot" => Self::PortfolioSnapshot,
148            #[cfg(feature = "defi")]
149            "Block" => Self::Block,
150            #[cfg(feature = "defi")]
151            "Pool" => Self::Pool,
152            #[cfg(feature = "defi")]
153            "PoolLiquidityUpdate" => Self::PoolLiquidityUpdate,
154            #[cfg(feature = "defi")]
155            "PoolFeeCollect" => Self::PoolFeeCollect,
156            #[cfg(feature = "defi")]
157            "PoolFlash" => Self::PoolFlash,
158            other => Self::Custom(Ustr::from(other)),
159        }
160    }
161
162    /// Returns the encoding policy category for this payload type.
163    #[must_use]
164    pub fn category(&self) -> BusPayloadCategory {
165        if self.has_bus_binary_schema() {
166            BusPayloadCategory::MarketData
167        } else {
168            match self {
169                Self::AccountState
170                | Self::OrderEvent
171                | Self::PositionEvent
172                | Self::PortfolioSnapshot => BusPayloadCategory::BuiltIn,
173                _ => BusPayloadCategory::Other,
174            }
175        }
176    }
177
178    /// Returns whether this payload type supports the given bus serialization encoding.
179    #[must_use]
180    pub fn supports(&self, encoding: SerializationEncoding) -> bool {
181        match encoding {
182            SerializationEncoding::Json | SerializationEncoding::MsgPack => true,
183            SerializationEncoding::Sbe => cfg!(feature = "sbe") && self.has_bus_binary_schema(),
184            SerializationEncoding::Capnp => cfg!(feature = "capnp") && self.has_bus_binary_schema(),
185        }
186    }
187
188    fn has_bus_binary_schema(&self) -> bool {
189        matches!(
190            self,
191            Self::OrderBookDeltas
192                | Self::OrderBookDepth10
193                | Self::QuoteTick
194                | Self::TradeTick
195                | Self::Bar
196                | Self::MarkPriceUpdate
197                | Self::IndexPriceUpdate
198                | Self::FundingRateUpdate
199                | Self::OptionGreeks
200        )
201    }
202}
203
204impl Display for BusPayloadType {
205    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
206        f.write_str(self.as_str())
207    }
208}
209
210impl Serialize for BusPayloadType {
211    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
212        serializer.serialize_str(self.as_str())
213    }
214}
215
216impl<'de> Deserialize<'de> for BusPayloadType {
217    fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
218        let name = String::deserialize(deserializer)?;
219        Ok(Self::from_name(&name))
220    }
221}
222
223/// Represents a bus message including a topic and serialized payload.
224///
225/// Control messages (such as `CLOSE`) that carry no typed payload use an empty
226/// [`BusPayloadType::Custom`].
227#[derive(Clone, Debug, Serialize, Deserialize)]
228#[cfg_attr(
229    feature = "python",
230    pyo3::pyclass(module = "nautilus_trader.core.nautilus_pyo3.common", from_py_object)
231)]
232#[cfg_attr(
233    feature = "python",
234    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.common")
235)]
236pub struct BusMessage {
237    /// The topic to publish the message on.
238    pub topic: Ustr,
239    /// The payload type, carried out-of-band so the receiver can dispatch the serialized payload
240    /// without parsing the topic or inspecting the bytes.
241    pub payload_type: BusPayloadType,
242    /// The serialized payload for the message.
243    pub payload: Bytes,
244    /// The encoding the `payload` is serialized with, so the receiver can decode it without
245    /// relying on its own configuration (mirrors a wire `content-type`).
246    pub encoding: SerializationEncoding,
247}
248
249impl BusMessage {
250    /// Creates a new [`BusMessage`] instance.
251    pub fn new(
252        topic: Ustr,
253        payload_type: BusPayloadType,
254        payload: Bytes,
255        encoding: SerializationEncoding,
256    ) -> Self {
257        debug_assert!(!topic.is_empty());
258        Self {
259            topic,
260            payload_type,
261            payload,
262            encoding,
263        }
264    }
265
266    /// Creates a new [`BusMessage`] instance with a string-like topic.
267    ///
268    /// This is a convenience constructor that converts any string-like type
269    /// (implementing `AsRef<str>`) into the required `Ustr` type.
270    pub fn with_str_topic<T: AsRef<str>>(
271        topic: T,
272        payload_type: BusPayloadType,
273        payload: Bytes,
274        encoding: SerializationEncoding,
275    ) -> Self {
276        Self::new(Ustr::from(topic.as_ref()), payload_type, payload, encoding)
277    }
278
279    /// Creates a new [`BusMessage`] instance with the `CLOSE` topic and empty payload.
280    pub fn new_close() -> Self {
281        Self::with_str_topic(
282            CLOSE_TOPIC,
283            BusPayloadType::Custom(Ustr::default()),
284            Bytes::new(),
285            SerializationEncoding::default(),
286        )
287    }
288}
289
290impl Display for BusMessage {
291    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
292        write!(
293            f,
294            "[{}] {} {} {}",
295            self.topic,
296            self.payload_type.as_str(),
297            String::from_utf8_lossy(&self.payload),
298            self.encoding
299        )
300    }
301}
302
303#[cfg(test)]
304mod tests {
305    use bytes::Bytes;
306    use rstest::rstest;
307
308    use super::*;
309
310    #[rstest]
311    #[case("test/topic", "payload data")]
312    #[case("events/trading", "Another payload")]
313    fn test_with_str_topic_str(#[case] topic: &str, #[case] payload_str: &str) {
314        let payload = Bytes::from(payload_str.to_string());
315
316        let message = BusMessage::with_str_topic(
317            topic,
318            BusPayloadType::QuoteTick,
319            payload.clone(),
320            SerializationEncoding::Json,
321        );
322
323        assert_eq!(message.topic.as_str(), topic);
324        assert_eq!(message.payload_type, BusPayloadType::QuoteTick);
325        assert_eq!(message.encoding, SerializationEncoding::Json);
326        assert_eq!(message.payload, payload);
327    }
328
329    #[rstest]
330    fn test_with_str_topic_string() {
331        let topic_string = String::from("orders/new");
332        let payload = Bytes::from("order payload data");
333
334        let message = BusMessage::with_str_topic(
335            topic_string.clone(),
336            BusPayloadType::OrderEvent,
337            payload.clone(),
338            SerializationEncoding::MsgPack,
339        );
340
341        assert_eq!(message.topic.as_str(), topic_string);
342        assert_eq!(message.payload_type, BusPayloadType::OrderEvent);
343        assert_eq!(message.encoding, SerializationEncoding::MsgPack);
344        assert_eq!(message.payload, payload);
345    }
346
347    #[rstest]
348    fn test_new_close() {
349        let message = BusMessage::new_close();
350
351        assert_eq!(message.topic.as_str(), "CLOSE");
352        assert!(message.payload.is_empty());
353    }
354
355    #[rstest]
356    #[case(BusPayloadType::QuoteTick, BusPayloadCategory::MarketData)]
357    #[case(BusPayloadType::OrderBookDeltas, BusPayloadCategory::MarketData)]
358    #[case(BusPayloadType::AccountState, BusPayloadCategory::BuiltIn)]
359    #[case(BusPayloadType::OrderEvent, BusPayloadCategory::BuiltIn)]
360    #[case(BusPayloadType::Instrument, BusPayloadCategory::Other)]
361    #[case(BusPayloadType::OptionGreeks, BusPayloadCategory::MarketData)]
362    #[case(
363        BusPayloadType::Custom(Ustr::from("CustomPayload")),
364        BusPayloadCategory::Other
365    )]
366    fn bus_payload_type_category(
367        #[case] payload_type: BusPayloadType,
368        #[case] expected: BusPayloadCategory,
369    ) {
370        assert_eq!(payload_type.category(), expected);
371    }
372
373    #[rstest]
374    #[case(BusPayloadType::QuoteTick, SerializationEncoding::Json, true)]
375    #[case(BusPayloadType::QuoteTick, SerializationEncoding::MsgPack, true)]
376    #[cfg_attr(
377        feature = "sbe",
378        case(BusPayloadType::QuoteTick, SerializationEncoding::Sbe, true)
379    )]
380    #[cfg_attr(
381        not(feature = "sbe"),
382        case(BusPayloadType::QuoteTick, SerializationEncoding::Sbe, false)
383    )]
384    #[cfg_attr(
385        feature = "capnp",
386        case(BusPayloadType::QuoteTick, SerializationEncoding::Capnp, true)
387    )]
388    #[cfg_attr(
389        not(feature = "capnp"),
390        case(BusPayloadType::QuoteTick, SerializationEncoding::Capnp, false)
391    )]
392    #[case(BusPayloadType::AccountState, SerializationEncoding::Json, true)]
393    #[case(BusPayloadType::AccountState, SerializationEncoding::Capnp, false)]
394    #[case(BusPayloadType::Instrument, SerializationEncoding::Sbe, false)]
395    #[cfg_attr(
396        feature = "sbe",
397        case(BusPayloadType::OptionGreeks, SerializationEncoding::Sbe, true)
398    )]
399    #[cfg_attr(
400        not(feature = "sbe"),
401        case(BusPayloadType::OptionGreeks, SerializationEncoding::Sbe, false)
402    )]
403    #[cfg_attr(
404        feature = "capnp",
405        case(BusPayloadType::OptionGreeks, SerializationEncoding::Capnp, true)
406    )]
407    #[cfg_attr(
408        not(feature = "capnp"),
409        case(BusPayloadType::OptionGreeks, SerializationEncoding::Capnp, false)
410    )]
411    #[case(
412        BusPayloadType::Custom(Ustr::from("CustomPayload")),
413        SerializationEncoding::MsgPack,
414        true
415    )]
416    #[case(
417        BusPayloadType::Custom(Ustr::from("CustomPayload")),
418        SerializationEncoding::Sbe,
419        false
420    )]
421    fn bus_payload_type_supports_encoding(
422        #[case] payload_type: BusPayloadType,
423        #[case] encoding: SerializationEncoding,
424        #[case] expected: bool,
425    ) {
426        assert_eq!(payload_type.supports(encoding), expected);
427    }
428}
nautilus_common/msgbus/message.rs

nautilus_common/msgbus/
message.rs
