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, Mutex,
31        atomic::{AtomicBool, AtomicU8, AtomicU64, Ordering},
32    },
33};
34
35use futures_util::Stream;
36use nautilus_common::live::get_runtime;
37use nautilus_core::{AtomicMap, string::secret::REDACTED};
38use nautilus_model::instruments::{Instrument, InstrumentAny};
39use nautilus_network::{
40    mode::ConnectionMode,
41    websocket::{
42        PingHandler, SubscriptionState, TransportBackend, WebSocketClient, WebSocketConfig,
43        channel_message_handler,
44    },
45};
46use tokio_util::sync::CancellationToken;
47use ustr::Ustr;
48
49use super::{
50    super::error::{BinanceWsError, BinanceWsResult},
51    handler::BinanceSpotWsFeedHandler,
52    messages::{BinanceSpotWsMessage, BinanceSpotWsStreamsCommand},
53    subscription::{MAX_CONNECTIONS, MAX_STREAMS_PER_CONNECTION},
54};
55use crate::common::{
56    consts::{
57        BINANCE_API_KEY_HEADER, BINANCE_RATE_LIMIT_KEY_SUBSCRIPTION, BINANCE_SPOT_SBE_WS_URL,
58        BINANCE_WS_CONNECTION_QUOTA, BINANCE_WS_SUBSCRIPTION_QUOTA,
59    },
60    credential::Ed25519Credential,
61};
62
63// State for a single WebSocket connection within the pool
64struct ConnectionSlot {
65    cmd_tx: tokio::sync::mpsc::UnboundedSender<BinanceSpotWsStreamsCommand>,
66    streams: Vec<String>,
67    subscriptions_state: SubscriptionState,
68    task_handle: tokio::task::JoinHandle<()>,
69    cancellation_token: CancellationToken,
70    connection_mode: Arc<AtomicU8>,
71}
72
73/// Binance Spot WebSocket client for SBE market data streams.
74///
75/// Manages a pool of up to 20 connections, each supporting up to 1024 streams.
76/// New connections are created automatically when subscribing exceeds the current
77/// connection's stream limit. All connections feed into a single output stream,
78/// transparent to the data client.
79#[derive(Clone)]
80pub struct BinanceSpotWebSocketClient {
81    url: String,
82    credential: Option<Arc<Ed25519Credential>>,
83    heartbeat: Option<u64>,
84    signal: Arc<AtomicBool>,
85    slots: Arc<Mutex<Vec<ConnectionSlot>>>,
86    connect_lock: Arc<tokio::sync::Mutex<()>>,
87    out_tx: Arc<Mutex<Option<tokio::sync::mpsc::UnboundedSender<BinanceSpotWsMessage>>>>,
88    out_rx: Arc<Mutex<Option<tokio::sync::mpsc::UnboundedReceiver<BinanceSpotWsMessage>>>>,
89    request_id_counter: Arc<AtomicU64>,
90    instruments_cache: Arc<AtomicMap<Ustr, InstrumentAny>>,
91    transport_backend: TransportBackend,
92}
93
94impl Debug for BinanceSpotWebSocketClient {
95    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
96        f.debug_struct(stringify!(BinanceSpotWebSocketClient))
97            .field("url", &self.url)
98            .field("credential", &self.credential.as_ref().map(|_| REDACTED))
99            .field("heartbeat", &self.heartbeat)
100            .finish_non_exhaustive()
101    }
102}
103
104impl Default for BinanceSpotWebSocketClient {
105    fn default() -> Self {
106        Self::new(None, None, None, None, TransportBackend::default()).unwrap()
107    }
108}
109
110impl BinanceSpotWebSocketClient {
111    /// Creates a new [`BinanceSpotWebSocketClient`] instance.
112    ///
113    /// # Errors
114    ///
115    /// Returns an error if credential creation fails.
116    pub fn new(
117        url: Option<String>,
118        api_key: Option<String>,
119        api_secret: Option<String>,
120        heartbeat: Option<u64>,
121        transport_backend: TransportBackend,
122    ) -> anyhow::Result<Self> {
123        let url = url.unwrap_or(BINANCE_SPOT_SBE_WS_URL.to_string());
124
125        let credential = match (api_key, api_secret) {
126            (Some(key), Some(secret)) => {
127                let credential = Ed25519Credential::new(key, &secret).map_err(|e| {
128                    anyhow::anyhow!(
129                        "Binance Spot SBE market-data streams require an Ed25519 API key \
130                         (HMAC keys are not supported): {e}"
131                    )
132                })?;
133                Some(Arc::new(credential))
134            }
135            _ => None,
136        };
137
138        Ok(Self {
139            url,
140            credential,
141            heartbeat,
142            signal: Arc::new(AtomicBool::new(false)),
143            slots: Arc::new(Mutex::new(Vec::new())),
144            connect_lock: Arc::new(tokio::sync::Mutex::new(())),
145            out_tx: Arc::new(Mutex::new(None)),
146            out_rx: Arc::new(Mutex::new(None)),
147            request_id_counter: Arc::new(AtomicU64::new(1)),
148            instruments_cache: Arc::new(AtomicMap::new()),
149            transport_backend,
150        })
151    }
152
153    /// Returns whether API credentials are configured.
154    #[must_use]
155    pub fn has_credentials(&self) -> bool {
156        self.credential.is_some()
157    }
158
159    /// Returns whether any connection in the pool is active.
160    #[must_use]
161    #[expect(clippy::missing_panics_doc, reason = "mutex poisoning is not expected")]
162    pub fn is_active(&self) -> bool {
163        let slots = self.slots.lock().expect("slots lock poisoned");
164        slots
165            .iter()
166            .any(|s| s.connection_mode.load(Ordering::Relaxed) == ConnectionMode::Active as u8)
167    }
168
169    /// Returns whether all connections in the pool are closed.
170    #[must_use]
171    #[expect(clippy::missing_panics_doc, reason = "mutex poisoning is not expected")]
172    pub fn is_closed(&self) -> bool {
173        let slots = self.slots.lock().expect("slots lock poisoned");
174        slots.is_empty()
175            || slots
176                .iter()
177                .all(|s| s.connection_mode.load(Ordering::Relaxed) == ConnectionMode::Closed as u8)
178    }
179
180    /// Returns the total number of confirmed subscriptions across all connections.
181    #[must_use]
182    #[expect(clippy::missing_panics_doc, reason = "mutex poisoning is not expected")]
183    pub fn subscription_count(&self) -> usize {
184        let slots = self.slots.lock().expect("slots lock poisoned");
185        slots.iter().map(|s| s.subscriptions_state.len()).sum()
186    }
187
188    /// Connects the first WebSocket connection in the pool.
189    ///
190    /// # Errors
191    ///
192    /// Returns an error if connection fails.
193    #[expect(clippy::missing_panics_doc, reason = "mutex poisoning is not expected")]
194    pub async fn connect(&mut self) -> BinanceWsResult<()> {
195        self.signal.store(false, Ordering::Relaxed);
196
197        let (out_tx, out_rx) = tokio::sync::mpsc::unbounded_channel();
198        *self.out_tx.lock().expect("out_tx lock poisoned") = Some(out_tx);
199        *self.out_rx.lock().expect("out_rx lock poisoned") = Some(out_rx);
200
201        let slot = self.create_connection().await?;
202        self.slots.lock().expect("slots lock poisoned").push(slot);
203
204        log::debug!(
205            "Connected to Binance Spot SBE stream pool: url={}",
206            self.url
207        );
208        Ok(())
209    }
210
211    /// Closes all WebSocket connections in the pool.
212    ///
213    /// # Errors
214    ///
215    /// Returns an error if disconnect fails.
216    #[expect(clippy::missing_panics_doc, reason = "mutex poisoning is not expected")]
217    pub async fn close(&mut self) -> BinanceWsResult<()> {
218        self.signal.store(true, Ordering::Relaxed);
219
220        let slots: Vec<ConnectionSlot> = {
221            let mut guard = self.slots.lock().expect("slots lock poisoned");
222            guard.drain(..).collect()
223        };
224
225        for slot in slots {
226            slot.cancellation_token.cancel();
227            let _result = slot.cmd_tx.send(BinanceSpotWsStreamsCommand::Disconnect);
228            let _result = slot.task_handle.await;
229        }
230
231        *self.out_tx.lock().expect("out_tx lock poisoned") = None;
232        *self.out_rx.lock().expect("out_rx lock poisoned") = None;
233
234        log::debug!("Disconnected from Binance Spot SBE stream pool");
235        Ok(())
236    }
237
238    /// Subscribes to the specified streams.
239    ///
240    /// Streams are distributed across pool connections. New connections are created
241    /// automatically when existing ones reach the 1024-stream limit, up to a maximum
242    /// of 20 connections.
243    ///
244    /// # Errors
245    ///
246    /// Returns an error if the pool is exhausted or command delivery fails.
247    #[expect(clippy::missing_panics_doc, reason = "mutex poisoning is not expected")]
248    pub async fn subscribe(&self, streams: Vec<String>) -> BinanceWsResult<()> {
249        // Serialize all phases so concurrent subscribers see a consistent
250        // pool state and can't trigger spurious `Pool exhausted`.
251        let _connect_guard = self.connect_lock.lock().await;
252
253        // Phase 1: filter already-subscribed streams (brief lock).
254        let new_streams: Vec<String> = {
255            let slots = self.slots.lock().expect("slots lock poisoned");
256            streams
257                .into_iter()
258                .filter(|s| !slots.iter().any(|slot| slot.streams.contains(s)))
259                .collect()
260        };
261
262        if new_streams.is_empty() {
263            return Ok(());
264        }
265
266        // Phase 2: create connections if needed.
267        loop {
268            let (remaining_capacity, slot_count) = {
269                let slots = self.slots.lock().expect("slots lock poisoned");
270                let cap: usize = slots
271                    .iter()
272                    .map(|s| MAX_STREAMS_PER_CONNECTION - s.streams.len())
273                    .sum();
274                (cap, slots.len())
275            };
276
277            if remaining_capacity >= new_streams.len() || slot_count >= MAX_CONNECTIONS {
278                break;
279            }
280
281            let new_slot = self.create_connection().await?;
282            let slot_count = {
283                let mut slots = self.slots.lock().expect("slots lock poisoned");
284                slots.push(new_slot);
285                slots.len()
286            };
287            log::debug!("Pool slot {} connected: url={}", slot_count - 1, self.url);
288        }
289
290        // Phase 3: assign streams to slots and send commands (brief lock).
291        // Stage assignments first so a capacity error leaves slots unchanged.
292        let mut slots = self.slots.lock().expect("slots lock poisoned");
293        let mut slot_batches: Vec<(usize, Vec<String>)> = Vec::new();
294        let mut slot_counts: Vec<usize> = slots.iter().map(|s| s.streams.len()).collect();
295
296        for stream in &new_streams {
297            let slot_idx = slot_counts
298                .iter()
299                .position(|&count| count < MAX_STREAMS_PER_CONNECTION)
300                .ok_or_else(|| {
301                    let max_total = MAX_CONNECTIONS * MAX_STREAMS_PER_CONNECTION;
302                    BinanceWsError::ClientError(format!(
303                        "Pool exhausted: {max_total} total subscriptions \
304                         ({MAX_CONNECTIONS} connections x {MAX_STREAMS_PER_CONNECTION} streams)"
305                    ))
306                })?;
307
308            slot_counts[slot_idx] += 1;
309
310            if let Some(batch) = slot_batches.iter_mut().find(|(i, _)| *i == slot_idx) {
311                batch.1.push(stream.clone());
312            } else {
313                slot_batches.push((slot_idx, vec![stream.clone()]));
314            }
315        }
316
317        // Send commands first; only update slot state on success
318        for (slot_idx, batch) in &slot_batches {
319            slots[*slot_idx]
320                .cmd_tx
321                .send(BinanceSpotWsStreamsCommand::Subscribe {
322                    streams: batch.clone(),
323                })
324                .map_err(|e| {
325                    BinanceWsError::ClientError(format!(
326                        "Handler not available for pool slot {slot_idx}: {e}"
327                    ))
328                })?;
329            slots[*slot_idx].streams.extend(batch.iter().cloned());
330        }
331
332        Ok(())
333    }
334
335    /// Unsubscribes from the specified streams.
336    ///
337    /// # Errors
338    ///
339    /// Returns an error if command delivery fails.
340    #[expect(clippy::missing_panics_doc, reason = "mutex poisoning is not expected")]
341    pub async fn unsubscribe(&self, streams: Vec<String>) -> BinanceWsResult<()> {
342        let mut slots = self.slots.lock().expect("slots lock poisoned");
343        let mut slot_batches: Vec<(usize, Vec<String>)> = Vec::new();
344
345        for stream in &streams {
346            if let Some(slot_idx) = slots.iter().position(|s| s.streams.contains(stream)) {
347                if let Some(batch) = slot_batches.iter_mut().find(|(i, _)| *i == slot_idx) {
348                    batch.1.push(stream.clone());
349                } else {
350                    slot_batches.push((slot_idx, vec![stream.clone()]));
351                }
352            }
353        }
354
355        // Send commands first; only update slot state on success
356        for (slot_idx, batch) in &slot_batches {
357            slots[*slot_idx]
358                .cmd_tx
359                .send(BinanceSpotWsStreamsCommand::Unsubscribe {
360                    streams: batch.clone(),
361                })
362                .map_err(|e| {
363                    BinanceWsError::ClientError(format!(
364                        "Handler not available for pool slot {slot_idx}: {e}"
365                    ))
366                })?;
367
368            for stream in batch {
369                slots[*slot_idx].streams.retain(|s| s != stream);
370            }
371        }
372
373        Ok(())
374    }
375
376    /// Returns a stream of messages from all WebSocket connections.
377    ///
378    /// This method can only be called once per connection lifecycle. Subsequent calls
379    /// return an empty stream.
380    ///
381    /// # Panics
382    ///
383    /// Panics if the internal output receiver mutex is poisoned.
384    pub fn stream(&self) -> impl Stream<Item = BinanceSpotWsMessage> + 'static {
385        let out_rx = self.out_rx.lock().expect("out_rx lock poisoned").take();
386        async_stream::stream! {
387            if let Some(mut rx) = out_rx {
388                while let Some(msg) = rx.recv().await {
389                    yield msg;
390                }
391            }
392        }
393    }
394
395    /// Bulk initialize the instrument cache.
396    pub fn cache_instruments(&self, instruments: &[InstrumentAny]) {
397        self.instruments_cache.rcu(|m| {
398            for inst in instruments {
399                m.insert(inst.symbol().inner(), inst.clone());
400            }
401        });
402    }
403
404    /// Update a single instrument in the cache.
405    pub fn cache_instrument(&self, instrument: InstrumentAny) {
406        self.instruments_cache
407            .insert(instrument.symbol().inner(), instrument);
408    }
409
410    /// Returns a shared reference to the instruments cache.
411    #[must_use]
412    pub fn instruments_cache(&self) -> Arc<AtomicMap<Ustr, InstrumentAny>> {
413        self.instruments_cache.clone()
414    }
415
416    /// Returns an instrument from the cache by symbol.
417    #[must_use]
418    pub fn get_instrument(&self, symbol: &str) -> Option<InstrumentAny> {
419        self.instruments_cache.get_cloned(&Ustr::from(symbol))
420    }
421
422    async fn create_connection(&self) -> BinanceWsResult<ConnectionSlot> {
423        let out_tx = self
424            .out_tx
425            .lock()
426            .expect("out_tx lock poisoned")
427            .clone()
428            .ok_or_else(|| {
429                BinanceWsError::ClientError("Output channel not initialized".to_string())
430            })?;
431
432        let (raw_handler, raw_rx) = channel_message_handler();
433        let ping_handler: PingHandler = Arc::new(move |_| {});
434
435        let headers = if let Some(ref cred) = self.credential {
436            vec![(
437                BINANCE_API_KEY_HEADER.to_string(),
438                cred.api_key().to_string(),
439            )]
440        } else {
441            vec![]
442        };
443
444        let config = WebSocketConfig {
445            url: self.url.clone(),
446            headers,
447            heartbeat: self.heartbeat,
448            heartbeat_msg: None,
449            reconnect_timeout_ms: Some(5_000),
450            reconnect_delay_initial_ms: Some(500),
451            reconnect_delay_max_ms: Some(5_000),
452            reconnect_backoff_factor: Some(2.0),
453            reconnect_jitter_ms: Some(250),
454            reconnect_max_attempts: None,
455            idle_timeout_ms: None,
456            backend: self.transport_backend,
457            proxy_url: None,
458        };
459
460        let keyed_quotas = vec![(
461            BINANCE_RATE_LIMIT_KEY_SUBSCRIPTION[0].as_str().to_string(),
462            *BINANCE_WS_SUBSCRIPTION_QUOTA,
463        )];
464
465        let client = WebSocketClient::connect(
466            config,
467            Some(raw_handler),
468            Some(ping_handler),
469            None,
470            keyed_quotas,
471            Some(*BINANCE_WS_CONNECTION_QUOTA),
472        )
473        .await
474        .map_err(|e| {
475            log::error!("WebSocket connection failed: {e}");
476            BinanceWsError::NetworkError(e.to_string())
477        })?;
478
479        let connection_mode = client.connection_mode_atomic();
480        let subscriptions_state = SubscriptionState::new('@');
481        let cancellation_token = CancellationToken::new();
482
483        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
484
485        let mut handler = BinanceSpotWsFeedHandler::new(
486            self.signal.clone(),
487            cmd_rx,
488            raw_rx,
489            out_tx.clone(),
490            subscriptions_state.clone(),
491            self.request_id_counter.clone(),
492        );
493
494        cmd_tx
495            .send(BinanceSpotWsStreamsCommand::SetClient(client))
496            .map_err(|e| BinanceWsError::ClientError(format!("Failed to set client: {e}")))?;
497
498        let signal = self.signal.clone();
499        let token = cancellation_token.clone();
500        let subs = subscriptions_state.clone();
501        let resubscribe_tx = cmd_tx.clone();
502
503        let task_handle = get_runtime().spawn(async move {
504            loop {
505                tokio::select! {
506                    () = token.cancelled() => {
507                        log::debug!("Handler task cancelled");
508                        break;
509                    }
510                    result = handler.next() => {
511                        match result {
512                            Some(BinanceSpotWsMessage::Reconnected) => {
513                                log::info!("WebSocket reconnected, restoring subscriptions");
514                                let all_topics = subs.all_topics();
515                                for topic in &all_topics {
516                                    subs.mark_failure(topic);
517                                }
518
519                                let streams = subs.all_topics();
520                                if !streams.is_empty()
521                                    && let Err(e) = resubscribe_tx.send(BinanceSpotWsStreamsCommand::Subscribe { streams }) {
522                                        log::error!("Failed to resubscribe after reconnect: {e}");
523                                    }
524
525                                if out_tx.send(BinanceSpotWsMessage::Reconnected).is_err() {
526                                    log::debug!("Output channel closed");
527                                    break;
528                                }
529                            }
530                            Some(msg) => {
531                                if out_tx.send(msg).is_err() {
532                                    log::debug!("Output channel closed");
533                                    break;
534                                }
535                            }
536                            None => {
537                                if signal.load(Ordering::Relaxed) {
538                                    log::debug!("Handler received shutdown signal");
539                                } else {
540                                    log::warn!("Handler loop ended unexpectedly");
541                                }
542                                break;
543                            }
544                        }
545                    }
546                }
547            }
548        });
549
550        Ok(ConnectionSlot {
551            cmd_tx,
552            streams: Vec::new(),
553            subscriptions_state,
554            task_handle,
555            cancellation_token,
556            connection_mode,
557        })
558    }
559}
560
561#[cfg(test)]
562mod tests {
563    use rstest::rstest;
564
565    use super::*;
566
567    #[rstest]
568    fn test_new_rejects_hmac_secret_with_actionable_error() {
569        // An all-zero 48-byte buffer base64-encodes to a non-Ed25519 secret
570        // (no PKCS#8 OID), standing in for an HMAC key. SBE market-data streams
571        // require Ed25519, so construction must fail with guidance that names
572        // HMAC, not the raw OID error.
573        let secret = base64::Engine::encode(&base64::engine::general_purpose::STANDARD, [0u8; 48]);
574
575        let result = BinanceSpotWebSocketClient::new(
576            None,
577            Some("test_key".to_string()),
578            Some(secret),
579            None,
580            TransportBackend::default(),
581        );
582
583        let err = result.expect_err("HMAC secret must be rejected");
584        let msg = err.to_string();
585        assert!(
586            msg.contains("Ed25519"),
587            "error should mention Ed25519, was: {msg}"
588        );
589        assert!(
590            msg.contains("HMAC"),
591            "error should mention HMAC, was: {msg}"
592        );
593    }
594}
nautilus_binance/spot/websocket/streams/client.rs

nautilus_binance/spot/websocket/streams/
client.rs
