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 types **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 a [`ConfigError`] when the default encoding cannot carry custom payloads, or when
95    /// a category 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    #[case("capnp", SerializationEncoding::Capnp)]
246    fn test_deserialize_message_bus_config_with_schema_encoding(
247        #[case] encoding_name: &str,
248        #[case] expected: SerializationEncoding,
249    ) {
250        let config_json = json!({
251            "encoding": encoding_name,
252        });
253
254        let config: MessageBusConfig = serde_json::from_value(config_json).unwrap();
255        assert_eq!(config.encoding, expected);
256    }
257
258    #[rstest]
259    fn message_bus_config_validate_accepts_default() {
260        let config = MessageBusConfig::default();
261
262        assert!(config.validate().is_ok());
263    }
264
265    #[rstest]
266    #[case(SerializationEncoding::Json)]
267    #[case(SerializationEncoding::MsgPack)]
268    fn message_bus_config_validate_accepts_custom_safe_default(
269        #[case] encoding: SerializationEncoding,
270    ) {
271        let config = MessageBusConfig {
272            encoding,
273            ..Default::default()
274        };
275
276        assert!(config.validate().is_ok());
277    }
278
279    #[rstest]
280    #[case(SerializationEncoding::Sbe)]
281    #[case(SerializationEncoding::Capnp)]
282    fn message_bus_config_validate_rejects_schema_default(#[case] encoding: SerializationEncoding) {
283        let config = MessageBusConfig {
284            encoding,
285            ..Default::default()
286        };
287
288        let error = config.validate().unwrap_err();
289        assert!(matches!(
290            error,
291            ConfigError::UnsupportedValue { field, .. }
292                if field == "MessageBusConfig.encoding"
293        ));
294    }
295
296    #[cfg(any(feature = "sbe", feature = "capnp"))]
297    #[rstest]
298    #[cfg_attr(feature = "sbe", case(SerializationEncoding::Sbe))]
299    #[cfg_attr(feature = "capnp", case(SerializationEncoding::Capnp))]
300    fn message_bus_config_validate_accepts_market_data_override(
301        #[case] encoding: SerializationEncoding,
302    ) {
303        let config = MessageBusConfig {
304            encoding_market_data: Some(encoding),
305            ..Default::default()
306        };
307
308        assert!(config.validate().is_ok());
309    }
310
311    #[cfg(not(feature = "sbe"))]
312    #[rstest]
313    fn message_bus_config_validate_rejects_market_data_sbe_without_feature() {
314        let config = MessageBusConfig {
315            encoding_market_data: Some(SerializationEncoding::Sbe),
316            ..Default::default()
317        };
318
319        let error = config.validate().unwrap_err();
320        assert!(matches!(
321            error,
322            ConfigError::UnsupportedValue { field, .. }
323                if field == "MessageBusConfig.encoding_market_data"
324        ));
325    }
326
327    #[cfg(not(feature = "capnp"))]
328    #[rstest]
329    fn message_bus_config_validate_rejects_market_data_capnp_without_feature() {
330        let config = MessageBusConfig {
331            encoding_market_data: Some(SerializationEncoding::Capnp),
332            ..Default::default()
333        };
334
335        let error = config.validate().unwrap_err();
336        assert!(matches!(
337            error,
338            ConfigError::UnsupportedValue { field, .. }
339                if field == "MessageBusConfig.encoding_market_data"
340        ));
341    }
342
343    #[rstest]
344    #[case(SerializationEncoding::Sbe)]
345    #[case(SerializationEncoding::Capnp)]
346    fn message_bus_config_validate_rejects_builtin_schema_override(
347        #[case] encoding: SerializationEncoding,
348    ) {
349        let config = MessageBusConfig {
350            encoding_builtin: Some(encoding),
351            ..Default::default()
352        };
353
354        let error = config.validate().unwrap_err();
355        assert!(matches!(
356            error,
357            ConfigError::UnsupportedValue { field, .. }
358                if field == "MessageBusConfig.encoding_builtin"
359        ));
360    }
361}
nautilus_common/msgbus/config.rs

nautilus_common/msgbus/
config.rs
