nautilus_network/socket/
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
16//! Socket configuration.
17//!
18//! # Reconnection Strategy
19//!
20//! The default configuration uses unlimited reconnection attempts (`reconnect_max_attempts: None`).
21//! This is intentional for trading systems because:
22//! - Venues may be down for extended periods but eventually recover.
23//! - Exponential backoff already prevents resource waste.
24//! - Automatic recovery can be useful when manual intervention is not desirable.
25//!
26//! Use `Some(n)` primarily for testing, development, or non-critical connections.
27
28use std::fmt::Debug;
29
30use tokio_tungstenite::tungstenite::stream::Mode;
31
32use super::types::TcpMessageHandler;
33use crate::error::{NetworkConfigError, NetworkConfigResult};
34
35/// Configuration for TCP socket connection.
36#[derive(bon::Builder)]
37#[builder(finish_fn(name = build_inner, vis = ""))]
38#[cfg_attr(
39    feature = "python",
40    pyo3::pyclass(module = "nautilus_trader.core.nautilus_pyo3.network", from_py_object)
41)]
42#[cfg_attr(
43    feature = "python",
44    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.network")
45)]
46pub struct SocketConfig {
47    /// The URL to connect to.
48    pub url: String,
49    /// The connection mode {Plain, TLS}.
50    pub mode: Mode,
51    /// The sequence of bytes which separates lines.
52    pub suffix: Vec<u8>,
53    /// The optional function to handle incoming messages.
54    pub message_handler: Option<TcpMessageHandler>,
55    /// The optional heartbeat with period and beat message.
56    pub heartbeat: Option<(u64, Vec<u8>)>,
57    /// The timeout (milliseconds) for reconnection attempts.
58    pub reconnect_timeout_ms: Option<u64>,
59    /// The initial reconnection delay (milliseconds) for reconnects.
60    pub reconnect_delay_initial_ms: Option<u64>,
61    /// The maximum reconnect delay (milliseconds) for exponential backoff.
62    pub reconnect_delay_max_ms: Option<u64>,
63    /// The exponential backoff factor for reconnection delays.
64    pub reconnect_backoff_factor: Option<f64>,
65    /// The maximum jitter (milliseconds) added to reconnection delays.
66    pub reconnect_jitter_ms: Option<u64>,
67    /// The maximum number of initial connection attempts (default: 5).
68    pub connection_max_retries: Option<u32>,
69    /// The maximum number of reconnection attempts before giving up.
70    /// - `None`: Unlimited reconnection attempts (default, recommended for production).
71    /// - `Some(n)`: After n failed attempts, transition to CLOSED state.
72    pub reconnect_max_attempts: Option<u32>,
73    /// The idle timeout (milliseconds) for the read task.
74    /// When set, the read task will break and trigger reconnection if no data
75    /// is received within this duration. Useful for detecting silently dead
76    /// connections where the server stops sending without closing.
77    pub idle_timeout_ms: Option<u64>,
78    /// The path to the certificates directory.
79    pub certs_dir: Option<String>,
80}
81
82impl<S: socket_config_builder::IsComplete> SocketConfigBuilder<S> {
83    /// Validates and builds the [`SocketConfig`].
84    ///
85    /// # Errors
86    ///
87    /// Returns a [`NetworkConfigError`] if any field fails validation
88    /// (see [`SocketConfig::validate`]).
89    pub fn build(self) -> NetworkConfigResult<SocketConfig> {
90        let config = self.build_inner();
91        config.validate()?;
92        Ok(config)
93    }
94}
95
96impl SocketConfig {
97    /// Checks whether all socket settings are valid.
98    ///
99    /// # Errors
100    ///
101    /// Returns a [`NetworkConfigError`] if `url` is empty, the heartbeat interval or a
102    /// reconnection timing field is not positive, `reconnect_backoff_factor` is not finite and
103    /// at least `1.0`, or `reconnect_delay_initial_ms` exceeds `reconnect_delay_max_ms`.
104    pub fn validate(&self) -> NetworkConfigResult<()> {
105        let mut errors = Vec::new();
106
107        if self.url.trim().is_empty() {
108            errors.push(NetworkConfigError::invalid("url", "must not be empty"));
109        }
110
111        if let Some((interval, _)) = &self.heartbeat
112            && *interval == 0
113        {
114            errors.push(NetworkConfigError::invalid(
115                "heartbeat",
116                "interval must be positive",
117            ));
118        }
119
120        // `reconnect_jitter_ms` is intentionally unchecked: zero disables jitter and
121        // `ExponentialBackoff::new` accepts it.
122        for (field, value) in [
123            ("reconnect_timeout_ms", self.reconnect_timeout_ms),
124            (
125                "reconnect_delay_initial_ms",
126                self.reconnect_delay_initial_ms,
127            ),
128            ("reconnect_delay_max_ms", self.reconnect_delay_max_ms),
129            ("idle_timeout_ms", self.idle_timeout_ms),
130        ] {
131            if let Some(value) = value
132                && value == 0
133            {
134                errors.push(NetworkConfigError::invalid(
135                    field,
136                    format!("must be positive, was {value}"),
137                ));
138            }
139        }
140
141        if let Some(factor) = self.reconnect_backoff_factor
142            && !(factor.is_finite() && factor >= 1.0)
143        {
144            errors.push(NetworkConfigError::invalid(
145                "reconnect_backoff_factor",
146                format!("must be finite and >= 1.0, was {factor}"),
147            ));
148        }
149
150        if let (Some(initial), Some(max)) =
151            (self.reconnect_delay_initial_ms, self.reconnect_delay_max_ms)
152            && initial > max
153        {
154            errors.push(NetworkConfigError::invalid(
155                "reconnect_delay_initial_ms",
156                format!("must not exceed reconnect_delay_max_ms ({max}), was {initial}"),
157            ));
158        }
159
160        NetworkConfigError::collect(errors)
161    }
162}
163
164impl Debug for SocketConfig {
165    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
166        f.debug_struct(stringify!(SocketConfig))
167            .field("url", &self.url)
168            .field("mode", &self.mode)
169            .field("suffix", &self.suffix)
170            .field(
171                "message_handler",
172                &self.message_handler.as_ref().map(|_| "<function>"),
173            )
174            .field("heartbeat", &self.heartbeat)
175            .field("reconnect_timeout_ms", &self.reconnect_timeout_ms)
176            .field(
177                "reconnect_delay_initial_ms",
178                &self.reconnect_delay_initial_ms,
179            )
180            .field("reconnect_delay_max_ms", &self.reconnect_delay_max_ms)
181            .field("reconnect_backoff_factor", &self.reconnect_backoff_factor)
182            .field("reconnect_jitter_ms", &self.reconnect_jitter_ms)
183            .field("connection_max_retries", &self.connection_max_retries)
184            .field("reconnect_max_attempts", &self.reconnect_max_attempts)
185            .field("idle_timeout_ms", &self.idle_timeout_ms)
186            .field("certs_dir", &self.certs_dir)
187            .finish()
188    }
189}
190
191impl Clone for SocketConfig {
192    fn clone(&self) -> Self {
193        Self {
194            url: self.url.clone(),
195            mode: self.mode,
196            suffix: self.suffix.clone(),
197            message_handler: self.message_handler.clone(),
198            heartbeat: self.heartbeat.clone(),
199            reconnect_timeout_ms: self.reconnect_timeout_ms,
200            reconnect_delay_initial_ms: self.reconnect_delay_initial_ms,
201            reconnect_delay_max_ms: self.reconnect_delay_max_ms,
202            reconnect_backoff_factor: self.reconnect_backoff_factor,
203            reconnect_jitter_ms: self.reconnect_jitter_ms,
204            connection_max_retries: self.connection_max_retries,
205            reconnect_max_attempts: self.reconnect_max_attempts,
206            idle_timeout_ms: self.idle_timeout_ms,
207            certs_dir: self.certs_dir.clone(),
208        }
209    }
210}
211
212#[cfg(test)]
213mod tests {
214    use rstest::rstest;
215    use tokio_tungstenite::tungstenite::stream::Mode;
216
217    use super::SocketConfig;
218    use crate::error::NetworkConfigError;
219
220    fn valid_config() -> SocketConfig {
221        SocketConfig::builder()
222            .url("tcp://127.0.0.1:8080".to_string())
223            .mode(Mode::Plain)
224            .suffix(vec![b'\n'])
225            .build()
226            .expect("baseline socket config should be valid")
227    }
228
229    #[rstest]
230    fn test_builder_accepts_valid_config() {
231        let result = SocketConfig::builder()
232            .url("tcp://127.0.0.1:8080".to_string())
233            .mode(Mode::Plain)
234            .suffix(vec![b'\n'])
235            .build();
236
237        assert!(result.is_ok());
238    }
239
240    #[rstest]
241    fn test_validate_accepts_zero_jitter() {
242        let mut config = valid_config();
243        config.reconnect_jitter_ms = Some(0);
244
245        assert!(config.validate().is_ok());
246    }
247
248    #[rstest]
249    #[case::empty_url(|c: &mut SocketConfig| c.url = String::new(), "url")]
250    #[case::heartbeat(|c: &mut SocketConfig| c.heartbeat = Some((0, vec![])), "heartbeat")]
251    #[case::reconnect_timeout(|c: &mut SocketConfig| c.reconnect_timeout_ms = Some(0), "reconnect_timeout_ms")]
252    #[case::reconnect_delay_initial(|c: &mut SocketConfig| c.reconnect_delay_initial_ms = Some(0), "reconnect_delay_initial_ms")]
253    #[case::reconnect_delay_max(|c: &mut SocketConfig| c.reconnect_delay_max_ms = Some(0), "reconnect_delay_max_ms")]
254    #[case::idle_timeout(|c: &mut SocketConfig| c.idle_timeout_ms = Some(0), "idle_timeout_ms")]
255    fn test_validate_rejects_invalid_field(
256        #[case] mutate: fn(&mut SocketConfig),
257        #[case] expected_field: &str,
258    ) {
259        let mut config = valid_config();
260        mutate(&mut config);
261
262        let err = config
263            .validate()
264            .expect_err("invalid value should be rejected");
265
266        assert!(
267            matches!(err, NetworkConfigError::Invalid { field, .. } if field == expected_field)
268        );
269    }
270
271    #[rstest]
272    #[case::too_small(0.5)]
273    #[case::nan(f64::NAN)]
274    #[case::infinite(f64::INFINITY)]
275    fn test_validate_rejects_invalid_backoff_factor(#[case] factor: f64) {
276        let mut config = valid_config();
277        config.reconnect_backoff_factor = Some(factor);
278
279        let err = config
280            .validate()
281            .expect_err("invalid backoff factor should be rejected");
282
283        assert!(
284            matches!(err, NetworkConfigError::Invalid { field, .. } if field == "reconnect_backoff_factor")
285        );
286    }
287
288    #[rstest]
289    fn test_validate_rejects_delay_initial_exceeding_max() {
290        let mut config = valid_config();
291        config.reconnect_delay_initial_ms = Some(5_000);
292        config.reconnect_delay_max_ms = Some(1_000);
293
294        let err = config
295            .validate()
296            .expect_err("initial delay above max should be rejected");
297
298        assert!(
299            matches!(err, NetworkConfigError::Invalid { field, .. } if field == "reconnect_delay_initial_ms")
300        );
301    }
302
303    #[rstest]
304    fn test_validate_collects_multiple_errors() {
305        let mut config = valid_config();
306        config.url = String::new();
307        config.reconnect_timeout_ms = Some(0);
308
309        let err = config.validate().expect_err("multiple invalid fields");
310
311        match err {
312            NetworkConfigError::Multiple { errors } => assert_eq!(errors.len(), 2),
313            other @ NetworkConfigError::Invalid { .. } => {
314                panic!("expected Multiple, was {other:?}")
315            }
316        }
317    }
318}
nautilus_network/socket/config.rs

nautilus_network/socket/
config.rs
