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.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    /// The approximate maximum number of entries to retain in each stream.
57    /// Redis may retain slightly more entries than this value to improve trimming performance.
58    pub autotrim_maxlen: Option<u32>,
59    /// If a 'trader-' prefix is used for stream names.
60    #[builder(default = true)]
61    pub use_trader_prefix: bool,
62    /// If the trader's ID is used for stream names.
63    #[builder(default = true)]
64    pub use_trader_id: bool,
65    /// If the trader's instance ID is used for stream names. Default is `false`.
66    #[builder(default)]
67    pub use_instance_id: bool,
68    /// The prefix for externally published stream names.
69    #[builder(default = "stream".to_string())]
70    pub streams_prefix: String,
71    /// If `true`, messages will be written to separate streams per topic.
72    /// If `false`, all messages will be written to the same stream.
73    #[builder(default = true)]
74    pub stream_per_topic: bool,
75    /// The external stream keys the message bus will listen to for publishing deserialized message payloads internally.
76    pub external_streams: Option<Vec<String>>,
77    /// A list of serializable payload type names **not** to publish externally.
78    pub types_filter: Option<Vec<String>>,
79    /// The heartbeat interval (seconds).
80    pub heartbeat_interval_secs: Option<u16>,
81}
82
83impl Default for MessageBusConfig {
84    fn default() -> Self {
85        Self::builder().build()
86    }
87}
88
89impl MessageBusConfig {
90    /// Validates external message bus encoding policy.
91    ///
92    /// # Errors
93    ///
94    /// Returns an error if the default encoding cannot carry custom payloads, or when a category
95    /// override selects an encoding unsupported by any payload type in that category.
96    pub fn validate(&self) -> ConfigResult<()> {
97        let mut errors = ConfigErrorCollector::new();
98
99        if !BusPayloadType::Custom(Ustr::from("Custom")).supports(self.encoding) {
100            errors.push(ConfigError::unsupported_value(
101                "MessageBusConfig.encoding",
102                format!(
103                    "{} does not support custom or unmapped payloads",
104                    self.encoding
105                ),
106            ));
107        }
108
109        if let Some(encoding) = self.encoding_market_data {
110            validate_category_encoding(
111                &mut errors,
112                "MessageBusConfig.encoding_market_data",
113                BusPayloadCategory::MarketData,
114                encoding,
115            );
116        }
117
118        if let Some(encoding) = self.encoding_builtin {
119            validate_category_encoding(
120                &mut errors,
121                "MessageBusConfig.encoding_builtin",
122                BusPayloadCategory::BuiltIn,
123                encoding,
124            );
125        }
126
127        errors.into_result()
128    }
129}
130
131fn validate_category_encoding(
132    errors: &mut ConfigErrorCollector,
133    field: &'static str,
134    category: BusPayloadCategory,
135    encoding: SerializationEncoding,
136) {
137    let unsupported = BusPayloadType::PUBLISHED_TYPES
138        .iter()
139        .copied()
140        .filter(|payload_type| payload_type.category() == category)
141        .filter(|payload_type| !payload_type.supports(encoding))
142        .map(|payload_type| payload_type.as_str().to_string())
143        .collect::<Vec<_>>();
144
145    if unsupported.is_empty() {
146        return;
147    }
148
149    errors.push(ConfigError::unsupported_value(
150        field,
151        format!(
152            "{} is not supported by {}",
153            encoding,
154            unsupported.join(", ")
155        ),
156    ));
157}
158
159#[cfg(test)]
160mod tests {
161    use rstest::*;
162    use serde_json::json;
163
164    use super::*;
165    use crate::{config::ConfigError, enums::SerializationEncoding};
166
167    #[rstest]
168    fn test_default_message_bus_config() {
169        let config = MessageBusConfig::default();
170        assert_eq!(config.encoding, SerializationEncoding::Json);
171        assert_eq!(config.encoding_market_data, None);
172        assert_eq!(config.encoding_builtin, None);
173        assert!(!config.timestamps_as_iso8601);
174        assert_eq!(config.buffer_interval_ms, None);
175        assert_eq!(config.autotrim_mins, None);
176        assert_eq!(config.autotrim_maxlen, None);
177        assert!(config.use_trader_prefix);
178        assert!(config.use_trader_id);
179        assert!(!config.use_instance_id);
180        assert_eq!(config.streams_prefix, "stream");
181        assert!(config.stream_per_topic);
182        assert_eq!(config.external_streams, None);
183        assert_eq!(config.types_filter, None);
184    }
185
186    #[rstest]
187    fn test_deserialize_message_bus_config() {
188        let config_json = json!({
189            "encoding": "json",
190            "encoding_market_data": "sbe",
191            "encoding_builtin": "msgpack",
192            "timestamps_as_iso8601": true,
193            "buffer_interval_ms": 100,
194            "autotrim_mins": 60,
195            "autotrim_maxlen": 10_000,
196            "use_trader_prefix": false,
197            "use_trader_id": false,
198            "use_instance_id": true,
199            "streams_prefix": "data_streams",
200            "stream_per_topic": false,
201            "external_streams": ["stream1", "stream2"],
202            "types_filter": ["type1", "type2"]
203        });
204        let config: MessageBusConfig = serde_json::from_value(config_json).unwrap();
205        assert_eq!(config.encoding, SerializationEncoding::Json);
206        assert_eq!(
207            config.encoding_market_data,
208            Some(SerializationEncoding::Sbe)
209        );
210        assert_eq!(
211            config.encoding_builtin,
212            Some(SerializationEncoding::MsgPack)
213        );
214        assert!(config.timestamps_as_iso8601);
215        assert_eq!(config.buffer_interval_ms, Some(100));
216        assert_eq!(config.autotrim_mins, Some(60));
217        assert_eq!(config.autotrim_maxlen, Some(10_000));
218        assert!(!config.use_trader_prefix);
219        assert!(!config.use_trader_id);
220        assert!(config.use_instance_id);
221        assert_eq!(config.streams_prefix, "data_streams");
222        assert!(!config.stream_per_topic);
223        assert_eq!(
224            config.external_streams,
225            Some(vec!["stream1".to_string(), "stream2".to_string()])
226        );
227        assert_eq!(
228            config.types_filter,
229            Some(vec!["type1".to_string(), "type2".to_string()])
230        );
231    }
232
233    #[rstest]
234    fn test_deserialize_message_bus_config_rejects_backing_field() {
235        let config_json = json!({
236            "backing": {},
237        });
238
239        let error = serde_json::from_value::<MessageBusConfig>(config_json).unwrap_err();
240        assert!(error.to_string().contains("unknown field `backing`"));
241    }
242
243    #[rstest]
244    #[case("sbe", SerializationEncoding::Sbe)]
245    fn test_deserialize_message_bus_config_with_schema_encoding(
246        #[case] encoding_name: &str,
247        #[case] expected: SerializationEncoding,
248    ) {
249        let config_json = json!({
250            "encoding": encoding_name,
251        });
252
253        let config: MessageBusConfig = serde_json::from_value(config_json).unwrap();
254        assert_eq!(config.encoding, expected);
255    }
256
257    #[rstest]
258    fn message_bus_config_validate_accepts_default() {
259        let config = MessageBusConfig::default();
260
261        assert!(config.validate().is_ok());
262    }
263
264    #[rstest]
265    #[case("encoding")]
266    #[case("encoding_market_data")]
267    #[case("encoding_builtin")]
268    fn message_bus_config_rejects_removed_encoding(#[case] field: &str) {
269        let mut config = serde_json::Map::new();
270        config.insert(field.into(), json!("capnp"));
271
272        let error = serde_json::from_value::<MessageBusConfig>(config.into()).unwrap_err();
273
274        assert_eq!(
275            error.to_string(),
276            "unknown variant `capnp`, expected one of `json`, `msgpack`, `sbe`"
277        );
278    }
279
280    #[rstest]
281    #[case(SerializationEncoding::Json)]
282    #[case(SerializationEncoding::MsgPack)]
283    fn message_bus_config_validate_accepts_custom_safe_default(
284        #[case] encoding: SerializationEncoding,
285    ) {
286        let config = MessageBusConfig {
287            encoding,
288            ..Default::default()
289        };
290
291        assert!(config.validate().is_ok());
292    }
293
294    #[rstest]
295    #[case(SerializationEncoding::Sbe)]
296    fn message_bus_config_validate_rejects_schema_default(#[case] encoding: SerializationEncoding) {
297        let config = MessageBusConfig {
298            encoding,
299            ..Default::default()
300        };
301
302        let error = config.validate().unwrap_err();
303        assert!(matches!(
304            error,
305            ConfigError::UnsupportedValue { field, .. }
306                if field == "MessageBusConfig.encoding"
307        ));
308    }
309
310    #[cfg(feature = "sbe")]
311    #[rstest]
312    #[cfg_attr(feature = "sbe", case(SerializationEncoding::Sbe))]
313    fn message_bus_config_validate_accepts_market_data_override(
314        #[case] encoding: SerializationEncoding,
315    ) {
316        let config = MessageBusConfig {
317            encoding_market_data: Some(encoding),
318            ..Default::default()
319        };
320
321        assert!(config.validate().is_ok());
322    }
323
324    #[cfg(not(feature = "sbe"))]
325    #[rstest]
326    fn message_bus_config_validate_rejects_market_data_sbe_without_feature() {
327        let config = MessageBusConfig {
328            encoding_market_data: Some(SerializationEncoding::Sbe),
329            ..Default::default()
330        };
331
332        let error = config.validate().unwrap_err();
333        assert!(matches!(
334            error,
335            ConfigError::UnsupportedValue { field, .. }
336                if field == "MessageBusConfig.encoding_market_data"
337        ));
338    }
339
340    #[rstest]
341    #[case(SerializationEncoding::Sbe)]
342    fn message_bus_config_validate_rejects_builtin_schema_override(
343        #[case] encoding: SerializationEncoding,
344    ) {
345        let config = MessageBusConfig {
346            encoding_builtin: Some(encoding),
347            ..Default::default()
348        };
349
350        let error = config.validate().unwrap_err();
351        assert!(matches!(
352            error,
353            ConfigError::UnsupportedValue { field, .. }
354                if field == "MessageBusConfig.encoding_builtin"
355        ));
356    }
357}
nautilus_common/msgbus/config.rs

nautilus_common/msgbus/
config.rs
