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

nautilus_binance/spot/websocket/streams/
client.rs
