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

nautilus_binance/spot/websocket/streams/
client.rs
