nautilus_network/websocket/
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//! Configuration for WebSocket client connections.
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 serde::{Deserialize, Serialize};
31
32use crate::error::{NetworkConfigError, NetworkConfigResult};
33
34/// WebSocket transport backend selection.
35///
36/// Selection is runtime so multiple backends can compile side-by-side without
37/// a `compile_error!` collision under `--all-features`.
38///
39/// `Sockudo` is the default backend and is enabled by the `transport-sockudo`
40/// Cargo feature (on by default); it uses a local HTTP/1.1 handshake helper to
41/// pass custom upgrade headers through. When the feature is disabled the
42/// default falls back to `Tungstenite`, which is always compiled and supports
43/// custom HTTP upgrade headers on the WebSocket handshake (see
44/// [`WebSocketConfig::headers`]).
45#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash, Serialize, Deserialize)]
46#[serde(rename_all = "snake_case")]
47pub enum TransportBackend {
48    /// `tokio-tungstenite` backed transport (default when `transport-sockudo` is disabled).
49    #[cfg_attr(not(feature = "transport-sockudo"), default)]
50    Tungstenite,
51    /// `sockudo-ws` backed transport (default; gated on `transport-sockudo` feature).
52    #[cfg_attr(feature = "transport-sockudo", default)]
53    Sockudo,
54}
55
56/// Configuration for WebSocket client connections.
57///
58/// This struct contains only static configuration settings. Runtime callbacks
59/// (message handler, ping handler) are passed separately to `connect()`.
60///
61/// # Connection Modes
62///
63/// ## Handler Mode
64///
65/// - Use with [`crate::websocket::WebSocketClient::connect`].
66/// - Pass a message handler to `connect()` to receive messages via callback.
67/// - Client spawns internal task to read messages and call handler.
68/// - Supports automatic reconnection with exponential backoff.
69/// - Reconnection config fields (`reconnect_*`) are active.
70/// - Best for long-lived connections, Python bindings, callback-based APIs.
71///
72/// ## Stream Mode
73///
74/// - Use with [`crate::websocket::WebSocketClient::connect_stream`].
75/// - Returns a [`MessageReader`](super::types::MessageReader) stream for the caller to read from.
76/// - **Does NOT support automatic reconnection** (reader owned by caller).
77/// - Reconnection config fields are ignored.
78/// - On disconnect, client transitions to CLOSED state and caller must manually reconnect.
79#[cfg_attr(
80    feature = "python",
81    pyo3::pyclass(module = "nautilus_trader.core.nautilus_pyo3.network", from_py_object)
82)]
83#[cfg_attr(
84    feature = "python",
85    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.network")
86)]
87#[allow(
88    clippy::unsafe_derive_deserialize,
89    reason = "PyO3-backed config still needs serde deserialization for strict config decoding"
90)]
91#[derive(Clone, Debug, Serialize, Deserialize, bon::Builder)]
92#[builder(finish_fn(name = build_inner, vis = ""))]
93#[serde(deny_unknown_fields)]
94pub struct WebSocketConfig {
95    /// The URL to connect to.
96    pub url: String,
97    /// The default headers.
98    #[serde(default)]
99    #[builder(default)]
100    pub headers: Vec<(String, String)>,
101    /// The optional heartbeat interval (seconds).
102    #[serde(default)]
103    pub heartbeat: Option<u64>,
104    /// The optional heartbeat message.
105    #[serde(default)]
106    pub heartbeat_msg: Option<String>,
107    /// The timeout (milliseconds) for reconnection attempts.
108    /// **Note**: Only applies to handler mode. Ignored in stream mode.
109    /// Must be non-zero when set.
110    #[serde(default)]
111    pub reconnect_timeout_ms: Option<u64>,
112    /// The initial reconnection delay (milliseconds) for reconnects.
113    /// **Note**: Only applies to handler mode. Ignored in stream mode.
114    #[serde(default)]
115    pub reconnect_delay_initial_ms: Option<u64>,
116    /// The maximum reconnect delay (milliseconds) for exponential backoff.
117    /// **Note**: Only applies to handler mode. Ignored in stream mode.
118    #[serde(default)]
119    pub reconnect_delay_max_ms: Option<u64>,
120    /// The exponential backoff factor for reconnection delays.
121    /// **Note**: Only applies to handler mode. Ignored in stream mode.
122    #[serde(default)]
123    pub reconnect_backoff_factor: Option<f64>,
124    /// The maximum jitter (milliseconds) added to reconnection delays.
125    /// **Note**: Only applies to handler mode. Ignored in stream mode.
126    #[serde(default)]
127    pub reconnect_jitter_ms: Option<u64>,
128    /// The maximum number of reconnection attempts before giving up.
129    /// **Note**: Only applies to handler mode. Ignored in stream mode.
130    /// - `None`: Unlimited reconnection attempts (default, recommended for production).
131    /// - `Some(n)`: After n failed attempts, transition to CLOSED state.
132    #[serde(default)]
133    pub reconnect_max_attempts: Option<u32>,
134    /// The idle timeout (milliseconds) for the read task.
135    /// When set, the read task will break and trigger reconnection if no data
136    /// is received within this duration. Useful for detecting silently dead
137    /// connections where the server stops sending without closing.
138    /// **Note**: Only applies to handler mode. Ignored in stream mode.
139    #[serde(default)]
140    pub idle_timeout_ms: Option<u64>,
141    /// The transport backend to use for the WebSocket connection.
142    ///
143    /// Defaults to [`TransportBackend::Sockudo`] when the `transport-sockudo`
144    /// Cargo feature is enabled (the default), otherwise [`TransportBackend::Tungstenite`].
145    /// When the feature is disabled, `connect_with_server` returns an error if
146    /// `Sockudo` is selected. Both backends pass `headers` into the HTTP
147    /// upgrade request. The Sockudo backend does not yet support proxy tunnels;
148    /// when [`Self::proxy_url`] is set, `connect_with_server` logs a warning
149    /// and routes through Tungstenite regardless of this field.
150    #[serde(default)]
151    #[builder(default)]
152    pub backend: TransportBackend,
153    /// Optional forward proxy URL for the WebSocket connection.
154    ///
155    /// Routes the connection through an HTTP `CONNECT` tunnel. Accepts
156    /// `http://` and `https://` schemes; SOCKS schemes are not yet supported.
157    #[serde(default)]
158    pub proxy_url: Option<String>,
159}
160
161impl<S: web_socket_config_builder::IsComplete> WebSocketConfigBuilder<S> {
162    /// Validates and builds the [`WebSocketConfig`].
163    ///
164    /// # Errors
165    ///
166    /// Returns a [`NetworkConfigError`] if any field fails validation
167    /// (see [`WebSocketConfig::validate`]).
168    pub fn build(self) -> NetworkConfigResult<WebSocketConfig> {
169        let config = self.build_inner();
170        config.validate()?;
171        Ok(config)
172    }
173}
174
175impl WebSocketConfig {
176    /// Checks whether all WebSocket settings are valid.
177    ///
178    /// # Errors
179    ///
180    /// Returns a [`NetworkConfigError`] if `url` is empty, the heartbeat interval or a
181    /// reconnection timing field is not positive, `reconnect_backoff_factor` is not finite and
182    /// at least `1.0`, or `reconnect_delay_initial_ms` exceeds `reconnect_delay_max_ms`.
183    pub fn validate(&self) -> NetworkConfigResult<()> {
184        let mut errors = Vec::new();
185
186        if self.url.trim().is_empty() {
187            errors.push(NetworkConfigError::invalid("url", "must not be empty"));
188        }
189
190        if let Some(interval) = self.heartbeat
191            && interval == 0
192        {
193            errors.push(NetworkConfigError::invalid(
194                "heartbeat",
195                "interval must be positive",
196            ));
197        }
198
199        // `reconnect_jitter_ms` is intentionally unchecked: zero disables jitter and
200        // `ExponentialBackoff::new` accepts it.
201        for (field, value) in [
202            ("reconnect_timeout_ms", self.reconnect_timeout_ms),
203            (
204                "reconnect_delay_initial_ms",
205                self.reconnect_delay_initial_ms,
206            ),
207            ("reconnect_delay_max_ms", self.reconnect_delay_max_ms),
208            ("idle_timeout_ms", self.idle_timeout_ms),
209        ] {
210            if let Some(value) = value
211                && value == 0
212            {
213                errors.push(NetworkConfigError::invalid(
214                    field,
215                    format!("must be positive, was {value}"),
216                ));
217            }
218        }
219
220        if let Some(factor) = self.reconnect_backoff_factor
221            && !(factor.is_finite() && factor >= 1.0)
222        {
223            errors.push(NetworkConfigError::invalid(
224                "reconnect_backoff_factor",
225                format!("must be finite and >= 1.0, was {factor}"),
226            ));
227        }
228
229        if let (Some(initial), Some(max)) =
230            (self.reconnect_delay_initial_ms, self.reconnect_delay_max_ms)
231            && initial > max
232        {
233            errors.push(NetworkConfigError::invalid(
234                "reconnect_delay_initial_ms",
235                format!("must not exceed reconnect_delay_max_ms ({max}), was {initial}"),
236            ));
237        }
238
239        NetworkConfigError::collect(errors)
240    }
241}
242
243#[cfg(test)]
244mod tests {
245    use rstest::rstest;
246    use serde_json::json;
247
248    use super::WebSocketConfig;
249    use crate::error::NetworkConfigError;
250
251    #[rstest]
252    fn test_deserialize_websocket_config_rejects_unknown_field() {
253        let config = json!({
254            "url": "wss://example.com/ws",
255            "unexpected": true,
256        });
257
258        let error = serde_json::from_value::<WebSocketConfig>(config).unwrap_err();
259
260        assert!(error.to_string().contains("unknown field `unexpected`"));
261    }
262
263    fn valid_config() -> WebSocketConfig {
264        WebSocketConfig::builder()
265            .url("wss://example.com/ws".to_string())
266            .build()
267            .expect("baseline websocket config should be valid")
268    }
269
270    #[rstest]
271    fn test_builder_accepts_valid_config() {
272        let result = WebSocketConfig::builder()
273            .url("wss://example.com/ws".to_string())
274            .build();
275
276        assert!(result.is_ok());
277    }
278
279    #[rstest]
280    fn test_validate_accepts_zero_jitter() {
281        let mut config = valid_config();
282        config.reconnect_jitter_ms = Some(0);
283
284        assert!(config.validate().is_ok());
285    }
286
287    #[rstest]
288    #[case::empty_url(|c: &mut WebSocketConfig| c.url = String::new(), "url")]
289    #[case::heartbeat(|c: &mut WebSocketConfig| c.heartbeat = Some(0), "heartbeat")]
290    #[case::reconnect_timeout(|c: &mut WebSocketConfig| c.reconnect_timeout_ms = Some(0), "reconnect_timeout_ms")]
291    #[case::reconnect_delay_initial(|c: &mut WebSocketConfig| c.reconnect_delay_initial_ms = Some(0), "reconnect_delay_initial_ms")]
292    #[case::reconnect_delay_max(|c: &mut WebSocketConfig| c.reconnect_delay_max_ms = Some(0), "reconnect_delay_max_ms")]
293    #[case::idle_timeout(|c: &mut WebSocketConfig| c.idle_timeout_ms = Some(0), "idle_timeout_ms")]
294    fn test_validate_rejects_invalid_field(
295        #[case] mutate: fn(&mut WebSocketConfig),
296        #[case] expected_field: &str,
297    ) {
298        let mut config = valid_config();
299        mutate(&mut config);
300
301        let err = config
302            .validate()
303            .expect_err("invalid value should be rejected");
304
305        assert!(
306            matches!(err, NetworkConfigError::Invalid { field, .. } if field == expected_field)
307        );
308    }
309
310    #[rstest]
311    #[case::too_small(0.5)]
312    #[case::nan(f64::NAN)]
313    #[case::infinite(f64::INFINITY)]
314    fn test_validate_rejects_invalid_backoff_factor(#[case] factor: f64) {
315        let mut config = valid_config();
316        config.reconnect_backoff_factor = Some(factor);
317
318        let err = config
319            .validate()
320            .expect_err("invalid backoff factor should be rejected");
321
322        assert!(
323            matches!(err, NetworkConfigError::Invalid { field, .. } if field == "reconnect_backoff_factor")
324        );
325    }
326
327    #[rstest]
328    fn test_validate_rejects_delay_initial_exceeding_max() {
329        let mut config = valid_config();
330        config.reconnect_delay_initial_ms = Some(5_000);
331        config.reconnect_delay_max_ms = Some(1_000);
332
333        let err = config
334            .validate()
335            .expect_err("initial delay above max should be rejected");
336
337        assert!(
338            matches!(err, NetworkConfigError::Invalid { field, .. } if field == "reconnect_delay_initial_ms")
339        );
340    }
341
342    #[rstest]
343    fn test_validate_collects_multiple_errors() {
344        let mut config = valid_config();
345        config.url = String::new();
346        config.reconnect_timeout_ms = Some(0);
347
348        let err = config.validate().expect_err("multiple invalid fields");
349
350        match err {
351            NetworkConfigError::Multiple { errors } => assert_eq!(errors.len(), 2),
352            other @ NetworkConfigError::Invalid { .. } => {
353                panic!("expected Multiple, was {other:?}")
354            }
355        }
356    }
357}
nautilus_network/websocket/config.rs

nautilus_network/websocket/
config.rs
