nautilus_binance/spot/websocket/public_json/
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//! Binance Spot public JSON WebSocket client for market data streams.
17
18use std::{
19    fmt::Debug,
20    sync::{
21        Arc, Mutex,
22        atomic::{AtomicBool, AtomicU8, AtomicU64, Ordering},
23    },
24};
25
26use futures_util::Stream;
27use nautilus_common::live::get_runtime;
28use nautilus_core::AtomicMap;
29use nautilus_model::instruments::{Instrument, InstrumentAny};
30use nautilus_network::{
31    mode::ConnectionMode,
32    websocket::{
33        PingHandler, SubscriptionState, TransportBackend, WebSocketClient, WebSocketConfig,
34        channel_message_handler,
35    },
36};
37use tokio_tungstenite::tungstenite::Message;
38use tokio_util::sync::CancellationToken;
39use ustr::Ustr;
40
41use super::{
42    handler::BinanceSpotPublicWsHandler,
43    messages::{BinanceSpotPublicWsCommand, BinanceSpotPublicWsMessage},
44};
45use crate::common::consts::{
46    BINANCE_RATE_LIMIT_KEY_SUBSCRIPTION, BINANCE_SPOT_WS_URL, BINANCE_WS_CONNECTION_QUOTA,
47    BINANCE_WS_SUBSCRIPTION_QUOTA,
48};
49
50/// Maximum streams per Spot JSON WebSocket connection.
51pub const MAX_STREAMS_PER_CONNECTION: usize = 1024;
52/// Maximum pooled Spot JSON WebSocket connections.
53pub const MAX_CONNECTIONS: usize = 20;
54
55struct ConnectionSlot {
56    cmd_tx: tokio::sync::mpsc::UnboundedSender<BinanceSpotPublicWsCommand>,
57    streams: Vec<String>,
58    handler_task: tokio::task::JoinHandle<()>,
59    bytes_task: tokio::task::JoinHandle<()>,
60    cancellation_token: CancellationToken,
61    connection_mode: Arc<AtomicU8>,
62}
63
64/// Binance Spot public JSON WebSocket client.
65#[derive(Clone)]
66pub struct BinanceSpotPublicJsonWebSocketClient {
67    url: String,
68    heartbeat: Option<u64>,
69    signal: Arc<AtomicBool>,
70    slots: Arc<Mutex<Vec<ConnectionSlot>>>,
71    out_tx: Arc<Mutex<Option<tokio::sync::mpsc::UnboundedSender<BinanceSpotPublicWsMessage>>>>,
72    out_rx: Arc<Mutex<Option<tokio::sync::mpsc::UnboundedReceiver<BinanceSpotPublicWsMessage>>>>,
73    request_id_counter: Arc<AtomicU64>,
74    instruments_cache: Arc<AtomicMap<Ustr, InstrumentAny>>,
75    transport_backend: TransportBackend,
76}
77
78impl Debug for BinanceSpotPublicJsonWebSocketClient {
79    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
80        f.debug_struct(stringify!(BinanceSpotPublicJsonWebSocketClient))
81            .field("url", &self.url)
82            .field("heartbeat", &self.heartbeat)
83            .finish_non_exhaustive()
84    }
85}
86
87impl Default for BinanceSpotPublicJsonWebSocketClient {
88    fn default() -> Self {
89        Self::new(None, None, TransportBackend::default())
90    }
91}
92
93impl BinanceSpotPublicJsonWebSocketClient {
94    /// Creates a new Spot public JSON WebSocket client.
95    #[must_use]
96    pub fn new(
97        url: Option<String>,
98        heartbeat: Option<u64>,
99        transport_backend: TransportBackend,
100    ) -> Self {
101        let url = normalize_spot_json_stream_url(
102            url.unwrap_or_else(|| BINANCE_SPOT_WS_URL.to_string())
103                .as_str(),
104        );
105
106        Self {
107            url,
108            heartbeat,
109            signal: Arc::new(AtomicBool::new(false)),
110            slots: Arc::new(Mutex::new(Vec::new())),
111            out_tx: Arc::new(Mutex::new(None)),
112            out_rx: Arc::new(Mutex::new(None)),
113            request_id_counter: Arc::new(AtomicU64::new(1)),
114            instruments_cache: Arc::new(AtomicMap::new()),
115            transport_backend,
116        }
117    }
118
119    /// Returns whether any connection in the pool is active.
120    #[must_use]
121    #[expect(clippy::missing_panics_doc, reason = "mutex poisoning is not expected")]
122    pub fn is_active(&self) -> bool {
123        let slots = self.slots.lock().expect("slots lock poisoned");
124        slots
125            .iter()
126            .any(|s| s.connection_mode.load(Ordering::Relaxed) == ConnectionMode::Active as u8)
127    }
128
129    /// Returns whether all connections in the pool are closed.
130    #[must_use]
131    #[expect(clippy::missing_panics_doc, reason = "mutex poisoning is not expected")]
132    pub fn is_closed(&self) -> bool {
133        let slots = self.slots.lock().expect("slots lock poisoned");
134        slots.is_empty()
135            || slots
136                .iter()
137                .all(|s| s.connection_mode.load(Ordering::Relaxed) == ConnectionMode::Closed as u8)
138    }
139
140    /// Connects the first WebSocket connection in the pool.
141    ///
142    /// # Errors
143    ///
144    /// Returns an error if connection fails.
145    #[expect(clippy::missing_panics_doc, reason = "mutex poisoning is not expected")]
146    pub async fn connect(&mut self) -> anyhow::Result<()> {
147        self.signal.store(false, Ordering::Relaxed);
148
149        let (out_tx, out_rx) = tokio::sync::mpsc::unbounded_channel();
150        *self.out_tx.lock().expect("out_tx lock poisoned") = Some(out_tx);
151        *self.out_rx.lock().expect("out_rx lock poisoned") = Some(out_rx);
152
153        let slot = self.create_connection().await?;
154        self.slots.lock().expect("slots lock poisoned").push(slot);
155
156        log::debug!(
157            "Connected to Binance Spot public JSON stream pool: url={}",
158            self.url
159        );
160        Ok(())
161    }
162
163    /// Closes all WebSocket connections and tasks.
164    ///
165    /// # Errors
166    ///
167    /// Returns an error if command delivery fails while shutting down.
168    #[expect(clippy::missing_panics_doc, reason = "mutex poisoning is not expected")]
169    pub async fn close(&mut self) -> anyhow::Result<()> {
170        self.signal.store(true, Ordering::Relaxed);
171
172        let taken: Vec<ConnectionSlot> = {
173            let mut guard = self.slots.lock().expect("slots lock poisoned");
174            guard.drain(..).collect()
175        };
176
177        for slot in taken {
178            let _ = slot.cmd_tx.send(BinanceSpotPublicWsCommand::Disconnect);
179            slot.cancellation_token.cancel();
180            let _ = slot.bytes_task.await;
181            let _ = slot.handler_task.await;
182        }
183
184        *self.out_tx.lock().expect("out_tx lock poisoned") = None;
185        *self.out_rx.lock().expect("out_rx lock poisoned") = None;
186
187        log::debug!("Disconnected from Binance Spot public JSON stream pool");
188        Ok(())
189    }
190
191    /// Subscribes to stream names.
192    ///
193    /// # Errors
194    ///
195    /// Returns an error if command delivery fails or if the connection pool is exhausted.
196    #[expect(clippy::missing_panics_doc, reason = "mutex poisoning is not expected")]
197    pub async fn subscribe(&self, streams: Vec<String>) -> anyhow::Result<()> {
198        // Phase 1: filter already-subscribed streams (brief lock)
199        let new_streams: Vec<String> = {
200            let slots = self.slots.lock().expect("slots lock poisoned");
201            streams
202                .into_iter()
203                .filter(|s| !slots.iter().any(|slot| slot.streams.contains(s)))
204                .collect()
205        };
206
207        if new_streams.is_empty() {
208            return Ok(());
209        }
210
211        // Phase 2: create connections if needed (no lock held during async connect)
212        loop {
213            let (remaining_capacity, slot_count) = {
214                let slots = self.slots.lock().expect("slots lock poisoned");
215                let cap: usize = slots
216                    .iter()
217                    .map(|s| MAX_STREAMS_PER_CONNECTION.saturating_sub(s.streams.len()))
218                    .sum();
219                (cap, slots.len())
220            };
221
222            if remaining_capacity >= new_streams.len() || slot_count >= MAX_CONNECTIONS {
223                break;
224            }
225
226            let new_slot = self.create_connection().await?;
227            let slot_count = {
228                let mut slots = self.slots.lock().expect("slots lock poisoned");
229                slots.push(new_slot);
230                slots.len()
231            };
232            log::debug!(
233                "Spot JSON pool slot {} connected: url={}",
234                slot_count - 1,
235                self.url
236            );
237        }
238
239        // Phase 3: stage assignments, send commands, then commit slot state.
240        let mut slots = self.slots.lock().expect("slots lock poisoned");
241        let mut slot_batches: Vec<(usize, Vec<String>)> = Vec::new();
242        let mut slot_counts: Vec<usize> = slots.iter().map(|s| s.streams.len()).collect();
243
244        for stream in &new_streams {
245            let slot_idx = slot_counts
246                .iter()
247                .position(|&count| count < MAX_STREAMS_PER_CONNECTION)
248                .ok_or_else(|| {
249                    anyhow::anyhow!(
250                        "Spot public JSON stream pool exhausted ({MAX_CONNECTIONS} connections x {MAX_STREAMS_PER_CONNECTION} streams)",
251                    )
252                })?;
253
254            slot_counts[slot_idx] += 1;
255
256            if let Some(batch) = slot_batches.iter_mut().find(|(i, _)| *i == slot_idx) {
257                batch.1.push(stream.clone());
258            } else {
259                slot_batches.push((slot_idx, vec![stream.clone()]));
260            }
261        }
262
263        for (slot_idx, batch) in &slot_batches {
264            slots[*slot_idx]
265                .cmd_tx
266                .send(BinanceSpotPublicWsCommand::Subscribe {
267                    streams: batch.clone(),
268                })
269                .map_err(|e| {
270                    anyhow::anyhow!("Handler not available for Spot JSON pool slot {slot_idx}: {e}")
271                })?;
272            slots[*slot_idx].streams.extend(batch.iter().cloned());
273        }
274
275        Ok(())
276    }
277
278    /// Unsubscribes from stream names.
279    ///
280    /// # Errors
281    ///
282    /// Returns an error if command delivery fails.
283    #[expect(clippy::missing_panics_doc, reason = "mutex poisoning is not expected")]
284    pub async fn unsubscribe(&self, streams: Vec<String>) -> anyhow::Result<()> {
285        if streams.is_empty() {
286            return Ok(());
287        }
288
289        let mut slots = self.slots.lock().expect("slots lock poisoned");
290        let mut slot_batches: Vec<(usize, Vec<String>)> = Vec::new();
291
292        for stream in &streams {
293            if let Some(slot_idx) = slots
294                .iter()
295                .position(|s| s.streams.iter().any(|x| x == stream))
296            {
297                if let Some(batch) = slot_batches.iter_mut().find(|(i, _)| *i == slot_idx) {
298                    batch.1.push(stream.clone());
299                } else {
300                    slot_batches.push((slot_idx, vec![stream.clone()]));
301                }
302            }
303        }
304
305        for (slot_idx, batch) in &slot_batches {
306            slots[*slot_idx]
307                .cmd_tx
308                .send(BinanceSpotPublicWsCommand::Unsubscribe {
309                    streams: batch.clone(),
310                })
311                .map_err(|e| {
312                    anyhow::anyhow!("Handler not available for Spot JSON pool slot {slot_idx}: {e}")
313                })?;
314
315            for stream in batch {
316                slots[*slot_idx].streams.retain(|s| s != stream);
317            }
318        }
319
320        Ok(())
321    }
322
323    /// Returns a stream of output messages.
324    #[expect(clippy::missing_panics_doc, reason = "mutex poisoning is not expected")]
325    pub fn stream(&self) -> impl Stream<Item = BinanceSpotPublicWsMessage> + 'static {
326        let mut guard = self.out_rx.lock().expect("out_rx lock poisoned");
327        let out_rx = guard.take();
328        drop(guard);
329
330        async_stream::stream! {
331            if let Some(mut rx) = out_rx {
332                while let Some(msg) = rx.recv().await {
333                    yield msg;
334                }
335            }
336        }
337    }
338
339    /// Bulk initialize the instrument cache.
340    pub fn cache_instruments(&self, instruments: &[InstrumentAny]) {
341        self.instruments_cache.rcu(|m| {
342            for inst in instruments {
343                m.insert(inst.raw_symbol().inner(), inst.clone());
344            }
345        });
346    }
347
348    /// Returns a shared reference to the instruments cache.
349    #[must_use]
350    pub fn instruments_cache(&self) -> Arc<AtomicMap<Ustr, InstrumentAny>> {
351        self.instruments_cache.clone()
352    }
353
354    async fn create_connection(&self) -> anyhow::Result<ConnectionSlot> {
355        let out_tx = self
356            .out_tx
357            .lock()
358            .expect("out_tx lock poisoned")
359            .clone()
360            .ok_or_else(|| anyhow::anyhow!("Output channel not initialized"))?;
361
362        let (raw_handler, raw_rx) = channel_message_handler();
363        let ping_handler: PingHandler = Arc::new(move |_| {});
364
365        let config = WebSocketConfig {
366            url: self.url.clone(),
367            headers: vec![],
368            heartbeat: self.heartbeat,
369            heartbeat_msg: None,
370            reconnect_timeout_ms: Some(5_000),
371            reconnect_delay_initial_ms: Some(500),
372            reconnect_delay_max_ms: Some(5_000),
373            reconnect_backoff_factor: Some(2.0),
374            reconnect_jitter_ms: Some(250),
375            reconnect_max_attempts: None,
376            idle_timeout_ms: None,
377            backend: self.transport_backend,
378            proxy_url: None,
379        };
380
381        let keyed_quotas = vec![(
382            BINANCE_RATE_LIMIT_KEY_SUBSCRIPTION[0].as_str().to_string(),
383            *BINANCE_WS_SUBSCRIPTION_QUOTA,
384        )];
385
386        let client = WebSocketClient::connect(
387            config,
388            Some(raw_handler),
389            Some(ping_handler),
390            None,
391            keyed_quotas,
392            Some(*BINANCE_WS_CONNECTION_QUOTA),
393        )
394        .await
395        .map_err(|e| anyhow::anyhow!("Failed to connect Spot public JSON WS: {e}"))?;
396
397        let connection_mode = client.connection_mode_atomic();
398        let subscriptions_state = SubscriptionState::new('@');
399        let cancellation_token = CancellationToken::new();
400
401        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
402
403        let (bytes_tx, bytes_rx) = tokio::sync::mpsc::unbounded_channel::<Vec<u8>>();
404
405        let bytes_task = get_runtime().spawn(async move {
406            let mut raw_rx = raw_rx;
407            while let Some(msg) = raw_rx.recv().await {
408                let data = match msg {
409                    Message::Binary(data) => data.to_vec(),
410                    Message::Text(text) => text.as_bytes().to_vec(),
411                    Message::Close(_) => break,
412                    Message::Ping(_) | Message::Pong(_) | Message::Frame(_) => continue,
413                };
414
415                if bytes_tx.send(data).is_err() {
416                    break;
417                }
418            }
419        });
420
421        let mut handler = BinanceSpotPublicWsHandler::new(
422            self.signal.clone(),
423            cmd_rx,
424            bytes_rx,
425            subscriptions_state.clone(),
426            self.request_id_counter.clone(),
427        );
428
429        cmd_tx
430            .send(BinanceSpotPublicWsCommand::SetClient(client))
431            .map_err(|e| anyhow::anyhow!("Failed to set Spot public JSON WS client: {e}"))?;
432
433        let signal = self.signal.clone();
434        let token = cancellation_token.clone();
435        let resubscribe_tx = cmd_tx.clone();
436
437        let handler_task = get_runtime().spawn(async move {
438            loop {
439                tokio::select! {
440                    () = token.cancelled() => {
441                        log::debug!("Spot public JSON handler task cancelled");
442                        break;
443                    }
444                    result = handler.next() => {
445                        match result {
446                            Some(BinanceSpotPublicWsMessage::Reconnected) => {
447                                log::info!("Spot public JSON WebSocket reconnected, restoring subscriptions");
448                                let topics = subscriptions_state.all_topics();
449                                for topic in &topics {
450                                    subscriptions_state.mark_failure(topic);
451                                }
452
453                                let streams = subscriptions_state.all_topics();
454                                if !streams.is_empty()
455                                    && let Err(e) = resubscribe_tx.send(BinanceSpotPublicWsCommand::Subscribe { streams }) {
456                                        log::error!("Failed to resubscribe after reconnect: {e}");
457                                    }
458
459                                if out_tx.send(BinanceSpotPublicWsMessage::Reconnected).is_err() {
460                                    log::debug!("Output channel closed");
461                                    break;
462                                }
463                            }
464                            Some(msg) => {
465                                if out_tx.send(msg).is_err() {
466                                    log::debug!("Output channel closed");
467                                    break;
468                                }
469                            }
470                            None => {
471                                if signal.load(Ordering::Relaxed) {
472                                    break;
473                                }
474                                tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
475                            }
476                        }
477                    }
478                }
479            }
480        });
481
482        Ok(ConnectionSlot {
483            cmd_tx,
484            streams: Vec::new(),
485            handler_task,
486            bytes_task,
487            cancellation_token,
488            connection_mode,
489        })
490    }
491}
492
493fn normalize_spot_json_stream_url(base_url: &str) -> String {
494    let trimmed = base_url.trim_end_matches('/');
495
496    if trimmed.ends_with("/stream") {
497        return trimmed.to_string();
498    }
499
500    if let Some(prefix) = trimmed.strip_suffix("/ws") {
501        return format!("{prefix}/stream");
502    }
503
504    format!("{trimmed}/stream")
505}
506
507#[cfg(test)]
508mod tests {
509    use std::sync::atomic::AtomicU8;
510
511    use nautilus_network::mode::ConnectionMode;
512    use rstest::rstest;
513
514    use super::*;
515
516    fn make_slot_with_streams(
517        streams: Vec<String>,
518    ) -> (
519        ConnectionSlot,
520        tokio::sync::mpsc::UnboundedReceiver<BinanceSpotPublicWsCommand>,
521    ) {
522        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
523
524        let handler_task = tokio::spawn(async {});
525
526        let bytes_task = tokio::spawn(async {});
527
528        let slot = ConnectionSlot {
529            cmd_tx,
530            streams,
531            handler_task,
532            bytes_task,
533            cancellation_token: CancellationToken::new(),
534            connection_mode: Arc::new(AtomicU8::new(ConnectionMode::Active as u8)),
535        };
536
537        (slot, cmd_rx)
538    }
539
540    #[tokio::test]
541    async fn test_subscribe_reuses_existing_stream_and_only_subscribes_new_one() {
542        let client =
543            BinanceSpotPublicJsonWebSocketClient::new(None, None, TransportBackend::default());
544        let (slot, mut cmd_rx) = make_slot_with_streams(vec!["btcusdt@trade".to_string()]);
545        client.slots.lock().expect("slots lock poisoned").push(slot);
546
547        client
548            .subscribe(vec![
549                "btcusdt@trade".to_string(),
550                "ethusdt@trade".to_string(),
551            ])
552            .await
553            .expect("subscribe should succeed");
554
555        match cmd_rx
556            .try_recv()
557            .expect("one subscribe command should be sent")
558        {
559            BinanceSpotPublicWsCommand::Subscribe { streams } => {
560                assert_eq!(streams, vec!["ethusdt@trade".to_string()]);
561            }
562            _ => panic!("unexpected command type"),
563        }
564        assert!(matches!(
565            cmd_rx.try_recv(),
566            Err(tokio::sync::mpsc::error::TryRecvError::Empty)
567        ));
568
569        let slots = client.slots.lock().expect("slots lock poisoned");
570        assert_eq!(slots.len(), 1);
571        assert_eq!(
572            slots[0].streams,
573            vec!["btcusdt@trade".to_string(), "ethusdt@trade".to_string()]
574        );
575    }
576
577    #[tokio::test]
578    async fn test_unsubscribe_removes_only_target_stream_when_sibling_still_subscribed() {
579        let client =
580            BinanceSpotPublicJsonWebSocketClient::new(None, None, TransportBackend::default());
581        let (slot, mut cmd_rx) = make_slot_with_streams(vec![
582            "btcusdt@trade".to_string(),
583            "btcusdt@bookTicker".to_string(),
584        ]);
585        client.slots.lock().expect("slots lock poisoned").push(slot);
586
587        client
588            .unsubscribe(vec!["btcusdt@bookTicker".to_string()])
589            .await
590            .expect("unsubscribe should succeed");
591
592        match cmd_rx
593            .try_recv()
594            .expect("one unsubscribe command should be sent")
595        {
596            BinanceSpotPublicWsCommand::Unsubscribe { streams } => {
597                assert_eq!(streams, vec!["btcusdt@bookTicker".to_string()]);
598            }
599            _ => panic!("unexpected command type"),
600        }
601        assert!(matches!(
602            cmd_rx.try_recv(),
603            Err(tokio::sync::mpsc::error::TryRecvError::Empty)
604        ));
605
606        let slots = client.slots.lock().expect("slots lock poisoned");
607        assert_eq!(slots.len(), 1);
608        assert_eq!(slots[0].streams, vec!["btcusdt@trade".to_string()]);
609    }
610
611    #[tokio::test]
612    async fn test_unsubscribe_all_streams_clears_slot_state() {
613        let client =
614            BinanceSpotPublicJsonWebSocketClient::new(None, None, TransportBackend::default());
615        let (slot, mut cmd_rx) = make_slot_with_streams(vec![
616            "btcusdt@trade".to_string(),
617            "ethusdt@trade".to_string(),
618        ]);
619        client.slots.lock().expect("slots lock poisoned").push(slot);
620
621        client
622            .unsubscribe(vec![
623                "btcusdt@trade".to_string(),
624                "ethusdt@trade".to_string(),
625            ])
626            .await
627            .expect("unsubscribe should succeed");
628
629        let mut sent = match cmd_rx
630            .try_recv()
631            .expect("one unsubscribe command should be sent")
632        {
633            BinanceSpotPublicWsCommand::Unsubscribe { streams } => streams,
634            _ => panic!("unexpected command type"),
635        };
636
637        sent.sort();
638        assert_eq!(
639            sent,
640            vec!["btcusdt@trade".to_string(), "ethusdt@trade".to_string()]
641        );
642
643        assert!(matches!(
644            cmd_rx.try_recv(),
645            Err(tokio::sync::mpsc::error::TryRecvError::Empty)
646        ));
647
648        let slots = client.slots.lock().expect("slots lock poisoned");
649        assert_eq!(slots.len(), 1);
650        assert!(slots[0].streams.is_empty());
651    }
652
653    #[tokio::test]
654    async fn test_subscribe_batches_same_slot_streams_in_single_command() {
655        let client =
656            BinanceSpotPublicJsonWebSocketClient::new(None, None, TransportBackend::default());
657        let (slot, mut cmd_rx) = make_slot_with_streams(vec![]);
658        client.slots.lock().expect("slots lock poisoned").push(slot);
659
660        client
661            .subscribe(vec![
662                "btcusdt@trade".to_string(),
663                "ethusdt@trade".to_string(),
664            ])
665            .await
666            .expect("subscribe should succeed");
667
668        let mut sent = match cmd_rx
669            .try_recv()
670            .expect("one subscribe command should be sent")
671        {
672            BinanceSpotPublicWsCommand::Subscribe { streams } => streams,
673            _ => panic!("unexpected command type"),
674        };
675
676        sent.sort();
677        assert_eq!(
678            sent,
679            vec!["btcusdt@trade".to_string(), "ethusdt@trade".to_string()]
680        );
681        assert!(matches!(
682            cmd_rx.try_recv(),
683            Err(tokio::sync::mpsc::error::TryRecvError::Empty)
684        ));
685
686        let slots = client.slots.lock().expect("slots lock poisoned");
687        let mut stored = slots[0].streams.clone();
688        stored.sort();
689        assert_eq!(
690            stored,
691            vec!["btcusdt@trade".to_string(), "ethusdt@trade".to_string()]
692        );
693    }
694
695    #[tokio::test]
696    async fn test_unsubscribe_batches_same_slot_streams_in_single_command() {
697        let client =
698            BinanceSpotPublicJsonWebSocketClient::new(None, None, TransportBackend::default());
699        let (slot, mut cmd_rx) = make_slot_with_streams(vec![
700            "btcusdt@trade".to_string(),
701            "ethusdt@trade".to_string(),
702        ]);
703        client.slots.lock().expect("slots lock poisoned").push(slot);
704
705        client
706            .unsubscribe(vec![
707                "btcusdt@trade".to_string(),
708                "ethusdt@trade".to_string(),
709            ])
710            .await
711            .expect("unsubscribe should succeed");
712
713        let mut sent = match cmd_rx
714            .try_recv()
715            .expect("one unsubscribe command should be sent")
716        {
717            BinanceSpotPublicWsCommand::Unsubscribe { streams } => streams,
718            _ => panic!("unexpected command type"),
719        };
720
721        sent.sort();
722        assert_eq!(
723            sent,
724            vec!["btcusdt@trade".to_string(), "ethusdt@trade".to_string()]
725        );
726        assert!(matches!(
727            cmd_rx.try_recv(),
728            Err(tokio::sync::mpsc::error::TryRecvError::Empty)
729        ));
730
731        let slots = client.slots.lock().expect("slots lock poisoned");
732        assert_eq!(slots.len(), 1);
733        assert!(slots[0].streams.is_empty());
734    }
735
736    #[rstest]
737    #[case("wss://stream.binance.com/ws", "wss://stream.binance.com/stream")]
738    #[case("wss://stream.binance.com/stream", "wss://stream.binance.com/stream")]
739    #[case("wss://stream.binance.com/stream/", "wss://stream.binance.com/stream")]
740    fn test_normalize_spot_json_stream_url(#[case] input: &str, #[case] expected: &str) {
741        assert_eq!(normalize_spot_json_stream_url(input), expected);
742    }
743}
nautilus_binance/spot/websocket/public_json/client.rs

nautilus_binance/spot/websocket/public_json/
client.rs
