nautilus_tardis/
data.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//! Tardis data client for streaming replay or live data into the engine.
17
18use std::{
19    sync::{
20        Arc,
21        atomic::{AtomicBool, Ordering},
22    },
23    time::Duration,
24};
25
26use ahash::{AHashMap, AHashSet};
27use futures_util::{SinkExt, StreamExt};
28use nautilus_common::{
29    clients::DataClient,
30    live::{runner::get_data_event_sender, sender::EventSender},
31    messages::{
32        DataEvent,
33        data::{
34            subscribe::{SubscribeFundingRates, SubscribeIndexPrices, SubscribeMarkPrices},
35            unsubscribe::{UnsubscribeFundingRates, UnsubscribeIndexPrices, UnsubscribeMarkPrices},
36        },
37    },
38};
39use nautilus_core::{consts::NAUTILUS_USER_AGENT, string::urlencoding};
40use nautilus_live::task::TaskGroup;
41use nautilus_model::{
42    data::Data,
43    identifiers::{ClientId, Venue},
44};
45use tokio_tungstenite::{
46    connect_async,
47    tungstenite::{
48        self,
49        client::IntoClientRequest,
50        http::{HeaderValue, header::USER_AGENT},
51    },
52};
53use tokio_util::sync::CancellationToken;
54
55use crate::{
56    common::{
57        consts::{
58            WS_HEARTBEAT_INTERVAL_SECS, WS_INITIAL_RECONNECT_DELAY_SECS,
59            WS_MAX_RECONNECT_DELAY_SECS,
60        },
61        enums::TardisDataType,
62        urls::resolve_ws_base_url,
63    },
64    config::{BookSnapshotOutput, TardisDataClientConfig},
65    http::TardisHttpClient,
66    machine::{
67        cache::DerivativeTickerCache,
68        client::determine_instrument_info,
69        message::WsMessage,
70        parse::{
71            parse_derivative_ticker_index_price, parse_derivative_ticker_mark_price,
72            parse_tardis_ws_message_data, parse_tardis_ws_message_funding_rate,
73        },
74        types::{TardisInstrumentKey, TardisInstrumentMiniInfo},
75    },
76};
77
78/// Tardis data client for streaming replay or live data into the platform.
79#[derive(Debug)]
80pub struct TardisDataClient {
81    client_id: ClientId,
82    config: TardisDataClientConfig,
83    is_connected: Arc<AtomicBool>,
84    cancellation_token: CancellationToken,
85    tasks: TaskGroup,
86    data_sender: EventSender<DataEvent>,
87}
88
89impl TardisDataClient {
90    /// Creates a new [`TardisDataClient`] instance.
91    ///
92    /// # Errors
93    ///
94    /// Returns an error if the data event sender is not initialized.
95    pub fn new(client_id: ClientId, config: TardisDataClientConfig) -> anyhow::Result<Self> {
96        let data_sender = get_data_event_sender();
97
98        let tasks = TaskGroup::new();
99
100        Ok(Self {
101            client_id,
102            config,
103            is_connected: Arc::new(AtomicBool::new(false)),
104            cancellation_token: tasks.cancellation_token(),
105            tasks,
106            data_sender,
107        })
108    }
109
110    /// Returns `true` if the client is configured for live streaming mode.
111    fn is_stream_mode(&self) -> bool {
112        self.config.options.is_empty() && !self.config.stream_options.is_empty()
113    }
114
115    /// Builds the WebSocket URL for connecting to the Tardis Machine Server.
116    ///
117    /// Ensures `derivative_ticker` is included in the data types for each
118    /// option set so that mark price, index price, and funding rate events
119    /// are available without requiring manual configuration.
120    fn build_ws_url(&self, base_url: &str) -> anyhow::Result<String> {
121        let deriv = TardisDataType::DerivativeTicker.as_tardis_str();
122
123        if self.is_stream_mode() {
124            let mut options = self.config.stream_options.clone();
125            for opt in &mut options {
126                if !opt.data_types.iter().any(|dt| dt == deriv) {
127                    opt.data_types.push(deriv.to_string());
128                }
129            }
130            let options_json = serde_json::to_string(&options)?;
131            Ok(format!(
132                "{base_url}/ws-stream-normalized?options={}",
133                urlencoding::encode(&options_json)
134            ))
135        } else {
136            let mut options = self.config.options.clone();
137            for opt in &mut options {
138                if !opt.data_types.iter().any(|dt| dt == deriv) {
139                    opt.data_types.push(deriv.to_string());
140                }
141            }
142            let options_json = serde_json::to_string(&options)?;
143            Ok(format!(
144                "{base_url}/ws-replay-normalized?options={}",
145                urlencoding::encode(&options_json)
146            ))
147        }
148    }
149
150    /// Spawns the WebSocket message processing loop using an already-connected
151    /// stream. The initial handshake happens in `connect()` so callers get an
152    /// error if the first connection fails. In stream mode the spawned task
153    /// handles subsequent reconnections automatically.
154    fn spawn_ws_task(
155        &self,
156        ws_stream: tokio_tungstenite::WebSocketStream<
157            tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
158        >,
159        url: String,
160        instrument_map: AHashMap<TardisInstrumentKey, Arc<TardisInstrumentMiniInfo>>,
161        book_snapshot_output: BookSnapshotOutput,
162        extract_bbo_as_quotes: bool,
163        is_stream_mode: bool,
164    ) -> anyhow::Result<()> {
165        let sender = self.data_sender.clone();
166        let cancel = self.cancellation_token.clone();
167        let connected = self.is_connected.clone();
168
169        let future = async move {
170            let mut reconnect_delay = Duration::from_secs(WS_INITIAL_RECONNECT_DELAY_SECS);
171            let instrument_map = instrument_map;
172
173            // Process the initial (already-connected) stream
174            let should_reconnect = Self::run_ws_session(
175                ws_stream,
176                &cancel,
177                &sender,
178                &instrument_map,
179                &book_snapshot_output,
180                extract_bbo_as_quotes,
181            )
182            .await;
183
184            if !should_reconnect || !is_stream_mode || cancel.is_cancelled() {
185                connected.store(false, Ordering::Release);
186                return;
187            }
188
189            // Mark disconnected while reconnecting so health checks see the outage
190            connected.store(false, Ordering::Release);
191
192            // Reconnection loop (stream mode only)
193            loop {
194                log::warn!(
195                    "Stream disconnected, reconnecting in {}s",
196                    reconnect_delay.as_secs()
197                );
198
199                tokio::select! {
200                    () = tokio::time::sleep(reconnect_delay) => {}
201                    () = cancel.cancelled() => break,
202                }
203
204                reconnect_delay = std::cmp::min(
205                    reconnect_delay * 2,
206                    Duration::from_secs(WS_MAX_RECONNECT_DELAY_SECS),
207                );
208
209                // Reconnect WS first (critical path), then refresh instruments
210                let ws_request = match build_ws_request(&url) {
211                    Ok(request) => request,
212                    Err(e) => {
213                        log::error!("{e}");
214                        break;
215                    }
216                };
217
218                let ws_result = tokio::select! {
219                    result = connect_async(ws_request) => Some(result),
220                    () = cancel.cancelled() => None,
221                };
222
223                let Some(ws_result) = ws_result else {
224                    break;
225                };
226
227                match ws_result {
228                    Ok((ws_stream, _)) => {
229                        log::info!("Reconnected to Tardis Machine");
230                        connected.store(true, Ordering::Release);
231                        reconnect_delay = Duration::from_secs(WS_INITIAL_RECONNECT_DELAY_SECS);
232
233                        let should_reconnect = Self::run_ws_session(
234                            ws_stream,
235                            &cancel,
236                            &sender,
237                            &instrument_map,
238                            &book_snapshot_output,
239                            extract_bbo_as_quotes,
240                        )
241                        .await;
242
243                        if !should_reconnect || cancel.is_cancelled() {
244                            break;
245                        }
246
247                        connected.store(false, Ordering::Release);
248                    }
249                    Err(e) => {
250                        if cancel.is_cancelled() {
251                            break;
252                        }
253
254                        log::warn!(
255                            "Failed to reconnect to Tardis Machine: {e}, retrying in {}s",
256                            reconnect_delay.as_secs()
257                        );
258                    }
259                }
260            }
261
262            connected.store(false, Ordering::Release);
263        };
264
265        self.tasks
266            .spawn(future)
267            .map_err(|e| anyhow::anyhow!("failed to register Tardis stream task: {e}"))?;
268        Ok(())
269    }
270
271    /// Runs a single WebSocket session: starts heartbeat, processes messages,
272    /// and returns whether the caller should attempt reconnection.
273    async fn run_ws_session(
274        ws_stream: tokio_tungstenite::WebSocketStream<
275            tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
276        >,
277        cancel: &CancellationToken,
278        sender: &EventSender<DataEvent>,
279        instrument_map: &AHashMap<TardisInstrumentKey, Arc<TardisInstrumentMiniInfo>>,
280        book_snapshot_output: &BookSnapshotOutput,
281        extract_bbo_as_quotes: bool,
282    ) -> bool {
283        let (mut writer, mut reader) = ws_stream.split();
284
285        let heartbeat_token = cancel.child_token();
286        let heartbeat_signal = heartbeat_token.clone();
287
288        let heartbeat = async move {
289            let mut interval =
290                tokio::time::interval(Duration::from_secs(WS_HEARTBEAT_INTERVAL_SECS));
291            loop {
292                tokio::select! {
293                    _ = interval.tick() => {
294                        log::trace!("Sending PING");
295
296                        if let Err(e) = writer.send(tungstenite::Message::Ping(vec![].into())).await {
297                            log::debug!("Heartbeat send failed: {e}");
298                            break;
299                        }
300                    }
301                    () = heartbeat_signal.cancelled() => break,
302                }
303            }
304        };
305
306        let message_loop = async {
307            let should_reconnect = Self::run_ws_loop(
308                &mut reader,
309                cancel,
310                sender,
311                instrument_map,
312                book_snapshot_output,
313                extract_bbo_as_quotes,
314            )
315            .await;
316            heartbeat_token.cancel();
317            should_reconnect
318        };
319        let (should_reconnect, ()) = tokio::join!(message_loop, heartbeat);
320        should_reconnect
321    }
322
323    /// Extracts and sends all data events from a `DerivativeTicker` message:
324    /// funding rate, mark price, and index price. Only emits events when values
325    /// change from the previous update. Returns `false` if the channel is broken
326    /// and the caller should exit the loop.
327    fn send_derivative_ticker_events(
328        ws_msg: &WsMessage,
329        info: &Arc<TardisInstrumentMiniInfo>,
330        sender: &EventSender<DataEvent>,
331        cache: &mut DerivativeTickerCache,
332    ) -> bool {
333        if let Some(funding) = parse_tardis_ws_message_funding_rate(ws_msg.clone(), info)
334            && cache.should_emit_funding_rate(&funding)
335            && sender.send(DataEvent::FundingRate(funding)).is_err()
336        {
337            return false;
338        }
339
340        if let WsMessage::DerivativeTicker(msg) = ws_msg {
341            if let Ok(Some(mark_price)) =
342                parse_derivative_ticker_mark_price(msg, info.instrument_id, info.price_precision)
343                && cache.should_emit_mark_price(&mark_price)
344                && sender
345                    .send(DataEvent::Data(Data::MarkPrice(mark_price)))
346                    .is_err()
347            {
348                return false;
349            }
350
351            if let Ok(Some(index_price)) =
352                parse_derivative_ticker_index_price(msg, info.instrument_id, info.price_precision)
353                && cache.should_emit_index_price(&index_price)
354                && sender
355                    .send(DataEvent::Data(Data::IndexPrice(index_price)))
356                    .is_err()
357            {
358                return false;
359            }
360        }
361
362        true
363    }
364
365    /// Processes WebSocket messages until the stream ends, an error occurs, or
366    /// the cancellation token fires. Returns `true` if the caller should attempt
367    /// reconnection (stream mode only).
368    async fn run_ws_loop(
369        reader: &mut futures_util::stream::SplitStream<
370            tokio_tungstenite::WebSocketStream<
371                tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
372            >,
373        >,
374        cancel: &CancellationToken,
375        sender: &EventSender<DataEvent>,
376        instrument_map: &AHashMap<TardisInstrumentKey, Arc<TardisInstrumentMiniInfo>>,
377        book_snapshot_output: &BookSnapshotOutput,
378        extract_bbo_as_quotes: bool,
379    ) -> bool {
380        let mut ticker_cache = DerivativeTickerCache::default();
381
382        loop {
383            let msg = tokio::select! {
384                msg = reader.next() => msg,
385                () = cancel.cancelled() => {
386                    log::debug!("Stream task cancelled");
387                    return false;
388                }
389            };
390
391            match msg {
392                Some(Ok(tungstenite::Message::Text(text))) => {
393                    match serde_json::from_str::<WsMessage>(&text) {
394                        Ok(ws_msg) => {
395                            if matches!(ws_msg, WsMessage::Disconnect(_)) {
396                                log::debug!("Received disconnect message");
397                                continue;
398                            }
399
400                            let info = determine_instrument_info(&ws_msg, instrument_map);
401
402                            if let Some(info) = info {
403                                if matches!(ws_msg, WsMessage::DerivativeTicker(_)) {
404                                    if !Self::send_derivative_ticker_events(
405                                        &ws_msg,
406                                        &info,
407                                        sender,
408                                        &mut ticker_cache,
409                                    ) {
410                                        return false;
411                                    }
412                                } else {
413                                    let data = parse_tardis_ws_message_data(
414                                        ws_msg,
415                                        &info,
416                                        book_snapshot_output,
417                                        extract_bbo_as_quotes,
418                                    );
419
420                                    for data in data {
421                                        if let Err(e) = sender.send(DataEvent::Data(data)) {
422                                            log::error!("Failed to send data event: {e}");
423                                            return false;
424                                        }
425                                    }
426                                }
427                            }
428                        }
429                        Err(e) => {
430                            log::error!("Failed to deserialize message: {e}");
431                        }
432                    }
433                }
434                Some(Ok(tungstenite::Message::Close(frame))) => {
435                    if let Some(frame) = frame {
436                        log::debug!("WebSocket closed: {} {}", frame.code, frame.reason);
437                    } else {
438                        log::debug!("WebSocket closed");
439                    }
440                    return true;
441                }
442                Some(Ok(_)) => {}
443                Some(Err(e)) => {
444                    log::warn!("WebSocket error: {e}");
445                    return true;
446                }
447                None => {
448                    log::debug!("Stream ended");
449                    return true;
450                }
451            }
452        }
453    }
454}
455
456#[async_trait::async_trait(?Send)]
457impl DataClient for TardisDataClient {
458    fn client_id(&self) -> ClientId {
459        self.client_id
460    }
461
462    fn venue(&self) -> Option<Venue> {
463        None // Tardis is multi-venue
464    }
465
466    fn start(&mut self) -> anyhow::Result<()> {
467        log::info!("Starting {}", self.client_id);
468        Ok(())
469    }
470
471    fn stop(&mut self) -> anyhow::Result<()> {
472        log::info!("Stopping {}", self.client_id);
473        self.tasks.begin_shutdown();
474        self.is_connected.store(false, Ordering::Release);
475        Ok(())
476    }
477
478    fn reset(&mut self) -> anyhow::Result<()> {
479        self.tasks.begin_shutdown();
480        self.is_connected.store(false, Ordering::Release);
481        Ok(())
482    }
483
484    fn dispose(&mut self) -> anyhow::Result<()> {
485        self.stop()
486    }
487
488    fn is_connected(&self) -> bool {
489        self.is_connected.load(Ordering::Acquire)
490    }
491
492    fn is_disconnected(&self) -> bool {
493        !self.is_connected()
494    }
495
496    fn subscribe_mark_prices(&mut self, _cmd: SubscribeMarkPrices) -> anyhow::Result<()> {
497        Ok(())
498    }
499
500    fn subscribe_index_prices(&mut self, _cmd: SubscribeIndexPrices) -> anyhow::Result<()> {
501        Ok(())
502    }
503
504    fn subscribe_funding_rates(&mut self, _cmd: SubscribeFundingRates) -> anyhow::Result<()> {
505        Ok(())
506    }
507
508    fn unsubscribe_mark_prices(&mut self, _cmd: &UnsubscribeMarkPrices) -> anyhow::Result<()> {
509        Ok(())
510    }
511
512    fn unsubscribe_index_prices(&mut self, _cmd: &UnsubscribeIndexPrices) -> anyhow::Result<()> {
513        Ok(())
514    }
515
516    fn unsubscribe_funding_rates(&mut self, _cmd: &UnsubscribeFundingRates) -> anyhow::Result<()> {
517        Ok(())
518    }
519
520    async fn connect(&mut self) -> anyhow::Result<()> {
521        if self.is_connected() && self.tasks.is_open() {
522            return Ok(());
523        }
524
525        if self.config.options.is_empty() && self.config.stream_options.is_empty() {
526            anyhow::bail!("Either replay `options` or `stream_options` must be provided");
527        }
528
529        if !self.tasks.is_open() {
530            self.tasks
531                .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2))
532                .await
533                .map_err(|e| anyhow::anyhow!("Failed to terminate Tardis tasks: {e}"))?;
534            self.tasks
535                .start_generation()
536                .map_err(|e| anyhow::anyhow!("Failed to start Tardis task generation: {e}"))?;
537            self.cancellation_token = self.tasks.cancellation_token();
538        }
539
540        let is_stream_mode = self.is_stream_mode();
541        let book_snapshot_output = self.config.book_snapshot_output.clone();
542        let extract_bbo_as_quotes = self.config.extract_bbo_as_quotes;
543
544        let http_client = TardisHttpClient::new(
545            self.config
546                .api_key
547                .as_ref()
548                .map(|value| value.expose_secret()),
549            self.config
550                .tardis_http_url
551                .as_ref()
552                .map(|value| value.expose_secret()),
553            None,
554            self.config.normalize_symbols,
555            self.config
556                .proxy_url
557                .as_ref()
558                .map(|value| value.expose_secret().to_owned()),
559        )?;
560
561        let exchanges: AHashSet<_> = if is_stream_mode {
562            self.config
563                .stream_options
564                .iter()
565                .map(|opt| opt.exchange)
566                .collect()
567        } else {
568            self.config.options.iter().map(|opt| opt.exchange).collect()
569        };
570
571        let base_url = resolve_ws_base_url(
572            self.config
573                .tardis_ws_url
574                .as_ref()
575                .map(|value| value.expose_secret()),
576        )?;
577        let (instrument_map, instruments) = http_client
578            .bootstrap_instruments(&exchanges)
579            .await
580            .map_err(|e| anyhow::anyhow!("Failed to bootstrap instruments: {e}"))?;
581
582        for instrument in instruments {
583            if let Err(e) = self.data_sender.send(DataEvent::Instrument(instrument)) {
584                log::error!("Failed to send instrument event: {e}");
585            }
586        }
587
588        let url = self.build_ws_url(&base_url)?;
589
590        let mode_label = if is_stream_mode { "stream" } else { "replay" };
591        log::info!("Connecting to Tardis Machine {mode_label}");
592        log::debug!("URL: {url}");
593
594        let (ws_stream, _) = connect_async(build_ws_request(&url)?)
595            .await
596            .map_err(|e| anyhow::anyhow!("Failed to connect to Tardis Machine: {e}"))?;
597
598        log::info!("Connected to Tardis Machine");
599
600        if let Err(e) = self.spawn_ws_task(
601            ws_stream,
602            url,
603            instrument_map,
604            book_snapshot_output,
605            extract_bbo_as_quotes,
606            is_stream_mode,
607        ) {
608            self.tasks.begin_shutdown();
609            if let Err(teardown_error) = self
610                .tasks
611                .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2))
612                .await
613            {
614                return Err(e.context(format!("Tardis startup teardown failed: {teardown_error}")));
615            }
616            return Err(e);
617        }
618        self.is_connected.store(true, Ordering::Release);
619
620        log::info!("Connected: {}", self.client_id);
621        Ok(())
622    }
623
624    async fn disconnect(&mut self) -> anyhow::Result<()> {
625        self.tasks.begin_shutdown();
626        let had_tasks = !self.tasks.is_empty();
627        let tasks_result = self
628            .tasks
629            .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2))
630            .await
631            .map_err(|e| anyhow::anyhow!("Failed to terminate Tardis tasks: {e}"));
632
633        if had_tasks {
634            log::info!("Disconnected: {}", self.client_id);
635        }
636
637        self.is_connected.store(false, Ordering::Release);
638
639        tasks_result
640    }
641}
642
643/// Builds a Tardis Machine WebSocket handshake request carrying the Nautilus
644/// user agent.
645fn build_ws_request(url: &str) -> anyhow::Result<tungstenite::http::Request<()>> {
646    let mut request = url
647        .into_client_request()
648        .map_err(|e| anyhow::anyhow!("Failed to build Tardis Machine WebSocket request: {e}"))?;
649    request
650        .headers_mut()
651        .insert(USER_AGENT, HeaderValue::from_static(NAUTILUS_USER_AGENT));
652    Ok(request)
653}
654
655#[cfg(test)]
656mod tests {
657    use jiff::civil::Date;
658    use nautilus_common::live::runner::set_data_event_sender;
659    use rstest::rstest;
660
661    use super::*;
662    use crate::{
663        common::{consts::TARDIS_CLIENT_ID, enums::TardisExchange},
664        config::TardisDataClientConfig,
665        machine::types::ReplayNormalizedRequestOptions,
666    };
667
668    #[rstest]
669    fn test_build_ws_request_rejects_invalid_url() {
670        assert!(build_ws_request("not a url").is_err());
671    }
672
673    fn setup_test_env() {
674        use std::cell::OnceCell;
675
676        thread_local! {
677            static INIT: OnceCell<()> = const { OnceCell::new() };
678        }
679
680        INIT.with(|cell| {
681            cell.get_or_init(|| {
682                let (sender, _receiver) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
683                set_data_event_sender(sender);
684            });
685        });
686    }
687
688    #[rstest]
689    fn test_build_ws_url_injects_derivative_ticker() {
690        setup_test_env();
691
692        let config = TardisDataClientConfig {
693            options: vec![ReplayNormalizedRequestOptions {
694                exchange: TardisExchange::BinanceFutures,
695                symbols: Some(vec!["BTCUSDT".to_string()]),
696                from: Date::new(2024, 1, 1).unwrap(),
697                to: Date::new(2024, 1, 2).unwrap(),
698                data_types: vec!["trade".to_string()],
699                with_disconnect_messages: Some(false),
700            }],
701            ..Default::default()
702        };
703
704        let client = TardisDataClient::new(*TARDIS_CLIENT_ID, config).unwrap();
705        let url = client.build_ws_url("ws://localhost:8001").unwrap();
706
707        assert!(
708            url.contains("derivative_ticker"),
709            "URL should contain derivative_ticker but was: {url}"
710        );
711        assert!(url.contains("trade"), "URL should still contain trade");
712    }
713
714    #[rstest]
715    fn test_build_ws_url_does_not_duplicate_derivative_ticker() {
716        setup_test_env();
717
718        let config = TardisDataClientConfig {
719            options: vec![ReplayNormalizedRequestOptions {
720                exchange: TardisExchange::BinanceFutures,
721                symbols: Some(vec!["BTCUSDT".to_string()]),
722                from: Date::new(2024, 1, 1).unwrap(),
723                to: Date::new(2024, 1, 2).unwrap(),
724                data_types: vec!["trade".to_string(), "derivative_ticker".to_string()],
725                with_disconnect_messages: Some(false),
726            }],
727            ..Default::default()
728        };
729
730        let client = TardisDataClient::new(*TARDIS_CLIENT_ID, config).unwrap();
731        let ws_url = client.build_ws_url("ws://localhost:8001").unwrap();
732
733        let decoded = urlencoding::decode(ws_url.split("options=").nth(1).unwrap()).unwrap();
734        let count = decoded.matches("derivative_ticker").count();
735        assert_eq!(count, 1, "derivative_ticker should appear exactly once");
736    }
737
738    #[rstest]
739    fn test_stop_marks_client_disconnected_synchronously() {
740        setup_test_env();
741
742        let mut client =
743            TardisDataClient::new(*TARDIS_CLIENT_ID, TardisDataClientConfig::default()).unwrap();
744        client.is_connected.store(true, Ordering::Release);
745
746        client.stop().unwrap();
747
748        assert!(client.is_disconnected());
749    }
750
751    #[rstest]
752    fn test_reset_marks_client_disconnected_synchronously() {
753        setup_test_env();
754
755        let mut client =
756            TardisDataClient::new(*TARDIS_CLIENT_ID, TardisDataClientConfig::default()).unwrap();
757        client.is_connected.store(true, Ordering::Release);
758
759        client.reset().unwrap();
760
761        assert!(client.is_disconnected());
762    }
763}
nautilus_tardis/data.rs

nautilus_tardis/
data.rs
