nautilus_binance/spot/websocket/streams/
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 WebSocket client for SBE market data streams.
17//!
18//! ## Connection Details
19//!
20//! - Endpoint: `stream-sbe.binance.com` or `stream-sbe.binance.com:9443`
21//! - Authentication: Ed25519 API key in `X-MBX-APIKEY` header
22//! - Max streams: 1024 per connection
23//! - Max connections: 20 per pool (up to 20,480 total streams)
24//! - Connection validity: 24 hours
25//! - Ping/pong: Every 20 seconds
26
27use std::{
28    fmt::Debug,
29    sync::{
30        Arc,
31        atomic::{AtomicBool, AtomicU8, AtomicU64, Ordering},
32    },
33};
34
35use futures_util::Stream;
36use nautilus_core::{
37    AtomicMap,
38    string::secret::{REDACTED, SecretString},
39};
40use nautilus_live::{
41    SocketControl, SocketControlFactory,
42    task::{TaskJoinOutcome, TaskSlot, finish_task},
43};
44use nautilus_model::instruments::{Instrument, InstrumentAny};
45use nautilus_network::{
46    ConnectionUrl,
47    http::create_standard_nautilus_headers,
48    mode::ConnectionMode,
49    websocket::{
50        PingHandler, SubscriptionState, TransportBackend, WebSocketClient, WebSocketConfig,
51        channel_message_handler,
52    },
53};
54use parking_lot::Mutex;
55use tokio_util::sync::CancellationToken;
56use ustr::Ustr;
57
58use super::{
59    super::error::{BinanceWsError, BinanceWsResult},
60    handler::BinanceSpotWsFeedHandler,
61    messages::{BinanceSpotWsMessage, BinanceSpotWsStreamsCommand},
62    subscription::{MAX_CONNECTIONS, MAX_STREAMS_PER_CONNECTION},
63};
64use crate::common::{
65    consts::{
66        BINANCE_API_KEY_HEADER, BINANCE_RATE_LIMIT_KEY_SUBSCRIPTION, BINANCE_SPOT_SBE_WS_URL,
67        BINANCE_WS_CONNECTION_QUOTA, BINANCE_WS_SUBSCRIPTION_QUOTA,
68    },
69    credential::Ed25519Credential,
70};
71
72// State for a single WebSocket connection within the pool
73struct ConnectionSlot {
74    cmd_tx: tokio::sync::mpsc::UnboundedSender<BinanceSpotWsStreamsCommand>,
75    streams: Vec<String>,
76    subscriptions_state: SubscriptionState,
77    task_handle: TaskSlot<()>,
78    cancellation_token: CancellationToken,
79    connection_mode: Arc<AtomicU8>,
80    socket_control: Option<SocketControl>,
81    shutdown_errors: Vec<String>,
82}
83
84/// Binance Spot WebSocket client for SBE market data streams.
85///
86/// Manages a pool of up to 20 connections, each supporting up to 1024 streams.
87/// New connections are created automatically when subscribing exceeds the current
88/// connection's stream limit. All connections feed into a single output stream,
89/// transparent to the data client.
90#[derive(Clone)]
91pub struct BinanceSpotWebSocketClient {
92    url: ConnectionUrl,
93    credential: Option<Arc<Ed25519Credential>>,
94    heartbeat: Option<u64>,
95    signal: Arc<AtomicBool>,
96    slots: Arc<ConnectionSlots>,
97    connect_lock: Arc<tokio::sync::Mutex<()>>,
98    out_tx: Arc<Mutex<Option<tokio::sync::mpsc::UnboundedSender<BinanceSpotWsMessage>>>>,
99    out_rx: Arc<Mutex<Option<tokio::sync::mpsc::UnboundedReceiver<BinanceSpotWsMessage>>>>,
100    request_id_counter: Arc<AtomicU64>,
101    instruments_cache: Arc<AtomicMap<Ustr, InstrumentAny>>,
102    transport_backend: TransportBackend,
103    proxy_url: Option<SecretString>,
104    socket_factory: Option<SocketControlFactory>,
105    socket_endpoint: Option<String>,
106}
107
108impl Debug for BinanceSpotWebSocketClient {
109    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
110        f.debug_struct(stringify!(BinanceSpotWebSocketClient))
111            .field("url", &self.url)
112            .field("credential", &self.credential.as_ref().map(|_| REDACTED))
113            .field("heartbeat", &self.heartbeat)
114            .finish_non_exhaustive()
115    }
116}
117
118impl Default for BinanceSpotWebSocketClient {
119    fn default() -> Self {
120        Self::new(None, None, None, None, TransportBackend::default()).unwrap()
121    }
122}
123
124impl BinanceSpotWebSocketClient {
125    /// Creates a new [`BinanceSpotWebSocketClient`] instance.
126    ///
127    /// # Errors
128    ///
129    /// Returns an error if credential creation fails.
130    pub fn new(
131        url: Option<String>,
132        api_key: Option<String>,
133        api_secret: Option<String>,
134        heartbeat: Option<u64>,
135        transport_backend: TransportBackend,
136    ) -> anyhow::Result<Self> {
137        let url = url.unwrap_or(BINANCE_SPOT_SBE_WS_URL.to_string());
138
139        let credential = match (
140            api_key.map(SecretString::from),
141            api_secret.map(SecretString::from),
142        ) {
143            (Some(key), Some(secret)) => {
144                let credential = Ed25519Credential::new(key, secret).map_err(|e| {
145                    anyhow::anyhow!(
146                        "Binance Spot SBE market-data streams require an Ed25519 API key \
147                         (HMAC keys are not supported): {e}"
148                    )
149                })?;
150                Some(Arc::new(credential))
151            }
152            _ => None,
153        };
154
155        Ok(Self {
156            url: url.into(),
157            credential,
158            heartbeat,
159            signal: Arc::new(AtomicBool::new(false)),
160            slots: Arc::new(ConnectionSlots(Mutex::new(Vec::new()))),
161            connect_lock: Arc::new(tokio::sync::Mutex::new(())),
162            out_tx: Arc::new(Mutex::new(None)),
163            out_rx: Arc::new(Mutex::new(None)),
164            request_id_counter: Arc::new(AtomicU64::new(1)),
165            instruments_cache: Arc::new(AtomicMap::new()),
166            transport_backend,
167            proxy_url: None,
168            socket_factory: None,
169            socket_endpoint: None,
170        })
171    }
172
173    /// Configures the proxy used by every connection in the stream pool.
174    #[must_use]
175    pub fn with_proxy(mut self, proxy_url: Option<String>) -> Self {
176        self.proxy_url = proxy_url.map(SecretString::from);
177        self
178    }
179
180    /// Configures socket state reporting and reconnect control for the stream pool.
181    #[must_use]
182    pub fn with_socket_control(
183        mut self,
184        factory: SocketControlFactory,
185        endpoint: impl Into<String>,
186    ) -> Self {
187        self.socket_factory = Some(factory);
188        self.socket_endpoint = Some(endpoint.into());
189        self
190    }
191
192    /// Returns whether API credentials are configured.
193    #[must_use]
194    pub fn has_credentials(&self) -> bool {
195        self.credential.is_some()
196    }
197
198    /// Returns whether any connection in the pool is active.
199    #[must_use]
200    pub fn is_active(&self) -> bool {
201        let slots = self.slots.lock();
202        slots
203            .iter()
204            .any(|s| s.connection_mode.load(Ordering::Relaxed) == ConnectionMode::Active as u8)
205    }
206
207    /// Returns whether all connections in the pool are closed.
208    #[must_use]
209    pub fn is_closed(&self) -> bool {
210        let slots = self.slots.lock();
211        slots.is_empty()
212            || slots
213                .iter()
214                .all(|s| s.connection_mode.load(Ordering::Relaxed) == ConnectionMode::Closed as u8)
215    }
216
217    /// Returns the total number of confirmed subscriptions across all connections.
218    #[must_use]
219    pub fn subscription_count(&self) -> usize {
220        let slots = self.slots.lock();
221        slots.iter().map(|s| s.subscriptions_state.len()).sum()
222    }
223
224    /// Connects the first WebSocket connection in the pool.
225    ///
226    /// # Errors
227    ///
228    /// Returns an error if connection fails.
229    pub async fn connect(&mut self) -> BinanceWsResult<()> {
230        let connect_lock = Arc::clone(&self.connect_lock);
231        let _connect_guard = connect_lock.lock().await;
232
233        if !self.slots.lock().is_empty() {
234            self.close_connections().await?;
235        }
236
237        {
238            let _slots = self.slots.lock();
239            self.signal.store(false, Ordering::Release);
240        }
241
242        let (out_tx, out_rx) = tokio::sync::mpsc::unbounded_channel();
243        *self.out_tx.lock() = Some(out_tx);
244        *self.out_rx.lock() = Some(out_rx);
245
246        let slot = self.create_connection(0).await?;
247        let shutdown = {
248            let mut slots = self.slots.lock();
249            let shutdown = self.signal.load(Ordering::Acquire);
250            slots.push(slot);
251            shutdown
252        };
253
254        if shutdown {
255            let rollback = self.close_connections().await;
256            return Err(BinanceWsError::ClientError(match rollback {
257                Ok(()) => "Binance Spot SBE stream pool shutdown began during connect".to_string(),
258                Err(e) => format!(
259                    "Binance Spot SBE stream pool shutdown began during connect; rollback failed: {e}"
260                ),
261            }));
262        }
263
264        log::debug!(
265            "Connected to Binance Spot SBE stream pool: url={}",
266            self.url.origin()
267        );
268        Ok(())
269    }
270
271    /// Closes all WebSocket connections in the pool.
272    ///
273    /// # Errors
274    ///
275    /// Returns an error if disconnect fails.
276    pub async fn close(&mut self) -> BinanceWsResult<()> {
277        self.begin_shutdown();
278        let connect_lock = Arc::clone(&self.connect_lock);
279        let _connect_guard = connect_lock.lock().await;
280        self.close_connections().await
281    }
282
283    pub(crate) fn begin_shutdown(&self) {
284        let slots = self.slots.lock();
285        self.signal.store(true, Ordering::Release);
286
287        for slot in slots.iter() {
288            if let Some(control) = &slot.socket_control {
289                control.deregister();
290            }
291            slot.cancellation_token.cancel();
292            let _ = slot.cmd_tx.send(BinanceSpotWsStreamsCommand::Disconnect);
293        }
294    }
295
296    async fn close_connections(&self) -> BinanceWsResult<()> {
297        self.begin_shutdown();
298
299        let mut batch = ConnectionSlotBatch::take(&self.slots);
300        let mut index = batch.slots.len();
301        while index > 0 {
302            index -= 1;
303            let slot = &mut batch.slots[index];
304            if let Some(control) = &slot.socket_control {
305                control.deregister();
306            }
307            slot.cancellation_token.cancel();
308            let _result = slot.cmd_tx.send(BinanceSpotWsStreamsCommand::Disconnect);
309            let Some(outcome) = finish_task(
310                &mut slot.task_handle,
311                std::time::Duration::from_secs(2),
312                std::time::Duration::from_secs(2),
313            )
314            .await
315            else {
316                if slot.shutdown_errors.is_empty() {
317                    batch.slots.remove(index);
318                }
319                continue;
320            };
321
322            match outcome {
323                TaskJoinOutcome::Completed(()) | TaskJoinOutcome::Aborted => {}
324                TaskJoinOutcome::Failed(e) => slot
325                    .shutdown_errors
326                    .push(format!("Spot SBE handler failed: {e}")),
327                TaskJoinOutcome::Incomplete => {
328                    slot.shutdown_errors
329                        .push("Spot SBE handler did not stop after abort".to_string());
330                }
331            }
332
333            if slot.task_handle.is_none() && slot.shutdown_errors.is_empty() {
334                batch.slots.remove(index);
335            }
336        }
337
338        *self.out_tx.lock() = None;
339        *self.out_rx.lock() = None;
340
341        let errors = batch
342            .slots
343            .iter_mut()
344            .flat_map(|slot| std::mem::take(&mut slot.shutdown_errors))
345            .collect::<Vec<_>>();
346        batch.slots.retain(|slot| slot.task_handle.is_some());
347        if !errors.is_empty() {
348            return Err(BinanceWsError::ClientError(errors.join("; ")));
349        }
350        log::debug!("Disconnected from Binance Spot SBE stream pool");
351        Ok(())
352    }
353
354    /// Subscribes to the specified streams.
355    ///
356    /// Streams are distributed across pool connections. New connections are created
357    /// automatically when existing ones reach the 1024-stream limit, up to a maximum
358    /// of 20 connections.
359    ///
360    /// # Errors
361    ///
362    /// Returns an error if the pool is exhausted or command delivery fails.
363    pub async fn subscribe(&self, streams: Vec<String>) -> BinanceWsResult<()> {
364        // Serialize all phases so concurrent subscribers see a consistent
365        // pool state and can't trigger spurious `Pool exhausted`.
366        let _connect_guard = self.connect_lock.lock().await;
367
368        // Phase 1: filter already-subscribed streams (brief lock).
369        let new_streams: Vec<String> = {
370            let slots = self.slots.lock();
371
372            if self.signal.load(Ordering::Acquire) {
373                return Err(BinanceWsError::ClientError(
374                    "Binance Spot SBE stream pool is shutting down".to_string(),
375                ));
376            }
377            streams
378                .into_iter()
379                .filter(|s| !slots.iter().any(|slot| slot.streams.contains(s)))
380                .collect()
381        };
382
383        if new_streams.is_empty() {
384            return Ok(());
385        }
386
387        // Phase 2: create connections if needed.
388        loop {
389            let (remaining_capacity, slot_count) = {
390                let slots = self.slots.lock();
391                let cap: usize = slots
392                    .iter()
393                    .map(|s| MAX_STREAMS_PER_CONNECTION - s.streams.len())
394                    .sum();
395                (cap, slots.len())
396            };
397
398            if remaining_capacity >= new_streams.len() || slot_count >= MAX_CONNECTIONS {
399                break;
400            }
401
402            let new_slot = self.create_connection(slot_count).await?;
403            let (slot_count, shutdown) = {
404                let mut slots = self.slots.lock();
405                let shutdown = self.signal.load(Ordering::Acquire);
406                slots.push(new_slot);
407                (slots.len(), shutdown)
408            };
409
410            if shutdown {
411                let client = self.clone();
412                let rollback = client.close_connections().await;
413                return Err(BinanceWsError::ClientError(match rollback {
414                    Ok(()) => {
415                        "Binance Spot SBE stream pool shutdown began during subscribe".to_string()
416                    }
417                    Err(e) => format!(
418                        "Binance Spot SBE stream pool shutdown began during subscribe; rollback failed: {e}"
419                    ),
420                }));
421            }
422
423            log::debug!(
424                "Pool slot {} connected: url={}",
425                slot_count - 1,
426                self.url.origin()
427            );
428        }
429
430        // Phase 3: assign streams to slots and send commands (brief lock).
431        // Stage assignments first so a capacity error leaves slots unchanged.
432        let mut slots = self.slots.lock();
433
434        if self.signal.load(Ordering::Acquire) {
435            return Err(BinanceWsError::ClientError(
436                "Binance Spot SBE stream pool is shutting down".to_string(),
437            ));
438        }
439        let mut slot_batches: Vec<(usize, Vec<String>)> = Vec::new();
440        let mut slot_counts: Vec<usize> = slots.iter().map(|s| s.streams.len()).collect();
441
442        for stream in &new_streams {
443            let slot_idx = slot_counts
444                .iter()
445                .position(|&count| count < MAX_STREAMS_PER_CONNECTION)
446                .ok_or_else(|| {
447                    let max_total = MAX_CONNECTIONS * MAX_STREAMS_PER_CONNECTION;
448                    BinanceWsError::ClientError(format!(
449                        "Pool exhausted: {max_total} total subscriptions \
450                         ({MAX_CONNECTIONS} connections x {MAX_STREAMS_PER_CONNECTION} streams)"
451                    ))
452                })?;
453
454            slot_counts[slot_idx] += 1;
455
456            if let Some(batch) = slot_batches.iter_mut().find(|(i, _)| *i == slot_idx) {
457                batch.1.push(stream.clone());
458            } else {
459                slot_batches.push((slot_idx, vec![stream.clone()]));
460            }
461        }
462
463        // Send commands first; only update slot state on success
464        for (slot_idx, batch) in &slot_batches {
465            slots[*slot_idx]
466                .cmd_tx
467                .send(BinanceSpotWsStreamsCommand::Subscribe {
468                    streams: batch.clone(),
469                })
470                .map_err(|e| {
471                    BinanceWsError::ClientError(format!(
472                        "Handler not available for pool slot {slot_idx}: {e}"
473                    ))
474                })?;
475            slots[*slot_idx].streams.extend(batch.iter().cloned());
476        }
477
478        Ok(())
479    }
480
481    /// Unsubscribes from the specified streams.
482    ///
483    /// # Errors
484    ///
485    /// Returns an error if command delivery fails.
486    pub async fn unsubscribe(&self, streams: Vec<String>) -> BinanceWsResult<()> {
487        let _connect_guard = self.connect_lock.lock().await;
488        let mut slots = self.slots.lock();
489
490        if self.signal.load(Ordering::Acquire) {
491            return Err(BinanceWsError::ClientError(
492                "Binance Spot SBE stream pool is shutting down".to_string(),
493            ));
494        }
495        let mut slot_batches: Vec<(usize, Vec<String>)> = Vec::new();
496
497        for stream in &streams {
498            if let Some(slot_idx) = slots.iter().position(|s| s.streams.contains(stream)) {
499                if let Some(batch) = slot_batches.iter_mut().find(|(i, _)| *i == slot_idx) {
500                    batch.1.push(stream.clone());
501                } else {
502                    slot_batches.push((slot_idx, vec![stream.clone()]));
503                }
504            }
505        }
506
507        // Send commands first; only update slot state on success
508        for (slot_idx, batch) in &slot_batches {
509            slots[*slot_idx]
510                .cmd_tx
511                .send(BinanceSpotWsStreamsCommand::Unsubscribe {
512                    streams: batch.clone(),
513                })
514                .map_err(|e| {
515                    BinanceWsError::ClientError(format!(
516                        "Handler not available for pool slot {slot_idx}: {e}"
517                    ))
518                })?;
519
520            for stream in batch {
521                slots[*slot_idx].streams.retain(|s| s != stream);
522            }
523        }
524
525        Ok(())
526    }
527
528    /// Returns a stream of messages from all WebSocket connections.
529    ///
530    /// This method can only be called once per connection lifecycle. Subsequent calls
531    /// return an empty stream.
532    pub fn stream(&self) -> impl Stream<Item = BinanceSpotWsMessage> + 'static {
533        let out_rx = self.out_rx.lock().take();
534        async_stream::stream! {
535            if let Some(mut rx) = out_rx {
536                while let Some(msg) = rx.recv().await {
537                    yield msg;
538                }
539            }
540        }
541    }
542
543    /// Bulk initialize the instrument cache.
544    pub fn cache_instruments(&self, instruments: &[InstrumentAny]) {
545        self.instruments_cache.rcu(|m| {
546            for inst in instruments {
547                m.insert(inst.symbol().inner(), inst.clone());
548            }
549        });
550    }
551
552    /// Replaces the complete instrument cache.
553    pub fn replace_instruments(&self, instruments: &[InstrumentAny]) {
554        let cache = instruments
555            .iter()
556            .map(|instrument| (instrument.symbol().inner(), instrument.clone()))
557            .collect();
558        self.instruments_cache.store(cache);
559    }
560
561    /// Update a single instrument in the cache.
562    pub fn cache_instrument(&self, instrument: InstrumentAny) {
563        self.instruments_cache
564            .insert(instrument.symbol().inner(), instrument);
565    }
566
567    /// Returns a shared reference to the instruments cache.
568    #[must_use]
569    pub fn instruments_cache(&self) -> Arc<AtomicMap<Ustr, InstrumentAny>> {
570        self.instruments_cache.clone()
571    }
572
573    /// Returns an instrument from the cache by symbol.
574    #[must_use]
575    pub fn get_instrument(&self, symbol: &str) -> Option<InstrumentAny> {
576        self.instruments_cache.get_cloned(&Ustr::from(symbol))
577    }
578
579    async fn create_connection(&self, slot_index: usize) -> BinanceWsResult<ConnectionSlot> {
580        let out_tx = self.out_tx.lock().clone().ok_or_else(|| {
581            BinanceWsError::ClientError("Output channel not initialized".to_string())
582        })?;
583
584        let (raw_handler, raw_rx) = channel_message_handler();
585        let ping_handler: PingHandler = Arc::new(move |_| {});
586
587        let mut headers = create_standard_nautilus_headers();
588
589        if let Some(ref cred) = self.credential {
590            headers.push((
591                BINANCE_API_KEY_HEADER.to_string(),
592                cred.api_key().to_string(),
593            ));
594        }
595
596        let config = WebSocketConfig {
597            url: self.url.expose().to_owned(),
598            headers,
599            heartbeat_interval_secs: self.heartbeat,
600            heartbeat_payload: None,
601            connect_timeout_ms: Some(5_000),
602            reconnect_delay_initial_ms: Some(500),
603            reconnect_delay_max_ms: Some(5_000),
604            reconnect_backoff_factor: Some(2.0),
605            reconnect_jitter_ms: Some(250),
606            reconnect_max_attempts: None,
607            heartbeat_timeout_secs: None,
608            idle_timeout_ms: None,
609            writer_capacity: None,
610            backend: self.transport_backend,
611            proxy_url: self
612                .proxy_url
613                .as_ref()
614                .map(|value| value.expose_secret().to_owned()),
615            max_message_size_bytes: None,
616            max_frame_size_bytes: None,
617        };
618
619        let keyed_quotas = vec![(
620            BINANCE_RATE_LIMIT_KEY_SUBSCRIPTION[0].to_string(),
621            *BINANCE_WS_SUBSCRIPTION_QUOTA,
622        )];
623
624        let socket_control = self
625            .socket_factory
626            .as_ref()
627            .zip(self.socket_endpoint.as_ref())
628            .map(|(factory, endpoint)| {
629                let endpoint = if slot_index == 0 {
630                    endpoint.clone()
631                } else {
632                    format!("{endpoint}-{slot_index}")
633                };
634                factory.control(endpoint)
635            });
636        let client = WebSocketClient::builder()
637            .config(config)
638            .message_handler(raw_handler)
639            .ping_handler(ping_handler)
640            .keyed_quotas(keyed_quotas)
641            .default_quota(*BINANCE_WS_CONNECTION_QUOTA)
642            .maybe_state_sink(socket_control.as_ref().map(SocketControl::sink))
643            .connect()
644            .await
645            .map_err(|e| BinanceWsError::NetworkError(e.to_string()))?;
646
647        let connection_mode = client.connection_mode_atomic();
648        let reconnect_handle = client.reconnect_handle();
649        let subscriptions_state = SubscriptionState::new('@');
650        let cancellation_token = CancellationToken::new();
651
652        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
653
654        let mut handler = BinanceSpotWsFeedHandler::new(
655            self.signal.clone(),
656            cmd_rx,
657            raw_rx,
658            out_tx.clone(),
659            subscriptions_state.clone(),
660            self.request_id_counter.clone(),
661        );
662
663        cmd_tx
664            .send(BinanceSpotWsStreamsCommand::SetClient(client))
665            .map_err(|e| BinanceWsError::ClientError(format!("Failed to set client: {e}")))?;
666
667        let signal = self.signal.clone();
668        let token = cancellation_token.clone();
669        let subs = subscriptions_state.clone();
670        let resubscribe_tx = cmd_tx.clone();
671
672        let mut task_handle = TaskSlot::new();
673        if let Err(e) = task_handle.spawn(async move {
674            loop {
675                tokio::select! {
676                    () = token.cancelled() => {
677                        log::debug!("Handler task cancelled");
678                        break;
679                    }
680                    result = handler.next() => {
681                        match result {
682                            Some(BinanceSpotWsMessage::Reconnected) => {
683                                log::info!("WebSocket reconnected, restoring subscriptions");
684                                let all_topics = subs.all_topics();
685                                for topic in &all_topics {
686                                    subs.mark_failure(topic);
687                                }
688
689                                let streams = subs.all_topics();
690                                if !streams.is_empty()
691                                    && let Err(e) = resubscribe_tx.send(BinanceSpotWsStreamsCommand::Subscribe { streams }) {
692                                        log::error!("Failed to resubscribe after reconnect: {e}");
693                                    }
694
695                                if out_tx.send(BinanceSpotWsMessage::Reconnected).is_err() {
696                                    log::debug!("Output channel closed");
697                                    break;
698                                }
699                            }
700                            Some(msg) => {
701                                if out_tx.send(msg).is_err() {
702                                    log::debug!("Output channel closed");
703                                    break;
704                                }
705                            }
706                            None => {
707                                if signal.load(Ordering::Relaxed) {
708                                    log::debug!("Handler received shutdown signal");
709                                } else {
710                                    log::warn!("Handler loop ended unexpectedly");
711                                }
712                                break;
713                            }
714                        }
715                    }
716                }
717            }
718        }) {
719            cancellation_token.cancel();
720            let shutdown_error = match finish_task(
721                &mut task_handle,
722                std::time::Duration::ZERO,
723                std::time::Duration::from_secs(2),
724            )
725            .await
726            {
727                Some(TaskJoinOutcome::Failed(error)) => {
728                    Some(format!("Binance Spot WS handler task failed: {error}"))
729                }
730                Some(TaskJoinOutcome::Incomplete) => {
731                    Some("Binance Spot WS handler task did not stop after abort".to_string())
732                }
733                None | Some(TaskJoinOutcome::Completed(()) | TaskJoinOutcome::Aborted) => None,
734            };
735            return Err(BinanceWsError::ClientError(match shutdown_error {
736                Some(shutdown_error) => format!(
737                    "Failed to start Spot WS handler task: {e}; startup rollback failed: \
738                     {shutdown_error}"
739                ),
740                None => format!("Failed to start Spot WS handler task: {e}"),
741            }));
742        }
743
744        if let Some(control) = &socket_control {
745            control.register(move || reconnect_handle.request_reconnect());
746        }
747
748        Ok(ConnectionSlot {
749            cmd_tx,
750            streams: Vec::new(),
751            subscriptions_state,
752            task_handle,
753            cancellation_token,
754            connection_mode,
755            socket_control,
756            shutdown_errors: Vec::new(),
757        })
758    }
759}
760
761struct ConnectionSlots(Mutex<Vec<ConnectionSlot>>);
762
763impl std::ops::Deref for ConnectionSlots {
764    type Target = Mutex<Vec<ConnectionSlot>>;
765
766    fn deref(&self) -> &Self::Target {
767        &self.0
768    }
769}
770
771impl Drop for ConnectionSlots {
772    fn drop(&mut self) {
773        for slot in self.0.get_mut().iter() {
774            slot.cancellation_token.cancel();
775            if let Some(handle) = slot.task_handle.as_ref() {
776                handle.abort();
777            }
778
779            if let Some(control) = &slot.socket_control {
780                control.deregister();
781            }
782        }
783    }
784}
785
786struct ConnectionSlotBatch<'a> {
787    owner: &'a Mutex<Vec<ConnectionSlot>>,
788    slots: Vec<ConnectionSlot>,
789}
790
791impl<'a> ConnectionSlotBatch<'a> {
792    fn take(owner: &'a Mutex<Vec<ConnectionSlot>>) -> Self {
793        let slots = std::mem::take(&mut *owner.lock());
794        Self { owner, slots }
795    }
796}
797
798impl Drop for ConnectionSlotBatch<'_> {
799    fn drop(&mut self) {
800        self.owner.lock().extend(self.slots.drain(..));
801    }
802}
803
804#[cfg(test)]
805mod tests {
806    use rstest::rstest;
807
808    use super::*;
809
810    #[rstest]
811    #[case("wss://stream.example.com/feed", "wss://stream.example.com")]
812    #[case(
813        "wss://SYN-user:SYN-password@stream.example.com:9443/SYN-path?token=SYN-query#SYN-fragment",
814        "wss://stream.example.com:9443"
815    )]
816    fn test_debug_preserves_url_origin(#[case] url: &str, #[case] origin: &str) {
817        let client = BinanceSpotWebSocketClient::new(
818            Some(url.to_string()),
819            None,
820            None,
821            None,
822            TransportBackend::default(),
823        )
824        .unwrap();
825        let debug = format!("{client:?}");
826        assert!(debug.contains(&format!("url: {origin:?}")));
827        assert!(!debug.contains("SYN-"));
828        assert_eq!(client.url.expose(), url);
829    }
830
831    #[tokio::test]
832    async fn test_cancelled_close_retains_connection_slot() {
833        let mut client =
834            BinanceSpotWebSocketClient::new(None, None, None, None, TransportBackend::default())
835                .unwrap();
836        let (cmd_tx, mut cmd_rx) = tokio::sync::mpsc::unbounded_channel();
837        client.slots.lock().push(ConnectionSlot {
838            cmd_tx,
839            streams: Vec::new(),
840            subscriptions_state: SubscriptionState::new('@'),
841            task_handle: TaskSlot::from_handle(tokio::spawn(std::future::pending())),
842            cancellation_token: CancellationToken::new(),
843            connection_mode: Arc::new(AtomicU8::new(ConnectionMode::Active as u8)),
844            socket_control: None,
845            shutdown_errors: Vec::new(),
846        });
847
848        {
849            let close = client.close();
850            tokio::pin!(close);
851            tokio::select! {
852                result = &mut close => panic!("close completed unexpectedly: {result:?}"),
853                command = cmd_rx.recv() => assert!(command.is_some()),
854            }
855        }
856
857        let slots = client.slots.lock();
858        assert_eq!(slots.len(), 1);
859        assert!(slots[0].task_handle.is_some());
860    }
861
862    #[rstest]
863    fn test_new_rejects_hmac_secret_with_actionable_error() {
864        // An all-zero 48-byte buffer base64-encodes to a non-Ed25519 secret
865        // (no PKCS#8 OID), standing in for an HMAC key. SBE market-data streams
866        // require Ed25519, so construction must fail with guidance that names
867        // HMAC, not the raw OID error.
868        let secret = base64::Engine::encode(&base64::engine::general_purpose::STANDARD, [0u8; 48]);
869
870        let result = BinanceSpotWebSocketClient::new(
871            None,
872            Some("test_key".to_string()),
873            Some(secret),
874            None,
875            TransportBackend::default(),
876        );
877
878        let err = result.expect_err("HMAC secret must be rejected");
879        let msg = err.to_string();
880        assert!(
881            msg.contains("Ed25519"),
882            "error should mention Ed25519, was: {msg}"
883        );
884        assert!(
885            msg.contains("HMAC"),
886            "error should mention HMAC, was: {msg}"
887        );
888    }
889
890    #[rstest]
891    fn test_with_proxy_preserves_proxy_url() {
892        let client =
893            BinanceSpotWebSocketClient::new(None, None, None, None, TransportBackend::default())
894                .unwrap()
895                .with_proxy(Some("socks5://proxy.example:1080".to_string()));
896
897        assert_eq!(
898            client.proxy_url.as_ref().map(SecretString::expose_secret),
899            Some("socks5://proxy.example:1080")
900        );
901    }
902}
nautilus_binance/spot/websocket/streams/client.rs

nautilus_binance/spot/websocket/streams/
client.rs
