nautilus_common/msgbus/
config.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 serde::{Deserialize, Serialize};
17use ustr::Ustr;
18
19use super::{BusPayloadCategory, BusPayloadType};
20use crate::{
21    config::{ConfigError, ConfigErrorCollector, ConfigResult},
22    enums::SerializationEncoding,
23};
24
25/// Configuration for `MessageBus` instances.
26#[cfg_attr(
27    feature = "python",
28    pyo3::pyclass(module = "nautilus_trader.core.nautilus_pyo3.common", from_py_object)
29)]
30#[cfg_attr(
31    feature = "python",
32    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.common")
33)]
34#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, bon::Builder)]
35#[serde(default, deny_unknown_fields)]
36pub struct MessageBusConfig {
37    /// The default encoding for externally published payloads.
38    #[builder(default = SerializationEncoding::Json)]
39    pub encoding: SerializationEncoding,
40    /// The encoding for market data payloads supported by the external bus binary codecs.
41    pub encoding_market_data: Option<SerializationEncoding>,
42    /// The encoding for built-in account, portfolio, order, and position payloads.
43    pub encoding_builtin: Option<SerializationEncoding>,
44    /// If timestamps should be persisted as ISO 8601 strings.
45    /// If `false`, then timestamps will be persisted as UNIX nanoseconds.
46    #[builder(default)]
47    pub timestamps_as_iso8601: bool,
48    /// The buffer interval (milliseconds) between pipelined/batched transactions.
49    /// The recommended range if using buffered pipelining is [10, 1000] milliseconds,
50    /// with a good compromise being 100 milliseconds.
51    pub buffer_interval_ms: Option<u32>,
52    /// The lookback window in minutes for automatic stream trimming.
53    /// The actual window may extend up to one minute beyond the specified value since streams are trimmed at most once every minute.
54    /// This feature requires Redis version 6.2 or higher; otherwise, it will result in a command syntax error.
55    pub autotrim_mins: Option<u32>,
56    /// If a 'trader-' prefix is used for stream names.
57    #[builder(default = true)]
58    pub use_trader_prefix: bool,
59    /// If the trader's ID is used for stream names.
60    #[builder(default = true)]
61    pub use_trader_id: bool,
62    /// If the trader's instance ID is used for stream names. Default is `false`.
63    #[builder(default)]
64    pub use_instance_id: bool,
65    /// The prefix for externally published stream names.
66    #[builder(default = "stream".to_string())]
67    pub streams_prefix: String,
68    /// If `true`, messages will be written to separate streams per topic.
69    /// If `false`, all messages will be written to the same stream.
70    #[builder(default = true)]
71    pub stream_per_topic: bool,
72    /// The external stream keys the message bus will listen to for publishing deserialized message payloads internally.
73    pub external_streams: Option<Vec<String>>,
74    /// A list of serializable types **not** to publish externally.
75    pub types_filter: Option<Vec<String>>,
76    /// The heartbeat interval (seconds).
77    pub heartbeat_interval_secs: Option<u16>,
78}
79
80impl Default for MessageBusConfig {
81    fn default() -> Self {
82        Self::builder().build()
83    }
84}
85
86impl MessageBusConfig {
87    /// Validates external message bus encoding policy.
88    ///
89    /// # Errors
90    ///
91    /// Returns a [`ConfigError`] when the default encoding cannot carry custom payloads, or when
92    /// a category override selects an encoding unsupported by any payload type in that category.
93    pub fn validate(&self) -> ConfigResult<()> {
94        let mut errors = ConfigErrorCollector::new();
95
96        if !BusPayloadType::Custom(Ustr::from("Custom")).supports(self.encoding) {
97            errors.push(ConfigError::unsupported_value(
98                "MessageBusConfig.encoding",
99                format!(
100                    "{} does not support custom or unmapped payloads",
101                    self.encoding
102                ),
103            ));
104        }
105
106        if let Some(encoding) = self.encoding_market_data {
107            validate_category_encoding(
108                &mut errors,
109                "MessageBusConfig.encoding_market_data",
110                BusPayloadCategory::MarketData,
111                encoding,
112            );
113        }
114
115        if let Some(encoding) = self.encoding_builtin {
116            validate_category_encoding(
117                &mut errors,
118                "MessageBusConfig.encoding_builtin",
119                BusPayloadCategory::BuiltIn,
120                encoding,
121            );
122        }
123
124        errors.into_result()
125    }
126}
127
128fn validate_category_encoding(
129    errors: &mut ConfigErrorCollector,
130    field: &'static str,
131    category: BusPayloadCategory,
132    encoding: SerializationEncoding,
133) {
134    let unsupported = BusPayloadType::PUBLISHED_TYPES
135        .iter()
136        .copied()
137        .filter(|payload_type| payload_type.category() == category)
138        .filter(|payload_type| !payload_type.supports(encoding))
139        .map(|payload_type| payload_type.as_str().to_string())
140        .collect::<Vec<_>>();
141
142    if unsupported.is_empty() {
143        return;
144    }
145
146    errors.push(ConfigError::unsupported_value(
147        field,
148        format!(
149            "{} is not supported by {}",
150            encoding,
151            unsupported.join(", ")
152        ),
153    ));
154}
155
156#[cfg(test)]
157mod tests {
158    use rstest::*;
159    use serde_json::json;
160
161    use super::*;
162    use crate::{config::ConfigError, enums::SerializationEncoding};
163
164    #[rstest]
165    fn test_default_message_bus_config() {
166        let config = MessageBusConfig::default();
167        assert_eq!(config.encoding, SerializationEncoding::Json);
168        assert_eq!(config.encoding_market_data, None);
169        assert_eq!(config.encoding_builtin, None);
170        assert!(!config.timestamps_as_iso8601);
171        assert_eq!(config.buffer_interval_ms, None);
172        assert_eq!(config.autotrim_mins, None);
173        assert!(config.use_trader_prefix);
174        assert!(config.use_trader_id);
175        assert!(!config.use_instance_id);
176        assert_eq!(config.streams_prefix, "stream");
177        assert!(config.stream_per_topic);
178        assert_eq!(config.external_streams, None);
179        assert_eq!(config.types_filter, None);
180    }
181
182    #[rstest]
183    fn test_deserialize_message_bus_config() {
184        let config_json = json!({
185            "encoding": "json",
186            "encoding_market_data": "sbe",
187            "encoding_builtin": "msgpack",
188            "timestamps_as_iso8601": true,
189            "buffer_interval_ms": 100,
190            "autotrim_mins": 60,
191            "use_trader_prefix": false,
192            "use_trader_id": false,
193            "use_instance_id": true,
194            "streams_prefix": "data_streams",
195            "stream_per_topic": false,
196            "external_streams": ["stream1", "stream2"],
197            "types_filter": ["type1", "type2"]
198        });
199        let config: MessageBusConfig = serde_json::from_value(config_json).unwrap();
200        assert_eq!(config.encoding, SerializationEncoding::Json);
201        assert_eq!(
202            config.encoding_market_data,
203            Some(SerializationEncoding::Sbe)
204        );
205        assert_eq!(
206            config.encoding_builtin,
207            Some(SerializationEncoding::MsgPack)
208        );
209        assert!(config.timestamps_as_iso8601);
210        assert_eq!(config.buffer_interval_ms, Some(100));
211        assert_eq!(config.autotrim_mins, Some(60));
212        assert!(!config.use_trader_prefix);
213        assert!(!config.use_trader_id);
214        assert!(config.use_instance_id);
215        assert_eq!(config.streams_prefix, "data_streams");
216        assert!(!config.stream_per_topic);
217        assert_eq!(
218            config.external_streams,
219            Some(vec!["stream1".to_string(), "stream2".to_string()])
220        );
221        assert_eq!(
222            config.types_filter,
223            Some(vec!["type1".to_string(), "type2".to_string()])
224        );
225    }
226
227    #[rstest]
228    fn test_deserialize_message_bus_config_rejects_backing_field() {
229        let config_json = json!({
230            "backing": {},
231        });
232
233        let error = serde_json::from_value::<MessageBusConfig>(config_json).unwrap_err();
234        assert!(error.to_string().contains("unknown field `backing`"));
235    }
236
237    #[rstest]
238    #[case("sbe", SerializationEncoding::Sbe)]
239    #[case("capnp", SerializationEncoding::Capnp)]
240    fn test_deserialize_message_bus_config_with_schema_encoding(
241        #[case] encoding_name: &str,
242        #[case] expected: SerializationEncoding,
243    ) {
244        let config_json = json!({
245            "encoding": encoding_name,
246        });
247
248        let config: MessageBusConfig = serde_json::from_value(config_json).unwrap();
249        assert_eq!(config.encoding, expected);
250    }
251
252    #[rstest]
253    fn message_bus_config_validate_accepts_default() {
254        let config = MessageBusConfig::default();
255
256        assert!(config.validate().is_ok());
257    }
258
259    #[rstest]
260    #[case(SerializationEncoding::Json)]
261    #[case(SerializationEncoding::MsgPack)]
262    fn message_bus_config_validate_accepts_custom_safe_default(
263        #[case] encoding: SerializationEncoding,
264    ) {
265        let config = MessageBusConfig {
266            encoding,
267            ..Default::default()
268        };
269
270        assert!(config.validate().is_ok());
271    }
272
273    #[rstest]
274    #[case(SerializationEncoding::Sbe)]
275    #[case(SerializationEncoding::Capnp)]
276    fn message_bus_config_validate_rejects_schema_default(#[case] encoding: SerializationEncoding) {
277        let config = MessageBusConfig {
278            encoding,
279            ..Default::default()
280        };
281
282        let error = config.validate().unwrap_err();
283        assert!(matches!(
284            error,
285            ConfigError::UnsupportedValue { field, .. }
286                if field == "MessageBusConfig.encoding"
287        ));
288    }
289
290    #[cfg(any(feature = "sbe", feature = "capnp"))]
291    #[rstest]
292    #[cfg_attr(feature = "sbe", case(SerializationEncoding::Sbe))]
293    #[cfg_attr(feature = "capnp", case(SerializationEncoding::Capnp))]
294    fn message_bus_config_validate_accepts_market_data_override(
295        #[case] encoding: SerializationEncoding,
296    ) {
297        let config = MessageBusConfig {
298            encoding_market_data: Some(encoding),
299            ..Default::default()
300        };
301
302        assert!(config.validate().is_ok());
303    }
304
305    #[cfg(not(feature = "sbe"))]
306    #[rstest]
307    fn message_bus_config_validate_rejects_market_data_sbe_without_feature() {
308        let config = MessageBusConfig {
309            encoding_market_data: Some(SerializationEncoding::Sbe),
310            ..Default::default()
311        };
312
313        let error = config.validate().unwrap_err();
314        assert!(matches!(
315            error,
316            ConfigError::UnsupportedValue { field, .. }
317                if field == "MessageBusConfig.encoding_market_data"
318        ));
319    }
320
321    #[cfg(not(feature = "capnp"))]
322    #[rstest]
323    fn message_bus_config_validate_rejects_market_data_capnp_without_feature() {
324        let config = MessageBusConfig {
325            encoding_market_data: Some(SerializationEncoding::Capnp),
326            ..Default::default()
327        };
328
329        let error = config.validate().unwrap_err();
330        assert!(matches!(
331            error,
332            ConfigError::UnsupportedValue { field, .. }
333                if field == "MessageBusConfig.encoding_market_data"
334        ));
335    }
336
337    #[rstest]
338    #[case(SerializationEncoding::Sbe)]
339    #[case(SerializationEncoding::Capnp)]
340    fn message_bus_config_validate_rejects_builtin_schema_override(
341        #[case] encoding: SerializationEncoding,
342    ) {
343        let config = MessageBusConfig {
344            encoding_builtin: Some(encoding),
345            ..Default::default()
346        };
347
348        let error = config.validate().unwrap_err();
349        assert!(matches!(
350            error,
351            ConfigError::UnsupportedValue { field, .. }
352                if field == "MessageBusConfig.encoding_builtin"
353        ));
354    }
355}
nautilus_common/msgbus/config.rs

nautilus_common/msgbus/
config.rs
