nautilus_coinapi/websocket/
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//! CoinAPI WebSocket V1 connection with whole-scope replacement and replay.
17//!
18//! CoinAPI has no subscription acknowledgements. The connection authenticates through the
19//! `X-CoinAPI-Key` upgrade header, and the server validates it when the first `hello` arrives.
20//! A heartbeat or accepted data frame confirms the session and its subscriptions; an `error` frame
21//! is terminal and retires the connection instead of reconnecting.
22//!
23//! A server `reconnect` notice is only logged. Reconnecting early would open a second connection
24//! while the old one is open, which a one-connection tier rejects; the server's close triggers the
25//! normal reconnect instead.
26
27use std::{
28    fmt::Debug,
29    sync::{
30        Arc, Weak,
31        atomic::{AtomicBool, AtomicU64, Ordering},
32    },
33};
34
35use nautilus_core::string::secret::SecretString;
36use nautilus_network::{
37    Message, RECONNECTED, dst,
38    websocket::{
39        AuthTracker, MessageHandler, SubscriptionState, TransportBackend, WebSocketClient,
40        WebSocketConfig, proxy::ProxyTarget, split_topic,
41    },
42};
43use parking_lot::Mutex;
44use url::Url;
45use ustr::Ustr;
46
47use super::{
48    error::Error,
49    messages::{Channel, CoinApiWsMessage, DataType, Frame, Hello, ScopeChange, Subscription},
50};
51use crate::{
52    book::sync::BookSubscriptions,
53    common::{consts::COINAPI_NAUTILUS_REFERRER, credential::is_valid_api_key},
54};
55
56/// Primary CoinAPI Market Data WebSocket V1 endpoint.
57pub const WS_URL: &str = "wss://ws.coinapi.io/v1/";
58
59// The server sends a heartbeat after each second of silence
60const IDLE_TIMEOUT_MS: u64 = 5_000;
61const RECONNECT_DELAY_INITIAL_MS: u64 = 1_000;
62const RECONNECT_DELAY_MAX_MS: u64 = 30_000;
63const RECONNECT_BACKOFF_FACTOR: f64 = 2.0;
64const RECONNECT_JITTER_MS: u64 = 500;
65const TOPIC_DELIMITER: char = ':';
66const PERIOD_DELIMITER: char = '.';
67const PAYLOAD_EXCERPT_BYTES: usize = 128;
68
69// Live probes found that back-to-back hellos can leave a connection streaming nothing, and that
70// CoinAPI closes a connection once hellos exceed about one every two seconds per IP
71const HELLO_INTERVAL_MIN: dst::time::Duration = dst::time::Duration::from_secs(2);
72const RESTART_INTERVAL_MIN: dst::time::Duration = dst::time::Duration::from_secs(5);
73
74/// Transport configuration for the raw WebSocket client.
75#[derive(Debug, Clone)]
76pub struct CoinApiWebSocketConfig {
77    /// Endpoint URL; accepts local simulated peers.
78    pub url: String,
79    /// Transport implementation; simulated peers require Tungstenite.
80    pub backend: TransportBackend,
81    /// Optional explicit proxy URL for the WebSocket transport.
82    pub proxy_url: Option<SecretString>,
83}
84
85impl Default for CoinApiWebSocketConfig {
86    fn default() -> Self {
87        Self {
88            url: WS_URL.into(),
89            backend: TransportBackend::default(),
90            proxy_url: None,
91        }
92    }
93}
94
95/// Authenticated, data-only CoinAPI WebSocket client using one connection.
96///
97/// CoinAPI tiers can limit an organization to one concurrent connection. Subscription changes send
98/// a `hello` that replaces the whole session scope; changes applied together share one `hello`,
99/// and the client sends at most one every two seconds, including across connection recreation.
100/// Because CoinAPI combines every listed data type, symbol, and period, the client drops frames
101/// outside the subscribed pairs; the provider still bills for them. A scope can list only one
102/// book type. Removing the last subscription closes the connection, and the next subscription
103/// opens a new one. Reconnects replay the scope and emit [`CoinApiWsMessage::Reconnected`].
104#[derive(Debug)]
105pub struct CoinApiWebSocketClient {
106    api_key: SecretString,
107    config: CoinApiWebSocketConfig,
108    connection: Arc<tokio::sync::Mutex<Option<Connection>>>,
109    hello_at: Arc<Mutex<Option<dst::time::Instant>>>,
110    next_connection_id: AtomicU64,
111    books: Option<Arc<BookSubscriptions>>,
112    out_tx: tokio::sync::mpsc::UnboundedSender<CoinApiWsMessage>,
113}
114
115impl CoinApiWebSocketClient {
116    /// Creates a client and the receiver for its messages without connecting.
117    ///
118    /// # Errors
119    ///
120    /// Returns an error if the credentials, endpoint, or proxy URL are invalid.
121    pub fn new(
122        api_key: SecretString,
123        config: CoinApiWebSocketConfig,
124    ) -> Result<(Self, tokio::sync::mpsc::UnboundedReceiver<CoinApiWsMessage>), Error> {
125        if !is_valid_api_key(api_key.expose_secret()) {
126            return Err(Error::Validation(
127                "API key must contain visible ASCII characters".into(),
128            ));
129        }
130
131        let url = Url::parse(&config.url).map_err(|_| Error::Validation("invalid URL".into()))?;
132
133        if !matches!(url.scheme(), "ws" | "wss")
134            || url.host_str().is_none()
135            || !url.username().is_empty()
136            || url.password().is_some()
137            || url.query().is_some()
138            || url.fragment().is_some()
139        {
140            return Err(Error::Validation(
141                "URL must be WS(S) without credentials, query, or fragment".into(),
142            ));
143        }
144
145        if let Some(proxy) = &config.proxy_url {
146            ProxyTarget::parse(proxy.expose_secret())
147                .map_err(|_| Error::Validation("proxy URL must be a valid HTTP(S) URL".into()))?;
148        }
149
150        let (out_tx, out_rx) = tokio::sync::mpsc::unbounded_channel();
151
152        let client = Self {
153            api_key,
154            config,
155            connection: Arc::new(tokio::sync::Mutex::new(None)),
156            hello_at: Arc::new(Mutex::new(None)),
157            next_connection_id: AtomicU64::new(0),
158            books: None,
159            out_tx,
160        };
161
162        Ok((client, out_rx))
163    }
164
165    /// Adds a subscription, connecting on first use.
166    ///
167    /// Returns once the new scope is queued for sending; the first data frame confirms it.
168    ///
169    /// # Errors
170    ///
171    /// Returns an error if validation, connecting, or sending fails. A book type other than the one
172    /// already subscribed is a validation error. A failed send leaves the pair unsubscribed.
173    pub async fn subscribe(&self, subscription: Subscription) -> Result<(), Error> {
174        self.apply_one(ScopeChange::Subscribe(subscription)).await
175    }
176
177    /// Removes a subscription.
178    ///
179    /// CoinAPI does not acknowledge the replacement scope, so frames still in flight for the
180    /// pair are dropped locally.
181    ///
182    /// # Errors
183    ///
184    /// Returns an error if validation or sending fails. A failed send keeps the pair subscribed.
185    pub async fn unsubscribe(&self, subscription: Subscription) -> Result<(), Error> {
186        self.apply_one(ScopeChange::Unsubscribe(subscription)).await
187    }
188
189    /// Applies scope changes in order and sends one `hello` for the resulting scope.
190    ///
191    /// Connects when a subscribe finds no connection, and closes the connection when the scope
192    /// becomes empty, since an empty `hello` is a terminal protocol error. The `hello` waits until
193    /// two seconds after the client's previous one. Returns once the scope is queued for
194    /// sending, with each change rejected by validation, including a book type other than the
195    /// one already subscribed.
196    ///
197    /// # Errors
198    ///
199    /// Returns an error if connecting or sending fails; every change in the batch is reverted.
200    pub async fn apply(&self, changes: &[ScopeChange]) -> Result<Vec<(ScopeChange, Error)>, Error> {
201        self.apply_registered(changes, &[]).await
202    }
203
204    pub(crate) fn bind_books(&mut self, books: Arc<BookSubscriptions>) {
205        self.books = Some(books);
206    }
207
208    pub(crate) async fn apply_registered(
209        &self,
210        changes: &[ScopeChange],
211        registrations: &[Option<Arc<AtomicBool>>],
212    ) -> Result<Vec<(ScopeChange, Error)>, Error> {
213        let mut rejected = Vec::new();
214        let mut valid = Vec::with_capacity(changes.len());
215
216        for (index, change) in changes.iter().enumerate() {
217            match change.subscription().validate() {
218                Ok(()) => valid.push((*change, registrations.get(index).cloned().flatten())),
219                Err(e) => rejected.push((*change, e)),
220            }
221        }
222
223        let mut guard = self.connection.lock().await;
224        let fresh = guard.is_none();
225
226        if fresh
227            && !valid
228                .iter()
229                .any(|(change, _)| matches!(change, ScopeChange::Subscribe(_)))
230        {
231            return Ok(rejected);
232        }
233
234        if fresh {
235            let connection = self.connect().await?;
236            connection.auth.begin();
237            *guard = Some(connection);
238        }
239
240        let Some(connection) = guard.as_mut() else {
241            return Ok(rejected);
242        };
243
244        let result = connection.apply_scope(&valid, &mut rejected).await;
245        if (matches!(result, Ok(false)) || (fresh && result.is_err()))
246            && let Some(connection) = guard.take()
247        {
248            connection.close().await;
249        }
250
251        result.map(|_| rejected)
252    }
253
254    async fn apply_one(&self, change: ScopeChange) -> Result<(), Error> {
255        match self.apply(&[change]).await?.pop() {
256            Some((_, e)) => Err(e),
257            None => Ok(()),
258        }
259    }
260
261    /// Restarts a subscribed book pair's stream so CoinAPI sends it a fresh snapshot.
262    ///
263    /// CoinAPI replays nothing for a repeated scope, but restarts a pair's sequence with a
264    /// snapshot when the pair leaves the scope and returns. The client sends the scope without the
265    /// pair's symbol, then the full scope two seconds later, so other symbols keep streaming. When
266    /// no other symbol is subscribed, the removal scope keeps the symbol's other channels instead,
267    /// or lists only its trades when the book is the whole scope; the client drops those
268    /// unsubscribed trades. The connection stays open, since a one-connection tier rejects a second
269    /// one.
270    ///
271    /// Restarts on a connection start at least five seconds apart, because CoinAPI closes a
272    /// connection that exceeds its per-IP `hello` rate. Subscription changes wait while a restart
273    /// is pending. Calls `paused` as soon as a scope without the pair's symbol is queued, since
274    /// that pauses the symbol's other channels; the call happens even if the restart is then
275    /// cancelled. Returns once both scopes are queued for sending.
276    ///
277    /// # Errors
278    ///
279    /// Returns an error if the pair is not a subscribed book pair or sending fails.
280    pub async fn resubscribe(
281        &self,
282        subscription: Subscription,
283        paused: impl FnOnce(),
284    ) -> Result<(), Error> {
285        subscription.validate()?;
286
287        if !subscription.channel.data_type.is_book() {
288            return Err(Error::Validation(format!(
289                "{} has no snapshot to restart",
290                topic(&subscription)
291            )));
292        }
293
294        let mut guard = self.connection.lock().await;
295
296        let Some(connection) = guard
297            .as_mut()
298            .filter(|connection| is_subscribed(&connection.subscriptions, &subscription))
299        else {
300            return Err(Error::Validation(format!(
301                "{} is not subscribed",
302                topic(&subscription)
303            )));
304        };
305
306        if let Some(restarted_at) = connection.restarted_at {
307            dst::time::sleep_until(restarted_at + RESTART_INTERVAL_MIN).await;
308        }
309
310        let scope = scope(&connection.subscriptions);
311        let removal = removal_scope(&scope, &subscription);
312        let pauses = removal
313            .iter()
314            .all(|other| other.symbol_id != subscription.symbol_id);
315
316        let removed = send_paced(connection, &removal).await;
317        connection.restarted_at = *connection.hello_at.lock();
318        removed?;
319
320        if pauses {
321            paused();
322        }
323
324        send_paced(connection, &scope).await
325    }
326
327    /// Closes the connection and forgets its subscriptions; repeated calls are safe.
328    ///
329    /// Dropping the client also closes the connection.
330    pub async fn close(&self) {
331        let mut guard = self.connection.lock().await;
332        if let Some(connection) = guard.take() {
333            connection.close().await;
334        }
335    }
336
337    async fn connect(&self) -> Result<Connection, Error> {
338        let config = WebSocketConfig::builder()
339            .url(self.config.url.clone())
340            .headers(vec![
341                (
342                    "X-CoinAPI-Key".into(),
343                    self.api_key.expose_secret().to_owned(),
344                ),
345                ("X-Referrer".into(), COINAPI_NAUTILUS_REFERRER.into()),
346            ])
347            .idle_timeout_ms(IDLE_TIMEOUT_MS)
348            .reconnect_delay_initial_ms(RECONNECT_DELAY_INITIAL_MS)
349            .reconnect_delay_max_ms(RECONNECT_DELAY_MAX_MS)
350            .reconnect_backoff_factor(RECONNECT_BACKOFF_FACTOR)
351            .reconnect_jitter_ms(RECONNECT_JITTER_MS)
352            .backend(self.config.backend)
353            .maybe_proxy_url(
354                self.config
355                    .proxy_url
356                    .as_ref()
357                    .map(|url| url.expose_secret().to_owned()),
358            )
359            .build()
360            .map_err(|e| Error::Validation(e.to_string()))?;
361
362        let (raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel();
363        let handler: MessageHandler = Arc::new(move |message| {
364            let _ = raw_tx.send(message);
365        });
366
367        let client = WebSocketClient::builder()
368            .config(config)
369            .message_handler(handler)
370            .connect()
371            .await?;
372
373        let auth = AuthTracker::new();
374        client.set_auth_tracker(auth.clone(), false);
375        let client = Arc::new(client);
376        let subscriptions = SubscriptionState::new(TOPIC_DELIMITER);
377        let id = self.next_connection_id.fetch_add(1, Ordering::Relaxed);
378
379        let handler = ConnectionHandler {
380            id,
381            client: Arc::clone(&client),
382            subscriptions: subscriptions.clone(),
383            auth: auth.clone(),
384            connection: Arc::downgrade(&self.connection),
385            out_tx: self.out_tx.clone(),
386            api_key: self.api_key.clone(),
387            books: self.books.clone(),
388        };
389
390        let task = dst::task::spawn(handler.run(raw_rx));
391
392        Ok(Connection {
393            id,
394            client,
395            subscriptions,
396            auth,
397            task,
398            hello_at: Arc::clone(&self.hello_at),
399            restarted_at: None,
400        })
401    }
402}
403
404type RegisteredChange = (ScopeChange, Option<Arc<AtomicBool>>);
405
406struct Connection {
407    id: u64,
408    client: Arc<WebSocketClient>,
409    subscriptions: SubscriptionState,
410    auth: AuthTracker,
411    task: dst::task::JoinHandle<()>,
412    hello_at: Arc<Mutex<Option<dst::time::Instant>>>,
413    restarted_at: Option<dst::time::Instant>,
414}
415
416impl Debug for Connection {
417    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
418        f.debug_struct(stringify!(Connection))
419            .field("id", &self.id)
420            .field("subscriptions", &self.subscriptions)
421            .field("auth", &self.auth.auth_state())
422            .finish_non_exhaustive()
423    }
424}
425
426impl Connection {
427    async fn apply_scope(
428        &self,
429        changes: &[RegisteredChange],
430        rejected: &mut Vec<(ScopeChange, Error)>,
431    ) -> Result<bool, Error> {
432        let applied = self.change_scope(changes, rejected);
433        let scope = scope(&self.subscriptions);
434
435        if scope.is_empty() {
436            return Ok(false);
437        }
438
439        if applied.is_empty() {
440            return Ok(true);
441        }
442
443        if let Err(e) = send_paced(self, &scope).await {
444            revert(&self.subscriptions, &applied);
445            return Err(e);
446        }
447
448        self.commit_scope(changes, &applied);
449        Ok(true)
450    }
451
452    fn commit_scope(&self, changes: &[RegisteredChange], applied: &[ScopeChange]) {
453        // Retirement takes this same connection lock, so it observes every committed registration
454        for (change, registration) in changes {
455            if matches!(change, ScopeChange::Subscribe(_))
456                && applied.contains(change)
457                && let Some(registration) = registration
458            {
459                registration.store(true, Ordering::Relaxed);
460            }
461        }
462
463        for change in applied {
464            if let ScopeChange::Unsubscribe(subscription) = change {
465                self.subscriptions.confirm_unsubscribe(&topic(subscription));
466            }
467        }
468    }
469
470    fn change_scope(
471        &self,
472        changes: &[RegisteredChange],
473        rejected: &mut Vec<(ScopeChange, Error)>,
474    ) -> Vec<ScopeChange> {
475        let mut applied = Vec::with_capacity(changes.len());
476
477        for (change, _) in changes {
478            let change = *change;
479            let topic = topic(change.subscription());
480
481            match change {
482                ScopeChange::Subscribe(subscription) => {
483                    let scope = scope(&self.subscriptions);
484
485                    if let Err(e) = check_book_type(&scope, &subscription) {
486                        rejected.push((change, e));
487                    } else if self.subscriptions.try_mark_subscribe(&topic) {
488                        applied.push(change);
489                    }
490                }
491                ScopeChange::Unsubscribe(subscription) => {
492                    if is_subscribed(&self.subscriptions, &subscription) {
493                        self.subscriptions.mark_unsubscribe(&topic);
494                        applied.push(change);
495                    }
496                }
497            }
498        }
499
500        applied
501    }
502
503    async fn close(self) {
504        self.task.abort();
505        self.client.disconnect().await;
506    }
507}
508
509impl Drop for Connection {
510    fn drop(&mut self) {
511        self.task.abort();
512    }
513}
514
515// Holds the slot weakly so dropping the public client drops the connection and aborts this task
516struct ConnectionHandler {
517    id: u64,
518    client: Arc<WebSocketClient>,
519    subscriptions: SubscriptionState,
520    auth: AuthTracker,
521    connection: Weak<tokio::sync::Mutex<Option<Connection>>>,
522    out_tx: tokio::sync::mpsc::UnboundedSender<CoinApiWsMessage>,
523    api_key: SecretString,
524    books: Option<Arc<BookSubscriptions>>,
525}
526
527impl ConnectionHandler {
528    async fn run(self, mut raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>) {
529        while let Some(message) = raw_rx.recv().await {
530            let Message::Text(bytes) = message else {
531                continue;
532            };
533
534            if bytes.as_ref() == RECONNECTED.as_bytes() {
535                self.replay().await;
536                continue;
537            }
538
539            match Frame::decode(&bytes) {
540                Ok(Frame::Data(message)) => self.route(message),
541                Ok(Frame::Heartbeat) => self.confirm_session(),
542                Ok(Frame::Reconnect(notice)) => {
543                    log::info!(
544                        "CoinAPI will close the connection before {}",
545                        notice.before_time
546                    );
547                }
548                Ok(Frame::Error(error)) => {
549                    let message = error
550                        .message
551                        .replace(self.api_key.expose_secret(), "<redacted>");
552                    self.retire(message).await;
553                    return;
554                }
555                Ok(Frame::Unknown(kind)) => log::debug!("Ignoring CoinAPI frame type {kind}"),
556                Err(e) => {
557                    let excerpt =
558                        String::from_utf8_lossy(&bytes[..bytes.len().min(PAYLOAD_EXCERPT_BYTES)]);
559                    log::warn!("Failed to decode CoinAPI frame: {e}");
560                    log::debug!("Undecodable CoinAPI frame excerpt: {excerpt}");
561                }
562            }
563        }
564    }
565
566    fn route(&self, message: CoinApiWsMessage) {
567        let Some(subscription) = message.subscription() else {
568            return;
569        };
570
571        // The combined scope also delivers pairs nobody subscribed to
572        if !is_subscribed(&self.subscriptions, &subscription) {
573            log::trace!("Dropping unsubscribed CoinAPI {subscription:?}");
574            return;
575        }
576
577        self.subscriptions.confirm_subscribe(&topic(&subscription));
578        self.confirm_session();
579        let _ = self.out_tx.send(message);
580    }
581
582    fn confirm_session(&self) {
583        if !self.auth.is_authenticated() {
584            self.auth.succeed();
585        }
586    }
587
588    async fn replay(&self) {
589        let Some(connection) = self.connection.upgrade() else {
590            return;
591        };
592
593        let mut guard = connection.lock().await;
594
595        let Some(connection) = guard.as_mut().filter(|connection| connection.id == self.id) else {
596            return;
597        };
598
599        self.subscriptions.reset_after_reconnect();
600        self.auth.begin();
601
602        // A change in flight can reach the replacement socket first, so the replay waits its turn
603        match send_paced(connection, &scope(&self.subscriptions)).await {
604            Ok(()) => {
605                let _ = self.out_tx.send(CoinApiWsMessage::Reconnected);
606            }
607            Err(e) => log::error!("Failed to replay CoinAPI scope: {e}"),
608        }
609    }
610
611    async fn retire(&self, message: String) {
612        log::error!("CoinAPI rejected the session: {message}");
613        self.auth.fail(message.clone());
614
615        let Some(connection) = self.connection.upgrade() else {
616            return;
617        };
618
619        let mut guard = connection.lock().await;
620        self.client.disconnect().await;
621
622        if guard
623            .as_ref()
624            .is_some_and(|connection| connection.id == self.id)
625            && let Some(books) = &self.books
626        {
627            books.retire();
628        }
629
630        let _ = self.out_tx.send(CoinApiWsMessage::Error(message));
631
632        // Dropping the owned connection aborts this task, so nothing may await after it
633        if guard
634            .as_ref()
635            .is_some_and(|connection| connection.id == self.id)
636        {
637            guard.take();
638        }
639    }
640}
641
642// Waits out the client's `hello` interval, then sends `scope`
643async fn send_paced(connection: &Connection, scope: &[Subscription]) -> Result<(), Error> {
644    let hello_at = *connection.hello_at.lock();
645    if let Some(hello_at) = hello_at {
646        dst::time::sleep_until(hello_at + HELLO_INTERVAL_MIN).await;
647    }
648
649    // A caller cancelled during the write leaves the queued hello to the writer, so the interval
650    // starts before the write as well as after it
651    *connection.hello_at.lock() = Some(dst::time::Instant::now());
652    let result = send_scope(&connection.client, scope).await;
653    *connection.hello_at.lock() = Some(dst::time::Instant::now());
654    result
655}
656
657// Restores the subscription state from before a batch whose `hello` failed
658fn revert(subscriptions: &SubscriptionState, applied: &[ScopeChange]) {
659    for change in applied.iter().rev() {
660        let topic = topic(change.subscription());
661
662        match change {
663            ScopeChange::Subscribe(_) => {
664                subscriptions.mark_unsubscribe(&topic);
665                subscriptions.confirm_unsubscribe(&topic);
666            }
667            ScopeChange::Unsubscribe(_) => subscriptions.mark_subscribe(&topic),
668        }
669    }
670}
671
672// Writes the hello on the current connection only; the transport would otherwise buffer it
673// through a reconnect and send it on the replacement just ahead of the replayed scope
674async fn send_scope(client: &WebSocketClient, scope: &[Subscription]) -> Result<(), Error> {
675    if scope.is_empty() {
676        return Ok(());
677    }
678
679    let hello = serde_json::to_string(&Hello::new(scope))
680        .expect("CoinAPI hello contains only serializable fields");
681    client
682        .send_text_on_connection(hello, None, client.connection_epoch())
683        .await?;
684    Ok(())
685}
686
687fn scope(subscriptions: &SubscriptionState) -> Vec<Subscription> {
688    subscriptions
689        .all_topics()
690        .iter()
691        .filter_map(|topic| parse_topic(topic))
692        .collect()
693}
694
695// CoinAPI rejects a scope listing two book types with a terminal error frame
696fn check_book_type(scope: &[Subscription], subscription: &Subscription) -> Result<(), Error> {
697    let requested = subscription.channel.data_type;
698
699    if !requested.is_book() {
700        return Ok(());
701    }
702
703    match scope
704        .iter()
705        .map(|subscription| subscription.channel.data_type)
706        .find(|active| active.is_book() && *active != requested)
707    {
708        Some(active) => Err(Error::Validation(format!(
709            "scope already streams {}, so cannot add {}",
710            active.as_ref(),
711            requested.as_ref()
712        ))),
713        None => Ok(()),
714    }
715}
716
717// Leaves the book pair out of the effective scope, which combines every listed type and symbol.
718// An empty `hello` is a terminal error, so a book alone in the scope gives way to its trades.
719fn removal_scope(scope: &[Subscription], pair: &Subscription) -> Vec<Subscription> {
720    let others: Vec<Subscription> = scope
721        .iter()
722        .filter(|subscription| subscription.symbol_id != pair.symbol_id)
723        .copied()
724        .collect();
725
726    if !others.is_empty() {
727        return others;
728    }
729
730    let channels: Vec<Subscription> = scope
731        .iter()
732        .filter(|subscription| *subscription != pair)
733        .copied()
734        .collect();
735
736    if !channels.is_empty() {
737        return channels;
738    }
739
740    vec![Subscription {
741        channel: Channel {
742            data_type: DataType::Trade,
743            period_id: None,
744        },
745        symbol_id: pair.symbol_id,
746    }]
747}
748
749fn is_subscribed(subscriptions: &SubscriptionState, subscription: &Subscription) -> bool {
750    subscriptions.is_subscribed(
751        &Ustr::from(&channel_key(&subscription.channel)),
752        &subscription.symbol_id,
753    )
754}
755
756fn topic(subscription: &Subscription) -> String {
757    format!(
758        "{}{TOPIC_DELIMITER}{}",
759        channel_key(&subscription.channel),
760        subscription.symbol_id
761    )
762}
763
764fn channel_key(channel: &Channel) -> String {
765    match channel.period_id {
766        Some(period) => format!("{}{PERIOD_DELIMITER}{period}", channel.data_type.as_ref()),
767        None => channel.data_type.as_ref().to_owned(),
768    }
769}
770
771fn parse_topic(topic: &str) -> Option<Subscription> {
772    let (channel, symbol_id) = split_topic(topic, TOPIC_DELIMITER);
773
774    let (data_type, period_id) = match channel.split_once(PERIOD_DELIMITER) {
775        Some((data_type, period)) => (data_type, Some(Ustr::from(period))),
776        None => (channel, None),
777    };
778
779    Some(Subscription {
780        channel: Channel {
781            data_type: data_type.parse().ok()?,
782            period_id,
783        },
784        symbol_id: Ustr::from(symbol_id?),
785    })
786}
787
788#[cfg(test)]
789mod tests {
790    #[cfg(not(all(feature = "simulation", madsim)))]
791    use futures_util::{SinkExt, StreamExt};
792    #[cfg(not(all(feature = "simulation", madsim)))]
793    use nautilus_network::error::SendError;
794    use rstest::rstest;
795
796    use super::*;
797
798    fn sub(data_type: DataType, symbol_id: &str) -> Subscription {
799        Subscription {
800            channel: Channel {
801                data_type,
802                period_id: None,
803            },
804            symbol_id: symbol_id.into(),
805        }
806    }
807
808    #[rstest]
809    #[case::other_symbols_remain(
810        vec![sub(DataType::Book, "A"), sub(DataType::Trade, "A"), sub(DataType::Book, "B")],
811        vec![sub(DataType::Book, "B")]
812    )]
813    #[case::only_symbol_keeps_other_channels(
814        vec![sub(DataType::Book, "A"), sub(DataType::Trade, "A"), sub(DataType::Quote, "A")],
815        vec![sub(DataType::Trade, "A"), sub(DataType::Quote, "A")]
816    )]
817    #[case::pair_is_whole_scope(vec![sub(DataType::Book, "A")], vec![sub(DataType::Trade, "A")])]
818    fn removal_scope_leaves_pair_out_of_effective_scope(
819        #[case] scope: Vec<Subscription>,
820        #[case] expected: Vec<Subscription>,
821    ) {
822        assert_eq!(removal_scope(&scope, &sub(DataType::Book, "A")), expected);
823    }
824
825    #[rstest]
826    #[case(DataType::Book, None)]
827    #[case(DataType::Trade, None)]
828    #[case(
829        DataType::Book5,
830        Some("Invalid CoinAPI WebSocket request: scope already streams book, so cannot add book5")
831    )]
832    #[case(
833        DataType::BookL3,
834        Some(
835            "Invalid CoinAPI WebSocket request: scope already streams book, so cannot add book_l3"
836        )
837    )]
838    fn scope_accepts_one_book_type(#[case] requested: DataType, #[case] expected: Option<&str>) {
839        let scope = [sub(DataType::Trade, "A"), sub(DataType::Book, "A")];
840
841        let result = check_book_type(&scope, &sub(requested, "B"));
842
843        assert_eq!(result.err().map(|e| e.to_string()).as_deref(), expected);
844    }
845
846    #[rstest]
847    fn rollback_undoes_repeated_scope_changes_in_reverse_order() {
848        let subscriptions = SubscriptionState::new(TOPIC_DELIMITER);
849        let a = sub(DataType::Trade, "A");
850        let b = sub(DataType::Trade, "B");
851        let c = sub(DataType::Trade, "C");
852        let d = sub(DataType::Trade, "D");
853
854        for subscription in [a, b] {
855            subscriptions.mark_subscribe(&topic(&subscription));
856            subscriptions.confirm_subscribe(&topic(&subscription));
857        }
858
859        let changes = [
860            ScopeChange::Unsubscribe(a),
861            ScopeChange::Subscribe(c),
862            ScopeChange::Unsubscribe(c),
863            ScopeChange::Subscribe(a),
864            ScopeChange::Unsubscribe(b),
865            ScopeChange::Subscribe(d),
866        ];
867
868        for change in &changes {
869            match change {
870                ScopeChange::Subscribe(subscription) => {
871                    subscriptions.mark_subscribe(&topic(subscription));
872                }
873                ScopeChange::Unsubscribe(subscription) => {
874                    subscriptions.mark_unsubscribe(&topic(subscription));
875                }
876            }
877        }
878
879        revert(&subscriptions, &changes);
880
881        assert_eq!(scope(&subscriptions), [a, b]);
882    }
883
884    #[cfg(not(all(feature = "simulation", madsim)))]
885    #[tokio::test]
886    async fn failed_send_on_existing_connection_rolls_back_scope() {
887        tokio::time::timeout(std::time::Duration::from_secs(10), async {
888            let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
889
890            let (client, _messages) = CoinApiWebSocketClient::new(
891                "test-key".into(),
892                CoinApiWebSocketConfig {
893                    url: format!("ws://{}", listener.local_addr().unwrap()),
894                    backend: TransportBackend::Tungstenite,
895                    ..CoinApiWebSocketConfig::default()
896                },
897            )
898            .unwrap();
899
900            let peer = tokio::spawn(async move {
901                let (socket, _) = listener.accept().await.unwrap();
902                let mut socket = tokio_tungstenite::accept_async(socket).await.unwrap();
903                let hello = socket.next().await.unwrap().unwrap();
904                while let Some(message) = socket.next().await {
905                    if message.unwrap().is_close() {
906                        socket.flush().await.unwrap();
907                        break;
908                    }
909                }
910
911                hello
912            });
913
914            let original = sub(DataType::Trade, "COINBASE_SPOT_BTC_USD");
915            let replacement = sub(DataType::Quote, "COINBASE_SPOT_BTC_USD");
916            assert!(
917                client
918                    .apply(&[ScopeChange::Subscribe(original)])
919                    .await
920                    .unwrap()
921                    .is_empty()
922            );
923            let transport = client
924                .connection
925                .lock()
926                .await
927                .as_ref()
928                .unwrap()
929                .client
930                .clone();
931            transport.disconnect().await;
932            let hello = peer.await.unwrap();
933
934            let error = client
935                .apply(&[
936                    ScopeChange::Unsubscribe(original),
937                    ScopeChange::Subscribe(replacement),
938                ])
939                .await
940                .unwrap_err();
941            let remaining = scope(
942                &client
943                    .connection
944                    .lock()
945                    .await
946                    .as_ref()
947                    .unwrap()
948                    .subscriptions,
949            );
950            client.close().await;
951
952            assert_eq!(hello.into_text().unwrap(), r#"{"type":"hello","heartbeat":true,"subscribe_data_type":["trade"],"subscribe_filter_symbol_id":["COINBASE_SPOT_BTC_USD$"]}"#);
953            assert!(matches!(error, Error::Send(SendError::Closed)));
954            assert_eq!(remaining, [original]);
955        })
956        .await
957        .unwrap();
958    }
959}
nautilus_coinapi/websocket/client.rs

nautilus_coinapi/websocket/
client.rs
