nautilus_common/messages/system/
socket.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 std::{any::Any, fmt::Display};
17
18use nautilus_core::{UUID4, UnixNanos};
19use nautilus_model::identifiers::{ClientId, TraderId, Venue};
20use ustr::Ustr;
21
22#[cfg(any(feature = "live", test))]
23const ENDPOINT_MAX_LEN: usize = 128;
24
25#[cfg(any(feature = "live", test))]
26pub(crate) fn socket_endpoint(endpoint: &str) -> anyhow::Result<Ustr> {
27    if endpoint.is_empty() {
28        anyhow::bail!("Socket endpoint cannot be empty");
29    }
30
31    if endpoint.len() > ENDPOINT_MAX_LEN {
32        anyhow::bail!("Socket endpoint cannot exceed {ENDPOINT_MAX_LEN} bytes");
33    }
34
35    if !endpoint
36        .bytes()
37        .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'-' | b'_'))
38    {
39        anyhow::bail!("Socket endpoint must contain only ASCII letters, digits, '.', '-', or '_'");
40    }
41
42    Ok(Ustr::from(endpoint))
43}
44
45/// Command requesting reconnect of one socket endpoint owned by one client.
46#[repr(C)]
47#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
48#[cfg_attr(
49    feature = "python",
50    pyo3::pyclass(module = "nautilus_trader.common", from_py_object)
51)]
52#[cfg_attr(
53    feature = "python",
54    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.common")
55)]
56pub struct ReconnectSocket {
57    pub trader_id: TraderId,
58    pub client_id: ClientId,
59    pub endpoint: Ustr,
60    pub ts_init: UnixNanos,
61}
62
63impl ReconnectSocket {
64    /// Creates a new [`ReconnectSocket`] instance.
65    #[must_use]
66    pub const fn new(
67        trader_id: TraderId,
68        client_id: ClientId,
69        endpoint: Ustr,
70        ts_init: UnixNanos,
71    ) -> Self {
72        Self {
73            trader_id,
74            client_id,
75            endpoint,
76            ts_init,
77        }
78    }
79}
80
81impl Display for ReconnectSocket {
82    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
83        write!(
84            f,
85            "{}(trader_id={}, client_id={}, endpoint={})",
86            stringify!(ReconnectSocket),
87            self.trader_id,
88            self.client_id,
89            self.endpoint,
90        )
91    }
92}
93
94/// Represents the availability state of a socket transport.
95#[repr(C)]
96#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
97#[cfg_attr(
98    feature = "python",
99    pyo3::pyclass(
100        frozen,
101        eq,
102        eq_int,
103        module = "nautilus_trader.common",
104        from_py_object,
105        rename_all = "SCREAMING_SNAKE_CASE",
106    )
107)]
108#[cfg_attr(
109    feature = "python",
110    pyo3_stub_gen::derive::gen_stub_pyclass_enum(module = "nautilus_trader.common")
111)]
112pub enum SocketState {
113    /// The transport is available.
114    Connected,
115    /// An active transport was lost.
116    Disconnected,
117}
118
119/// Transport-neutral notification that a socket state changed.
120#[derive(Debug, Clone, Copy, PartialEq, Eq)]
121pub struct SocketStateChange {
122    /// The client ID associated with the transport.
123    pub client_id: ClientId,
124    /// The venue associated with the transport, if any.
125    pub venue: Option<Venue>,
126    /// Stable, non-secret label identifying the endpoint.
127    pub endpoint: Ustr,
128    /// The current transport state.
129    pub state: SocketState,
130}
131
132impl SocketStateChange {
133    /// Creates a new [`SocketStateChange`] instance.
134    #[must_use]
135    pub const fn new(
136        client_id: ClientId,
137        venue: Option<Venue>,
138        endpoint: Ustr,
139        state: SocketState,
140    ) -> Self {
141        Self {
142            client_id,
143            venue,
144            endpoint,
145            state,
146        }
147    }
148}
149
150impl Display for SocketStateChange {
151    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
152        write!(
153            f,
154            "{}(client_id={}, venue={:?}, endpoint={}, state={:?})",
155            stringify!(SocketStateChange),
156            self.client_id,
157            self.venue,
158            self.endpoint,
159            self.state,
160        )
161    }
162}
163
164/// Represents an event where a socket transport state has changed.
165#[repr(C)]
166#[derive(Debug, Clone, PartialEq, Eq)]
167#[cfg_attr(
168    feature = "python",
169    pyo3::pyclass(module = "nautilus_trader.common", from_py_object)
170)]
171#[cfg_attr(
172    feature = "python",
173    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.common")
174)]
175pub struct SocketStateChanged {
176    /// The trader ID associated with the event.
177    pub trader_id: TraderId,
178    /// The client ID associated with the transport.
179    pub client_id: ClientId,
180    /// The venue associated with the transport, if any.
181    pub venue: Option<Venue>,
182    /// Stable, non-secret label identifying the endpoint.
183    pub endpoint: Ustr,
184    /// The current transport state.
185    pub state: SocketState,
186    /// The event ID.
187    pub event_id: UUID4,
188    /// UNIX timestamp (nanoseconds) when the event occurred.
189    pub ts_event: UnixNanos,
190    /// UNIX timestamp (nanoseconds) when the instance was initialized.
191    pub ts_init: UnixNanos,
192}
193
194impl SocketStateChanged {
195    /// Creates a new [`SocketStateChanged`] instance.
196    #[expect(clippy::too_many_arguments)]
197    #[must_use]
198    pub const fn new(
199        trader_id: TraderId,
200        client_id: ClientId,
201        venue: Option<Venue>,
202        endpoint: Ustr,
203        state: SocketState,
204        event_id: UUID4,
205        ts_event: UnixNanos,
206        ts_init: UnixNanos,
207    ) -> Self {
208        Self {
209            trader_id,
210            client_id,
211            venue,
212            endpoint,
213            state,
214            event_id,
215            ts_event,
216            ts_init,
217        }
218    }
219
220    pub fn as_any(&self) -> &dyn Any {
221        self
222    }
223}
224
225impl Display for SocketStateChanged {
226    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
227        write!(
228            f,
229            "{}(trader_id={}, client_id={}, venue={:?}, endpoint={}, state={:?}, event_id={})",
230            stringify!(SocketStateChanged),
231            self.trader_id,
232            self.client_id,
233            self.venue,
234            self.endpoint,
235            self.state,
236            self.event_id,
237        )
238    }
239}
240
241#[cfg(test)]
242mod tests {
243    use rstest::rstest;
244
245    use super::*;
246
247    #[rstest]
248    #[case("market")]
249    #[case("polymarket-market-streams-1")]
250    #[case("feed.v2_primary")]
251    fn test_socket_endpoint_accepts_identifier_labels(#[case] endpoint: &str) {
252        assert_eq!(socket_endpoint(endpoint).unwrap().as_str(), endpoint);
253    }
254
255    #[rstest]
256    #[case("")]
257    #[case("wss://example.com/feed?token=secret")]
258    #[case("user@example.com")]
259    #[case("contains space")]
260    fn test_socket_endpoint_rejects_non_identifier_values(#[case] endpoint: &str) {
261        assert!(socket_endpoint(endpoint).is_err());
262    }
263
264    #[rstest]
265    fn test_socket_endpoint_enforces_maximum_length() {
266        let maximum = "a".repeat(ENDPOINT_MAX_LEN);
267        let too_long = "a".repeat(ENDPOINT_MAX_LEN + 1);
268
269        assert_eq!(socket_endpoint(&maximum).unwrap().as_str(), maximum);
270        assert_eq!(
271            socket_endpoint(&too_long).unwrap_err().to_string(),
272            "Socket endpoint cannot exceed 128 bytes",
273        );
274    }
275
276    #[rstest]
277    #[case(
278        Some("BINANCE"),
279        SocketState::Disconnected,
280        "binance-futures-market-streams"
281    )]
282    #[case(None, SocketState::Connected, "direct-feed")]
283    fn test_socket_state_changed_new_assigns_all_fields(
284        #[case] venue: Option<&str>,
285        #[case] state: SocketState,
286        #[case] endpoint: &str,
287    ) {
288        let trader_id = TraderId::from("TRADER-001");
289        let client_id = ClientId::from("BINANCE");
290        let venue = venue.map(Venue::from);
291        let endpoint = Ustr::from(endpoint);
292        let event_id = UUID4::from("00000000-0000-4000-8000-000000000001");
293        let ts_event = UnixNanos::from(29);
294        let ts_init = UnixNanos::from(31);
295
296        let event = SocketStateChanged::new(
297            trader_id, client_id, venue, endpoint, state, event_id, ts_event, ts_init,
298        );
299
300        assert_eq!(event.trader_id, trader_id);
301        assert_eq!(event.client_id, client_id);
302        assert_eq!(event.venue, venue);
303        assert_eq!(event.endpoint, endpoint);
304        assert_eq!(event.state, state);
305        assert_eq!(event.event_id, event_id);
306        assert_eq!(event.ts_event, ts_event);
307        assert_eq!(event.ts_init, ts_init);
308    }
309}
nautilus_common/messages/system/socket.rs

nautilus_common/messages/system/
socket.rs
