nautilus_betfair/stream/
client.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//! Betfair Exchange Stream API client.
17//!
18//! Connects to the Betfair raw TLS stream (CRLF-delimited JSON), authenticates,
19//! and manages market/order subscriptions with automatic clk-based resubscription
20//! on reconnection.
21
22use std::sync::{
23    Arc, OnceLock,
24    atomic::{AtomicBool, AtomicU64, Ordering},
25};
26
27use bytes::Bytes;
28use nautilus_network::socket::{SocketClient, SocketConfig, TcpMessageHandler, WriterCommand};
29use tokio::sync::watch; // tokio-import-ok
30use tokio_tungstenite::tungstenite::stream::Mode;
31
32use super::{
33    config::BetfairStreamConfig,
34    error::BetfairStreamError,
35    messages::{
36        Authentication, CricketSubscription, MarketDataFilter, MarketSubscription, OrderFilter,
37        OrderSubscription, RaceSubscription, StreamMarketFilter, StreamMessage, stream_decode,
38    },
39};
40use crate::common::{
41    consts::{STREAM_OP_MARKET_SUBSCRIPTION, STREAM_OP_ORDER_SUBSCRIPTION},
42    credential::BetfairCredential,
43    enums::StatusErrorCode,
44};
45
46/// Betfair Exchange Stream API client using raw TLS (CRLF-delimited JSON).
47///
48/// On connect, authenticates immediately. On reconnection, replays authentication
49/// and any active subscriptions with the latest `clk` token for delta resumption.
50///
51/// The auth bytes are stored in a watch channel so the caller can push refreshed
52/// session tokens via [`update_auth`](Self::update_auth) after keep-alive or HTTP
53/// reconnect. The `closed` flag distinguishes permanent shutdown from transient
54/// reconnect.
55#[derive(Debug)]
56pub struct BetfairStreamClient {
57    socket: SocketClient,
58    market_sub_tx: watch::Sender<Option<MarketSubscription>>,
59    market_clk_tx: watch::Sender<Option<String>>,
60    market_initial_clk_tx: watch::Sender<Option<String>>,
61    order_sub_tx: watch::Sender<Option<OrderSubscription>>,
62    order_clk_tx: watch::Sender<Option<String>>,
63    order_initial_clk_tx: watch::Sender<Option<String>>,
64    market_active_sub_id: Arc<AtomicU64>,
65    order_active_sub_id: Arc<AtomicU64>,
66    request_id: AtomicU64,
67    auth_bytes_tx: watch::Sender<Bytes>,
68    closed: AtomicBool,
69}
70
71impl BetfairStreamClient {
72    /// Connects to the Betfair stream API and authenticates.
73    ///
74    /// # Errors
75    ///
76    /// Returns an error if the connection fails or authentication cannot be sent.
77    pub async fn connect(
78        credential: &BetfairCredential,
79        session_token: String,
80        handler: TcpMessageHandler,
81        config: BetfairStreamConfig,
82    ) -> Result<Self, BetfairStreamError> {
83        let auth = Authentication::new(credential.app_key().to_string(), session_token);
84        let auth_bytes_vec = serde_json::to_vec(&auth)?;
85        let auth_bytes = Bytes::from(auth_bytes_vec.clone());
86        let (auth_bytes_tx, auth_bytes_rx) = watch::channel(auth_bytes);
87        let mode = if config.use_tls {
88            Mode::Tls
89        } else {
90            Mode::Plain
91        };
92
93        let (market_clk_tx, market_clk_rx) = watch::channel(None::<String>);
94        let (market_initial_clk_tx, market_initial_clk_rx) = watch::channel(None::<String>);
95        let (order_clk_tx, order_clk_rx) = watch::channel(None::<String>);
96        let (order_initial_clk_tx, order_initial_clk_rx) = watch::channel(None::<String>);
97        let (market_sub_tx, market_sub_rx) = watch::channel(None::<MarketSubscription>);
98        let (order_sub_tx, order_sub_rx) = watch::channel(None::<OrderSubscription>);
99
100        // Populated after connect() returns; OnceLock gives lock-free reads thereafter.
101        let shared_tx: Arc<OnceLock<tokio::sync::mpsc::UnboundedSender<WriterCommand>>> =
102            Arc::new(OnceLock::new());
103
104        // Clone senders for the handler; struct keeps originals to reset on re-subscribe.
105        let (market_clk_tx_h, market_initial_clk_tx_h) =
106            (market_clk_tx.clone(), market_initial_clk_tx.clone());
107        let (order_clk_tx_h, order_initial_clk_tx_h) =
108            (order_clk_tx.clone(), order_initial_clk_tx.clone());
109
110        let market_active_sub_id = Arc::new(AtomicU64::new(0));
111        let order_active_sub_id = Arc::new(AtomicU64::new(0));
112        let market_active_sub_id_h = Arc::clone(&market_active_sub_id);
113        let order_active_sub_id_h = Arc::clone(&order_active_sub_id);
114
115        let message_handler: TcpMessageHandler = Arc::new(move |data: &[u8]| {
116            if let Ok(msg) = stream_decode(data) {
117                match &msg {
118                    StreamMessage::MarketChange(mcm) => {
119                        let active = market_active_sub_id_h.load(Ordering::SeqCst);
120                        // Accept only when a subscription is active (active > 0) and
121                        // the message carries no id (can't discriminate, e.g. heartbeat)
122                        // or its id matches the active subscription. Reject messages that
123                        // explicitly carry a different (stale) subscription id.
124                        if active > 0 && mcm.id.is_none_or(|id| id == active) {
125                            if mcm.clk.is_some() {
126                                let _ = market_clk_tx_h.send(mcm.clk.clone());
127                            }
128
129                            if mcm.initial_clk.is_some() {
130                                let _ = market_initial_clk_tx_h.send(mcm.initial_clk.clone());
131                            }
132                        }
133                    }
134                    StreamMessage::OrderChange(ocm) => {
135                        let active = order_active_sub_id_h.load(Ordering::SeqCst);
136                        if active > 0 && ocm.id.is_none_or(|id| id == active) {
137                            if ocm.clk.is_some() {
138                                let _ = order_clk_tx_h.send(ocm.clk.clone());
139                            }
140
141                            if ocm.initial_clk.is_some() {
142                                let _ = order_initial_clk_tx_h.send(ocm.initial_clk.clone());
143                            }
144                        }
145                    }
146                    StreamMessage::Status(status) => {
147                        // Betfair rejects stale replay tokens with INVALID_CLOCK and then
148                        // closes the connection, so a loop of reconnect → same stale clk →
149                        // reject would follow unless we clear the clocks here and fall back
150                        // to a full-image resubscription on the next reconnect.
151                        if status.error_code == Some(StatusErrorCode::InvalidClock) {
152                            let _ = market_clk_tx_h.send(None);
153                            let _ = market_initial_clk_tx_h.send(None);
154                            let _ = order_clk_tx_h.send(None);
155                            let _ = order_initial_clk_tx_h.send(None);
156                            log::warn!(
157                                "Betfair stream INVALID_CLOCK: clocks cleared, \
158                                 next reconnect will request a full image",
159                            );
160                        } else if status.connection_closed {
161                            log::warn!(
162                                "Betfair stream connection closed by server: {:?} - {:?}",
163                                status.error_code,
164                                status.error_message,
165                            );
166                        } else if status.error_code.is_some() {
167                            log::warn!(
168                                "Betfair stream status error: {:?} - {:?}",
169                                status.error_code,
170                                status.error_message,
171                            );
172                        }
173                    }
174                    _ => {}
175                }
176            }
177            handler(data);
178        });
179
180        let auth_bytes_reconnect = auth_bytes_rx;
181        let shared_tx_reconnect = Arc::clone(&shared_tx);
182        let post_reconnection: Arc<dyn Fn() + Send + Sync> = Arc::new(move || {
183            let Some(tx) = shared_tx_reconnect.get() else {
184                return;
185            };
186
187            let auth = auth_bytes_reconnect.borrow().clone();
188            let market_sub = market_sub_rx.borrow().clone();
189            let order_sub = order_sub_rx.borrow().clone();
190
191            let _ = tx.send(WriterCommand::Send(auth));
192
193            if let Some(mut sub) = market_sub {
194                sub.clk = market_clk_rx.borrow().clone();
195                sub.initial_clk = market_initial_clk_rx.borrow().clone();
196                if let Ok(sub_bytes) = serde_json::to_vec(&sub) {
197                    let _ = tx.send(WriterCommand::Send(Bytes::from(sub_bytes)));
198                }
199            }
200
201            if let Some(mut sub) = order_sub {
202                sub.clk = order_clk_rx.borrow().clone();
203                sub.initial_clk = order_initial_clk_rx.borrow().clone();
204                if let Ok(sub_bytes) = serde_json::to_vec(&sub) {
205                    let _ = tx.send(WriterCommand::Send(Bytes::from(sub_bytes)));
206                }
207            }
208        });
209
210        let url = format!("{}:{}", config.host, config.port);
211        let socket_config = SocketConfig {
212            url,
213            mode,
214            suffix: b"\r\n".to_vec(),
215            message_handler: Some(message_handler),
216            // SocketConfig.heartbeat interval is in seconds; round up to avoid zero
217            heartbeat: Some((
218                config.heartbeat_ms.div_ceil(1_000),
219                b"{\"op\":\"heartbeat\"}".to_vec(),
220            )),
221            reconnect_timeout_ms: None,
222            reconnect_delay_initial_ms: Some(config.reconnect_delay_initial_ms),
223            reconnect_delay_max_ms: Some(config.reconnect_delay_max_ms),
224            reconnect_backoff_factor: None,
225            reconnect_jitter_ms: None,
226            connection_max_retries: None,
227            reconnect_max_attempts: None,
228            idle_timeout_ms: Some(config.idle_timeout_ms),
229            certs_dir: None,
230        };
231
232        let socket = SocketClient::connect(socket_config, None, Some(post_reconnection), None)
233            .await
234            .map_err(|e| BetfairStreamError::ConnectionFailed(e.to_string()))?;
235
236        // Set once, then use lock-free reads
237        let _ = shared_tx.set(socket.writer_tx.clone());
238
239        socket
240            .send_bytes(auth_bytes_vec)
241            .await
242            .map_err(|e| BetfairStreamError::ConnectionFailed(e.to_string()))?;
243
244        Ok(Self {
245            socket,
246            market_sub_tx,
247            market_clk_tx,
248            market_initial_clk_tx,
249            order_sub_tx,
250            order_clk_tx,
251            order_initial_clk_tx,
252            market_active_sub_id,
253            order_active_sub_id,
254            request_id: AtomicU64::new(1),
255            auth_bytes_tx,
256            closed: AtomicBool::new(false),
257        })
258    }
259
260    /// Subscribes to market data for the given filter and data fields.
261    ///
262    /// Stores the subscription for automatic replay on reconnection.
263    ///
264    /// # Errors
265    ///
266    /// Returns an error if serialization or sending fails.
267    pub async fn subscribe_markets(
268        &self,
269        market_filter: StreamMarketFilter,
270        data_filter: MarketDataFilter,
271        heartbeat_ms: Option<u64>,
272        conflate_ms: Option<u64>,
273    ) -> Result<(), BetfairStreamError> {
274        if self.closed.load(Ordering::SeqCst) || self.socket.is_closed() {
275            return Err(BetfairStreamError::Disconnected(
276                "stream client is closed".to_string(),
277            ));
278        }
279        let id = self.request_id.fetch_add(1, Ordering::Relaxed);
280        // Advance the active ID before clearing clocks so that any in-flight MCMs
281        // from the previous subscription are immediately rejected by the handler.
282        self.market_active_sub_id.store(id, Ordering::SeqCst);
283        let sub = MarketSubscription {
284            op: STREAM_OP_MARKET_SUBSCRIPTION.to_string(),
285            id: Some(id),
286            market_filter,
287            market_data_filter: data_filter,
288            clk: None,
289            conflate_ms,
290            heartbeat_ms,
291            initial_clk: None,
292            segmentation_enabled: None,
293        };
294
295        // Reset clocks so a disconnect before the first MCM response doesn't replay
296        // stale tokens from a previous subscription with different filters.
297        let _ = self.market_clk_tx.send(None);
298        let _ = self.market_initial_clk_tx.send(None);
299        let _ = self.market_sub_tx.send(Some(sub.clone()));
300
301        let sub_bytes = serde_json::to_vec(&sub)?;
302        self.socket
303            .send_bytes(sub_bytes)
304            .await
305            .map_err(|e| BetfairStreamError::ConnectionFailed(e.to_string()))?;
306        Ok(())
307    }
308
309    /// Subscribes to order updates.
310    ///
311    /// Stores the subscription for automatic replay on reconnection.
312    ///
313    /// # Errors
314    ///
315    /// Returns an error if serialization or sending fails.
316    pub async fn subscribe_orders(
317        &self,
318        order_filter: Option<OrderFilter>,
319        heartbeat_ms: Option<u64>,
320    ) -> Result<(), BetfairStreamError> {
321        if self.closed.load(Ordering::SeqCst) || self.socket.is_closed() {
322            return Err(BetfairStreamError::Disconnected(
323                "stream client is closed".to_string(),
324            ));
325        }
326        let id = self.request_id.fetch_add(1, Ordering::Relaxed);
327        self.order_active_sub_id.store(id, Ordering::SeqCst);
328        let sub = OrderSubscription {
329            op: STREAM_OP_ORDER_SUBSCRIPTION.to_string(),
330            id: Some(id),
331            order_filter,
332            clk: None,
333            conflate_ms: None,
334            heartbeat_ms,
335            initial_clk: None,
336            segmentation_enabled: None,
337        };
338
339        // Reset clocks so a disconnect before the first OCM response doesn't replay
340        // stale tokens from a previous subscription with different filters.
341        let _ = self.order_clk_tx.send(None);
342        let _ = self.order_initial_clk_tx.send(None);
343        let _ = self.order_sub_tx.send(Some(sub.clone()));
344
345        let sub_bytes = serde_json::to_vec(&sub)?;
346        self.socket
347            .send_bytes(sub_bytes)
348            .await
349            .map_err(|e| BetfairStreamError::ConnectionFailed(e.to_string()))?;
350        Ok(())
351    }
352
353    /// Returns `true` if the connection is active.
354    #[must_use]
355    pub fn is_active(&self) -> bool {
356        self.socket.is_active()
357    }
358
359    /// Pushes refreshed auth bytes so the next reconnection or subscription uses
360    /// the current session token instead of the one from initial connect.
361    pub fn update_auth(&self, app_key: &str, session_token: String) {
362        let auth = Authentication::new(app_key.to_string(), session_token);
363        if let Ok(bytes) = serde_json::to_vec(&auth) {
364            let _ = self.auth_bytes_tx.send(Bytes::from(bytes));
365        }
366    }
367
368    /// Closes the stream connection.
369    pub async fn close(&self) {
370        self.closed.store(true, Ordering::SeqCst);
371        self.socket.close().await;
372    }
373}
374
375/// Betfair race stream client for Total Performance Data (TPD).
376///
377/// Connects to `sports-data-stream-api.betfair.com` and subscribes to Race Change
378/// Messages (RCM) with live GPS tracking data. Simpler than [`BetfairStreamClient`]:
379/// no clk-based delta resumption, just auth + raceSubscription on (re)connect.
380#[derive(Debug)]
381pub struct BetfairRaceStreamClient {
382    socket: SocketClient,
383    auth_bytes_tx: watch::Sender<Bytes>,
384    closed: AtomicBool,
385}
386
387impl BetfairRaceStreamClient {
388    /// Connects to the Betfair race stream and subscribes.
389    ///
390    /// The `race_fatal_tx` channel receives a signal when the server returns a
391    /// fatal status error (e.g. NOT_AUTHORIZED, no TPD entitlement). The caller
392    /// should monitor this channel and close the client when it fires.
393    ///
394    /// # Errors
395    ///
396    /// Returns an error if the connection fails or the initial send fails.
397    pub async fn connect(
398        credential: &BetfairCredential,
399        session_token: String,
400        handler: TcpMessageHandler,
401        config: BetfairStreamConfig,
402        race_fatal_tx: tokio::sync::mpsc::UnboundedSender<()>,
403    ) -> Result<Self, BetfairStreamError> {
404        let race_sub = RaceSubscription::new(1);
405        let race_sub_bytes = Bytes::from(serde_json::to_vec(&race_sub)?);
406        let subscription = AuxiliaryStreamSubscription {
407            bytes: race_sub_bytes,
408            label: "race",
409            fatal_hint: "check TPD entitlement on your Betfair app key",
410            fatal_tx: race_fatal_tx,
411        };
412        Self::connect_with_subscription(credential, session_token, handler, config, subscription)
413            .await
414    }
415
416    /// Connects to the Betfair sports data stream and subscribes to cricket.
417    ///
418    /// The `cricket_fatal_tx` channel receives a signal when the server returns
419    /// a fatal status error.
420    ///
421    /// # Errors
422    ///
423    /// Returns an error if the connection fails or the initial send fails.
424    pub async fn connect_cricket(
425        credential: &BetfairCredential,
426        session_token: String,
427        handler: TcpMessageHandler,
428        config: BetfairStreamConfig,
429        cricket_fatal_tx: tokio::sync::mpsc::UnboundedSender<()>,
430    ) -> Result<Self, BetfairStreamError> {
431        let cricket_sub = CricketSubscription::new(1);
432        let cricket_sub_bytes = Bytes::from(serde_json::to_vec(&cricket_sub)?);
433        let subscription = AuxiliaryStreamSubscription {
434            bytes: cricket_sub_bytes,
435            label: "cricket",
436            fatal_hint: "check cricket data entitlement on your Betfair app key",
437            fatal_tx: cricket_fatal_tx,
438        };
439        Self::connect_with_subscription(credential, session_token, handler, config, subscription)
440            .await
441    }
442
443    async fn connect_with_subscription(
444        credential: &BetfairCredential,
445        session_token: String,
446        handler: TcpMessageHandler,
447        config: BetfairStreamConfig,
448        subscription: AuxiliaryStreamSubscription,
449    ) -> Result<Self, BetfairStreamError> {
450        let AuxiliaryStreamSubscription {
451            bytes: sub_bytes,
452            label,
453            fatal_hint,
454            fatal_tx,
455        } = subscription;
456
457        let auth = Authentication::new(credential.app_key().to_string(), session_token);
458        let auth_bytes_vec = serde_json::to_vec(&auth)?;
459        let auth_bytes = Bytes::from(auth_bytes_vec.clone());
460        let (auth_bytes_tx, auth_bytes_rx) = watch::channel(auth_bytes.clone());
461
462        let mode = if config.use_tls {
463            Mode::Tls
464        } else {
465            Mode::Plain
466        };
467
468        let shared_tx: Arc<OnceLock<tokio::sync::mpsc::UnboundedSender<WriterCommand>>> =
469            Arc::new(OnceLock::new());
470
471        let message_handler: TcpMessageHandler = Arc::new(move |data: &[u8]| {
472            if let Ok(StreamMessage::Status(status)) = stream_decode(data) {
473                if let Some(ref code) = status.error_code
474                    && code.is_race_stream_fatal()
475                {
476                    log::error!(
477                        "Betfair {label} stream fatal error: {:?} - {:?} ({fatal_hint})",
478                        status.error_code,
479                        status.error_message,
480                    );
481                    let _ = fatal_tx.send(());
482                    return;
483                }
484
485                if status.connection_closed {
486                    log::warn!(
487                        "Betfair {label} stream closed: {:?} - {:?}",
488                        status.error_code,
489                        status.error_message,
490                    );
491                } else if status.error_code.is_some() {
492                    log::warn!(
493                        "Betfair {label} stream status: {:?} - {:?}",
494                        status.error_code,
495                        status.error_message,
496                    );
497                }
498            }
499            handler(data);
500        });
501
502        let auth_bytes_reconnect = auth_bytes_rx;
503        let sub_reconnect = sub_bytes.clone();
504        let shared_tx_reconnect = Arc::clone(&shared_tx);
505        let post_reconnection: Arc<dyn Fn() + Send + Sync> = Arc::new(move || {
506            let Some(tx) = shared_tx_reconnect.get() else {
507                return;
508            };
509            let auth = auth_bytes_reconnect.borrow().clone();
510            let mut combined = Vec::with_capacity(auth.len() + 2 + sub_reconnect.len());
511            combined.extend_from_slice(&auth);
512            combined.extend_from_slice(b"\r\n");
513            combined.extend_from_slice(&sub_reconnect);
514            let _ = tx.send(WriterCommand::Send(Bytes::from(combined)));
515        });
516
517        let url = format!("{}:{}", config.host, config.port);
518        let socket_config = SocketConfig {
519            url,
520            mode,
521            suffix: b"\r\n".to_vec(),
522            message_handler: Some(message_handler),
523            heartbeat: Some((
524                config.heartbeat_ms.div_ceil(1_000),
525                b"{\"op\":\"heartbeat\"}".to_vec(),
526            )),
527            reconnect_timeout_ms: None,
528            reconnect_delay_initial_ms: Some(config.reconnect_delay_initial_ms),
529            reconnect_delay_max_ms: Some(config.reconnect_delay_max_ms),
530            reconnect_backoff_factor: None,
531            reconnect_jitter_ms: None,
532            connection_max_retries: None,
533            reconnect_max_attempts: None,
534            idle_timeout_ms: Some(config.idle_timeout_ms),
535            certs_dir: None,
536        };
537
538        let socket = SocketClient::connect(socket_config, None, Some(post_reconnection), None)
539            .await
540            .map_err(|e| BetfairStreamError::ConnectionFailed(e.to_string()))?;
541
542        let _ = shared_tx.set(socket.writer_tx.clone());
543
544        let mut combined = Vec::with_capacity(auth_bytes_vec.len() + 2 + sub_bytes.len());
545        combined.extend_from_slice(&auth_bytes_vec);
546        combined.extend_from_slice(b"\r\n");
547        combined.extend_from_slice(&sub_bytes);
548        socket
549            .send_bytes(combined)
550            .await
551            .map_err(|e| BetfairStreamError::ConnectionFailed(e.to_string()))?;
552
553        Ok(Self {
554            socket,
555            auth_bytes_tx,
556            closed: AtomicBool::new(false),
557        })
558    }
559
560    /// Returns `true` if the connection is active.
561    #[must_use]
562    pub fn is_active(&self) -> bool {
563        self.socket.is_active()
564    }
565
566    /// Pushes refreshed auth bytes so the next reconnection uses
567    /// the current session token instead of the one from initial connect.
568    pub fn update_auth(&self, app_key: &str, session_token: String) {
569        let auth = Authentication::new(app_key.to_string(), session_token);
570        if let Ok(bytes) = serde_json::to_vec(&auth) {
571            let _ = self.auth_bytes_tx.send(Bytes::from(bytes));
572        }
573    }
574
575    /// Closes the race stream connection.
576    pub async fn close(&self) {
577        self.closed.store(true, Ordering::SeqCst);
578        self.socket.close().await;
579    }
580}
581
582struct AuxiliaryStreamSubscription {
583    bytes: Bytes,
584    label: &'static str,
585    fatal_hint: &'static str,
586    fatal_tx: tokio::sync::mpsc::UnboundedSender<()>,
587}
588
589#[cfg(test)]
590mod tests {
591    use rstest::rstest;
592
593    use super::*;
594    use crate::stream::messages::{
595        Authentication, CricketSubscription, MarketDataFilter, RaceSubscription, StreamMarketFilter,
596    };
597
598    #[rstest]
599    fn test_invalid_clock_status_resets_clocks() {
600        let (market_clk_tx, market_clk_rx) = watch::channel(Some("old-market-clk".to_string()));
601        let (market_initial_clk_tx, market_initial_clk_rx) =
602            watch::channel(Some("old-market-iclk".to_string()));
603        let (order_clk_tx, order_clk_rx) = watch::channel(Some("old-order-clk".to_string()));
604        let (order_initial_clk_tx, order_initial_clk_rx) =
605            watch::channel(Some("old-order-iclk".to_string()));
606
607        let handler: TcpMessageHandler = Arc::new(move |data: &[u8]| {
608            if let Ok(msg) = stream_decode(data)
609                && let StreamMessage::Status(status) = &msg
610                && status.error_code == Some(StatusErrorCode::InvalidClock)
611            {
612                let _ = market_clk_tx.send(None);
613                let _ = market_initial_clk_tx.send(None);
614                let _ = order_clk_tx.send(None);
615                let _ = order_initial_clk_tx.send(None);
616            }
617        });
618
619        handler(
620            br#"{"op":"status","statusCode":"503","errorCode":"INVALID_CLOCK","connectionClosed":true}"#,
621        );
622
623        assert!(
624            market_clk_rx.borrow().is_none(),
625            "market clk must be cleared"
626        );
627        assert!(
628            market_initial_clk_rx.borrow().is_none(),
629            "market initialClk must be cleared"
630        );
631        assert!(order_clk_rx.borrow().is_none(), "order clk must be cleared");
632        assert!(
633            order_initial_clk_rx.borrow().is_none(),
634            "order initialClk must be cleared"
635        );
636    }
637
638    #[rstest]
639    fn test_auth_message_serialization() {
640        let auth = Authentication::new("my-app-key".to_string(), "my-session".to_string());
641        let json = serde_json::to_string(&auth).unwrap();
642        assert!(json.contains("\"op\":\"authentication\""));
643        assert!(json.contains("\"appKey\":\"my-app-key\""));
644        assert!(json.contains("\"session\":\"my-session\""));
645    }
646
647    #[rstest]
648    fn test_clk_is_updated_from_mcm() {
649        let (market_clk_tx, market_clk_rx) = watch::channel(None::<String>);
650        let (market_initial_clk_tx, market_initial_clk_rx) = watch::channel(None::<String>);
651        let (order_clk_tx, order_clk_rx) = watch::channel(None::<String>);
652        let (order_initial_clk_tx, order_initial_clk_rx) = watch::channel(None::<String>);
653        let market_active_sub_id = Arc::new(AtomicU64::new(5));
654        let order_active_sub_id = Arc::new(AtomicU64::new(6));
655
656        let handler: TcpMessageHandler = Arc::new(move |data: &[u8]| {
657            if let Ok(msg) = stream_decode(data) {
658                match &msg {
659                    StreamMessage::MarketChange(mcm) => {
660                        let active = market_active_sub_id.load(Ordering::SeqCst);
661                        if active > 0 && mcm.id.is_none_or(|id| id == active) {
662                            if mcm.clk.is_some() {
663                                let _ = market_clk_tx.send(mcm.clk.clone());
664                            }
665
666                            if mcm.initial_clk.is_some() {
667                                let _ = market_initial_clk_tx.send(mcm.initial_clk.clone());
668                            }
669                        }
670                    }
671                    StreamMessage::OrderChange(ocm) => {
672                        let active = order_active_sub_id.load(Ordering::SeqCst);
673                        if active > 0 && ocm.id.is_none_or(|id| id == active) {
674                            if ocm.clk.is_some() {
675                                let _ = order_clk_tx.send(ocm.clk.clone());
676                            }
677
678                            if ocm.initial_clk.is_some() {
679                                let _ = order_initial_clk_tx.send(ocm.initial_clk.clone());
680                            }
681                        }
682                    }
683                    _ => {}
684                }
685            }
686        });
687
688        // MCM/OCM with matching subscription id update clocks.
689        handler(br#"{"op":"mcm","id":5,"pt":1000,"initialClk":"mcm-iclk","clk":"mcm-clk"}"#);
690        handler(br#"{"op":"ocm","id":6,"pt":2000,"initialClk":"ocm-iclk","clk":"ocm-clk"}"#);
691
692        assert_eq!(market_clk_rx.borrow().as_deref(), Some("mcm-clk"));
693        assert_eq!(market_initial_clk_rx.borrow().as_deref(), Some("mcm-iclk"));
694        assert_eq!(order_clk_rx.borrow().as_deref(), Some("ocm-clk"));
695        assert_eq!(order_initial_clk_rx.borrow().as_deref(), Some("ocm-iclk"));
696
697        // MCM without an id (e.g. heartbeat) is accepted for the active subscription.
698        handler(br#"{"op":"mcm","pt":1001,"clk":"hb-clk"}"#);
699        assert_eq!(market_clk_rx.borrow().as_deref(), Some("hb-clk"));
700
701        // MCM from a stale subscription (explicit wrong id) must not overwrite stored clocks.
702        handler(br#"{"op":"mcm","id":4,"pt":1002,"clk":"stale-clk"}"#);
703        assert_eq!(market_clk_rx.borrow().as_deref(), Some("hb-clk"));
704    }
705
706    #[rstest]
707    fn test_reconnect_callback_sends_auth_and_subscription() {
708        let (market_clk_tx, market_clk_rx) = watch::channel(Some("mcm-clk1".to_string()));
709        let (market_initial_clk_tx, market_initial_clk_rx) =
710            watch::channel(Some("mcm-iclk1".to_string()));
711        let (order_clk_tx, order_clk_rx) = watch::channel(Some("ocm-clk1".to_string()));
712        let (order_initial_clk_tx, order_initial_clk_rx) =
713            watch::channel(Some("ocm-iclk1".to_string()));
714        let (market_sub_tx, market_sub_rx) = watch::channel(None::<MarketSubscription>);
715        let (order_sub_tx, order_sub_rx) = watch::channel(None::<OrderSubscription>);
716        let shared_tx: Arc<OnceLock<tokio::sync::mpsc::UnboundedSender<WriterCommand>>> =
717            Arc::new(OnceLock::new());
718
719        let auth = Authentication::new("key".to_string(), "token".to_string());
720        let auth_bytes = Bytes::from(serde_json::to_vec(&auth).unwrap());
721
722        let _ = market_sub_tx.send(Some(MarketSubscription {
723            op: STREAM_OP_MARKET_SUBSCRIPTION.to_string(),
724            id: Some(1),
725            market_filter: StreamMarketFilter::default(),
726            market_data_filter: MarketDataFilter::default(),
727            clk: None,
728            conflate_ms: None,
729            heartbeat_ms: None,
730            initial_clk: None,
731            segmentation_enabled: None,
732        }));
733        let _ = order_sub_tx.send(Some(OrderSubscription {
734            op: STREAM_OP_ORDER_SUBSCRIPTION.to_string(),
735            id: Some(2),
736            order_filter: None,
737            clk: None,
738            conflate_ms: None,
739            heartbeat_ms: None,
740            initial_clk: None,
741            segmentation_enabled: None,
742        }));
743
744        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<WriterCommand>();
745        let _ = shared_tx.set(tx);
746
747        // Build and invoke the reconnect closure (mirrors the logic in connect())
748        let auth_bytes_reconnect = auth_bytes;
749        let shared_tx_reconnect = Arc::clone(&shared_tx);
750        let post_reconnection: Arc<dyn Fn() + Send + Sync> = Arc::new(move || {
751            let Some(tx) = shared_tx_reconnect.get() else {
752                return;
753            };
754
755            let market_sub = market_sub_rx.borrow().clone();
756            let order_sub = order_sub_rx.borrow().clone();
757
758            let _ = tx.send(WriterCommand::Send(auth_bytes_reconnect.clone()));
759
760            if let Some(mut sub) = market_sub {
761                sub.clk = market_clk_rx.borrow().clone();
762                sub.initial_clk = market_initial_clk_rx.borrow().clone();
763                if let Ok(sub_bytes) = serde_json::to_vec(&sub) {
764                    let _ = tx.send(WriterCommand::Send(Bytes::from(sub_bytes)));
765                }
766            }
767
768            if let Some(mut sub) = order_sub {
769                sub.clk = order_clk_rx.borrow().clone();
770                sub.initial_clk = order_initial_clk_rx.borrow().clone();
771                if let Ok(sub_bytes) = serde_json::to_vec(&sub) {
772                    let _ = tx.send(WriterCommand::Send(Bytes::from(sub_bytes)));
773                }
774            }
775        });
776
777        drop(market_clk_tx);
778        drop(market_initial_clk_tx);
779        drop(order_clk_tx);
780        drop(order_initial_clk_tx);
781
782        post_reconnection();
783
784        let auth_cmd = rx.try_recv().expect("auth replay message");
785        let market_cmd = rx.try_recv().expect("market subscription message");
786        let order_cmd = rx.try_recv().expect("order subscription message");
787        assert!(rx.try_recv().is_err(), "no further messages expected");
788
789        let WriterCommand::Send(auth_bytes) = auth_cmd else {
790            panic!("expected Send");
791        };
792        let WriterCommand::Send(market_bytes) = market_cmd else {
793            panic!("expected Send");
794        };
795        let WriterCommand::Send(order_bytes) = order_cmd else {
796            panic!("expected Send");
797        };
798
799        let auth_str = std::str::from_utf8(&auth_bytes).unwrap();
800        let market_str = std::str::from_utf8(&market_bytes).unwrap();
801        let order_str = std::str::from_utf8(&order_bytes).unwrap();
802
803        assert!(auth_str.contains("\"op\":\"authentication\""));
804        assert!(market_str.contains("\"op\":\"marketSubscription\""));
805        // Both clk and initialClk must be injected into each resubscription
806        assert!(market_str.contains("\"clk\":\"mcm-clk1\""));
807        assert!(market_str.contains("\"initialClk\":\"mcm-iclk1\""));
808
809        assert!(order_str.contains("\"op\":\"orderSubscription\""));
810        assert!(order_str.contains("\"clk\":\"ocm-clk1\""));
811        assert!(order_str.contains("\"initialClk\":\"ocm-iclk1\""));
812    }
813
814    #[rstest]
815    fn test_race_subscription_serialization() {
816        let sub = RaceSubscription::new(42);
817        let json = serde_json::to_string(&sub).unwrap();
818        assert!(json.contains("\"op\":\"raceSubscription\""));
819        assert!(json.contains("\"id\":42"));
820    }
821
822    #[rstest]
823    fn test_cricket_subscription_serialization() {
824        let sub = CricketSubscription::new(42);
825        let json = serde_json::to_string(&sub).unwrap();
826        assert!(json.contains("\"op\":\"cricketSubscription\""));
827        assert!(json.contains("\"id\":42"));
828    }
829
830    #[rstest]
831    fn test_race_stream_reconnect_replays_auth_and_subscription() {
832        let auth = Authentication::new("key".to_string(), "token".to_string());
833        let auth_bytes = Bytes::from(serde_json::to_vec(&auth).unwrap());
834        let race_sub = RaceSubscription::new(1);
835        let race_sub_bytes = Bytes::from(serde_json::to_vec(&race_sub).unwrap());
836
837        let shared_tx: Arc<OnceLock<tokio::sync::mpsc::UnboundedSender<WriterCommand>>> =
838            Arc::new(OnceLock::new());
839
840        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<WriterCommand>();
841        let _ = shared_tx.set(tx);
842
843        let auth_reconnect = auth_bytes;
844        let sub_reconnect = race_sub_bytes;
845        let shared_tx_reconnect = Arc::clone(&shared_tx);
846        let post_reconnection: Arc<dyn Fn() + Send + Sync> = Arc::new(move || {
847            let Some(tx) = shared_tx_reconnect.get() else {
848                return;
849            };
850            let mut combined = Vec::with_capacity(auth_reconnect.len() + 2 + sub_reconnect.len());
851            combined.extend_from_slice(&auth_reconnect);
852            combined.extend_from_slice(b"\r\n");
853            combined.extend_from_slice(&sub_reconnect);
854            let _ = tx.send(WriterCommand::Send(Bytes::from(combined)));
855        });
856
857        post_reconnection();
858
859        let cmd = rx.try_recv().expect("auth+race subscription message");
860        assert!(rx.try_recv().is_err(), "no further messages expected");
861
862        let WriterCommand::Send(bytes) = cmd else {
863            panic!("expected Send");
864        };
865
866        let text = std::str::from_utf8(&bytes).unwrap();
867        let (auth_part, sub_part) = text
868            .split_once("\r\n")
869            .expect("CRLF separator in combined message");
870
871        assert!(auth_part.contains("\"op\":\"authentication\""));
872        assert!(sub_part.contains("\"op\":\"raceSubscription\""));
873    }
874
875    #[rstest]
876    fn test_race_stream_handler_fatal_status_sends_kill_signal() {
877        let (race_fatal_tx, mut race_fatal_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
878        let inner_handler: TcpMessageHandler = Arc::new(|_data: &[u8]| {});
879
880        let handler: TcpMessageHandler = Arc::new(move |data: &[u8]| {
881            if let Ok(StreamMessage::Status(status)) = stream_decode(data)
882                && let Some(ref code) = status.error_code
883                && code.is_race_stream_fatal()
884            {
885                let _ = race_fatal_tx.send(());
886                return;
887            }
888            inner_handler(data);
889        });
890
891        // Fatal: NOT_AUTHORIZED
892        handler(
893            br#"{"op":"status","statusCode":"503","errorCode":"NOT_AUTHORIZED","connectionClosed":true}"#,
894        );
895        assert!(
896            race_fatal_rx.try_recv().is_ok(),
897            "fatal error must send kill signal"
898        );
899
900        // Non-fatal: INVALID_CLOCK
901        handler(
902            br#"{"op":"status","statusCode":"503","errorCode":"INVALID_CLOCK","connectionClosed":true}"#,
903        );
904        assert!(
905            race_fatal_rx.try_recv().is_err(),
906            "non-fatal error must not send kill signal"
907        );
908    }
909}
nautilus_betfair/stream/client.rs

nautilus_betfair/stream/
client.rs
