From 89b903f95dcc4598432adcfcc6dd51975674a130 Mon Sep 17 00:00:00 2001 From: quran Date: Sat, 31 Jan 2026 18:25:12 +0800 Subject: [PATCH] refactor(binance): update WebSocket API to return receivers and improve health monitoring - Change subscription methods to return tokio::sync::mpsc::Receiver - Add channel capacity configuration with customizable backpressure settings - Implement WsHealthSnapshot with latency, message counts, and uptime metrics - Add graceful shutdown mechanism with proper resource cleanup - Introduce DepthLevel and UpdateSpeed enums for order book configuration - Add retry logic with configurable attempts for listen key refresh operations - Implement symbol normalization for consistent subscription handling - Add comprehensive documentation and examples for WebSocket usage - Include Drop implementation to warn about unclosed connections - Add health monitoring with message receive/drop statistics --- ccxt-core/src/types/orderbook.rs | 4 +- ccxt-exchanges/src/binance/ws/binance_impl.rs | 76 +- ccxt-exchanges/src/binance/ws/handlers.rs | 53 +- ccxt-exchanges/src/binance/ws/listen_key.rs | 89 +- ccxt-exchanges/src/binance/ws/mod.rs | 934 ++++++++++++++---- ccxt-exchanges/src/binance/ws/parsers.rs | 126 +++ ccxt-exchanges/src/binance/ws/streams.rs | 50 + ccxt-exchanges/src/binance/ws/types.rs | 431 ++++++++ .../src/binance/ws_exchange_impl.rs | 12 +- 9 files changed, 1491 insertions(+), 284 deletions(-) create mode 100644 ccxt-exchanges/src/binance/ws/parsers.rs create mode 100644 ccxt-exchanges/src/binance/ws/types.rs diff --git a/ccxt-core/src/types/orderbook.rs b/ccxt-core/src/types/orderbook.rs index 5c5446c..3c04e26 100644 --- a/ccxt-core/src/types/orderbook.rs +++ b/ccxt-core/src/types/orderbook.rs @@ -9,8 +9,8 @@ use super::{Amount, Price, Symbol, Timestamp}; /// Maximum number of buffered messages for orderbook sync const MAX_BUFFERED_MESSAGES: usize = 100; -/// Minimum resync interval in milliseconds -const MIN_RESYNC_INTERVAL_MS: i64 = 1000; +/// Minimum resync interval in milliseconds (5 seconds) +const MIN_RESYNC_INTERVAL_MS: i64 = 5000; /// Orderbook delta update message #[derive(Debug, Clone, Serialize, Deserialize)] diff --git a/ccxt-exchanges/src/binance/ws/binance_impl.rs b/ccxt-exchanges/src/binance/ws/binance_impl.rs index f4b3c52..8e84f77 100644 --- a/ccxt-exchanges/src/binance/ws/binance_impl.rs +++ b/ccxt-exchanges/src/binance/ws/binance_impl.rs @@ -2,31 +2,59 @@ // It's separated to keep mod.rs under 800 lines impl Binance { - /// Subscribes to the ticker stream for a unified symbol - pub async fn subscribe_ticker(&self, symbol: &str) -> Result<()> { + /// Subscribes to the ticker stream for a unified symbol. + /// + /// Returns a receiver that will receive ticker updates as JSON values. + pub async fn subscribe_ticker( + &self, + symbol: &str, + ) -> Result> { let ws = self.connection_manager.get_public_connection().await?; let binance_symbol = symbol.replace('/', "").to_lowercase(); ws.subscribe_ticker(&binance_symbol).await } - /// Subscribes to the trade stream for a unified symbol - pub async fn subscribe_trades(&self, symbol: &str) -> Result<()> { + /// Subscribes to the trade stream for a unified symbol. + /// + /// Returns a receiver that will receive trade updates as JSON values. + pub async fn subscribe_trades( + &self, + symbol: &str, + ) -> Result> { let ws = self.connection_manager.get_public_connection().await?; let binance_symbol = symbol.replace('/', "").to_lowercase(); ws.subscribe_trades(&binance_symbol).await } - /// Subscribes to the order book stream for a unified symbol - pub async fn subscribe_orderbook(&self, symbol: &str, levels: Option) -> Result<()> { + /// Subscribes to the order book stream for a unified symbol. + /// + /// Returns a receiver that will receive order book updates as JSON values. + pub async fn subscribe_orderbook( + &self, + symbol: &str, + levels: Option, + ) -> Result> { + use super::ws::{DepthLevel, UpdateSpeed}; + let ws = self.connection_manager.get_public_connection().await?; let binance_symbol = symbol.replace('/', "").to_lowercase(); - let depth_levels = levels.unwrap_or(20); - ws.subscribe_orderbook(&binance_symbol, depth_levels, "1000ms") + let depth_level = match levels.unwrap_or(20) { + 5 => DepthLevel::L5, + 10 => DepthLevel::L10, + _ => DepthLevel::L20, + }; + ws.subscribe_orderbook(&binance_symbol, depth_level, UpdateSpeed::Ms1000) .await } - /// Subscribes to the candlestick stream for a unified symbol - pub async fn subscribe_kline(&self, symbol: &str, interval: &str) -> Result<()> { + /// Subscribes to the candlestick stream for a unified symbol. + /// + /// Returns a receiver that will receive kline updates as JSON values. + pub async fn subscribe_kline( + &self, + symbol: &str, + interval: &str, + ) -> Result> { let ws = self.connection_manager.get_public_connection().await?; let binance_symbol = symbol.replace('/', "").to_lowercase(); ws.subscribe_kline(&binance_symbol, interval).await @@ -137,6 +165,8 @@ impl Binance { limit: Option, params: Option>, ) -> Result { + use super::ws::UpdateSpeed; + self.load_markets(false).await?; let market = self.base.market(symbol).await?; @@ -146,19 +176,14 @@ impl Binance { market.market_type == MarketType::Swap || market.market_type == MarketType::Futures; let update_speed = if let Some(p) = ¶ms { - p.get("speed") - .and_then(serde_json::Value::as_i64) - .unwrap_or(100) as i32 + match p.get("speed").and_then(serde_json::Value::as_i64).unwrap_or(100) { + 1000 => UpdateSpeed::Ms1000, + _ => UpdateSpeed::Ms100, + } } else { - 100 + UpdateSpeed::Ms100 }; - if update_speed != 100 && update_speed != 1000 { - return Err(Error::invalid_request( - "Update speed must be 100 or 1000 milliseconds", - )); - } - let ws = self.connection_manager.get_public_connection().await?; ws.watch_orderbook_internal(self, &binance_symbol, limit, update_speed, is_futures) .await @@ -171,6 +196,8 @@ impl Binance { limit: Option, params: Option>, ) -> Result> { + use super::ws::UpdateSpeed; + if symbols.is_empty() { return Err(Error::invalid_request("Symbols list cannot be empty")); } @@ -201,11 +228,12 @@ impl Binance { } let update_speed = if let Some(p) = ¶ms { - p.get("speed") - .and_then(serde_json::Value::as_i64) - .unwrap_or(100) as i32 + match p.get("speed").and_then(serde_json::Value::as_i64).unwrap_or(100) { + 1000 => UpdateSpeed::Ms1000, + _ => UpdateSpeed::Ms100, + } } else { - 100 + UpdateSpeed::Ms100 }; let ws = self.connection_manager.get_public_connection().await?; diff --git a/ccxt-exchanges/src/binance/ws/handlers.rs b/ccxt-exchanges/src/binance/ws/handlers.rs index 490e57a..816ee6b 100644 --- a/ccxt-exchanges/src/binance/ws/handlers.rs +++ b/ccxt-exchanges/src/binance/ws/handlers.rs @@ -167,6 +167,31 @@ impl MessageRouter { self.is_connected.load(std::sync::atomic::Ordering::SeqCst) } + /// Returns the current connection latency in milliseconds. + /// + /// This is calculated from the ping/pong round-trip time. + /// Returns None if no latency data is available. + pub fn latency(&self) -> Option { + // Try to get latency from the underlying WsClient + // This requires a blocking read, so we use try_read + if let Ok(guard) = self.ws_client.try_read() { + if let Some(ref client) = *guard { + return client.latency(); + } + } + None + } + + /// Returns the number of reconnection attempts. + pub fn reconnect_count(&self) -> u32 { + if let Ok(guard) = self.ws_client.try_read() { + if let Some(ref client) = *guard { + return client.reconnect_count(); + } + } + 0 + } + /// Applies a new reconnection configuration pub async fn set_reconnect_config(&self, config: super::subscriptions::ReconnectConfig) { *self.reconnect_config.write().await = config; @@ -416,36 +441,20 @@ impl MessageRouter { let symbol_opt = payload.get("s").and_then(|s| s.as_str()); if let Some(symbol) = symbol_opt { - let mut active_streams = subscription_manager - .get_subscriptions_by_symbol(symbol) + // Normalize symbol to lowercase since subscriptions are stored normalized + let normalized_symbol = symbol.to_lowercase(); + let active_streams = subscription_manager + .get_subscriptions_by_symbol(&normalized_symbol) .await; tracing::debug!( - "Routing message for symbol {}: stream_name={}, active_subscriptions={}", + "Routing message for symbol {} (normalized: {}): stream_name={}, active_subscriptions={}", symbol, + normalized_symbol, stream_name, active_streams.len() ); - // If no streams found for uppercase symbol (e.g. "BTCUSDT"), try lowercase (e.g. "btcusdt") - // This is common since Binance API sends uppercase symbols but we usually subscribe with lowercase - if active_streams.is_empty() { - let lower_symbol = symbol.to_lowercase(); - if lower_symbol != symbol { - let lower_streams = subscription_manager - .get_subscriptions_by_symbol(&lower_symbol) - .await; - if !lower_streams.is_empty() { - active_streams = lower_streams; - tracing::debug!( - "Found subscriptions for lowercased symbol {}: count={}", - lower_symbol, - active_streams.len() - ); - } - } - } - let mut fallback_sent = false; for sub in active_streams { diff --git a/ccxt-exchanges/src/binance/ws/listen_key.rs b/ccxt-exchanges/src/binance/ws/listen_key.rs index aab7870..9fb2fd9 100644 --- a/ccxt-exchanges/src/binance/ws/listen_key.rs +++ b/ccxt-exchanges/src/binance/ws/listen_key.rs @@ -7,8 +7,24 @@ use std::time::{Duration, Instant}; use tokio::sync::{Mutex, RwLock}; use tokio::task::JoinHandle; -/// Listen key refresh interval (30 minutes) -const LISTEN_KEY_REFRESH_INTERVAL: Duration = Duration::from_secs(30 * 60); +/// Listen key refresh interval (25 minutes) +/// +/// Binance listen keys expire after 60 minutes. We refresh at 25 minutes +/// to provide a safety margin and avoid expiration during network delays. +const LISTEN_KEY_REFRESH_INTERVAL: Duration = Duration::from_secs(25 * 60); + +/// Maximum age before forcing a new listen key (20 minutes) +/// +/// If a listen key is older than this, we create a new one instead of +/// returning the cached key. This is more conservative than the refresh +/// interval to handle cases where auto-refresh might have failed. +const LISTEN_KEY_MAX_AGE: Duration = Duration::from_secs(20 * 60); + +/// Maximum number of retry attempts for refresh failures +const MAX_REFRESH_RETRIES: u32 = 3; + +/// Delay between refresh retry attempts +const REFRESH_RETRY_DELAY: Duration = Duration::from_secs(5); /// Listen key manager /// @@ -50,8 +66,13 @@ impl ListenKeyManager { let created = self.created_at.read().await; if let Some(created_time) = *created { let elapsed = created_time.elapsed(); - if elapsed > Duration::from_secs(50 * 60) { + // Use the more conservative max age threshold + if elapsed > LISTEN_KEY_MAX_AGE { drop(created); + tracing::info!( + "Listen key is {} minutes old, creating new one", + elapsed.as_secs() / 60 + ); return self.create_new().await; } } @@ -105,33 +126,49 @@ impl ListenKeyManager { let key_opt = listen_key.read().await.clone(); if let Some(key) = key_opt { - match binance.refresh_listen_key(&key).await { - Ok(()) => { - *created_at.write().await = Some(Instant::now()); - } - Err(e) => { - tracing::warn!( - "Failed to refresh listen key: {}. Attempting regeneration.", - e - ); - match binance.create_listen_key().await { - Ok(new_key) => { - tracing::info!("Regenerated listen key successfully"); - *listen_key.write().await = Some(new_key); - *created_at.write().await = Some(Instant::now()); - } - Err(create_err) => { - tracing::error!( - "Failed to regenerate listen key: {}", - create_err - ); - *listen_key.write().await = None; - *created_at.write().await = None; - break; + // Try to refresh with retries + let mut refresh_succeeded = false; + for attempt in 1..=MAX_REFRESH_RETRIES { + match binance.refresh_listen_key(&key).await { + Ok(()) => { + *created_at.write().await = Some(Instant::now()); + tracing::debug!("Listen key refreshed successfully"); + refresh_succeeded = true; + break; + } + Err(e) => { + tracing::warn!( + "Failed to refresh listen key (attempt {}/{}): {}", + attempt, + MAX_REFRESH_RETRIES, + e + ); + if attempt < MAX_REFRESH_RETRIES { + tokio::time::sleep(REFRESH_RETRY_DELAY).await; } } } } + + // If all refresh attempts failed, try to regenerate + if !refresh_succeeded { + tracing::warn!( + "All refresh attempts failed, attempting to regenerate listen key" + ); + match binance.create_listen_key().await { + Ok(new_key) => { + tracing::info!("Regenerated listen key successfully"); + *listen_key.write().await = Some(new_key); + *created_at.write().await = Some(Instant::now()); + } + Err(create_err) => { + tracing::error!("Failed to regenerate listen key: {}", create_err); + *listen_key.write().await = None; + *created_at.write().await = None; + break; + } + } + } } else { break; } diff --git a/ccxt-exchanges/src/binance/ws/mod.rs b/ccxt-exchanges/src/binance/ws/mod.rs index 192f2e9..205e6a4 100644 --- a/ccxt-exchanges/src/binance/ws/mod.rs +++ b/ccxt-exchanges/src/binance/ws/mod.rs @@ -1,21 +1,73 @@ //! Binance WebSocket implementation //! -//! Provides WebSocket real-time data stream subscriptions for the Binance exchange +//! Provides WebSocket real-time data stream subscriptions for the Binance exchange. +//! +//! # Architecture +//! +//! The WebSocket implementation is organized into several modules: +//! +//! - [`types`] - Configuration structures and enums (`BinanceWsConfig`, `DepthLevel`, etc.) +//! - [`parsers`] - Stream message parsers implementing [`StreamParser`] trait +//! - [`connection_manager`] - Connection sharding and management +//! - `handlers` - Message routing and dispatch +//! - `subscriptions` - Subscription tracking and management +//! +//! # Usage +//! +//! ```rust,ignore +//! use ccxt_exchanges::binance::ws::{BinanceWs, BinanceWsConfig, DepthLevel, UpdateSpeed}; +//! +//! // Create a WebSocket client +//! let ws = BinanceWs::new("wss://stream.binance.com:9443/ws".to_string()); +//! +//! // Subscribe to ticker updates +//! let ticker = ws.watch_ticker("BTC/USDT", MarketType::Spot).await?; +//! +//! // Check connection health +//! let health = ws.health(); +//! println!("Latency: {:?}ms", health.latency_ms); +//! ``` +//! +//! # Health Monitoring +//! +//! Use the [`BinanceWs::health`] method to get connection metrics: +//! - Latency (from ping/pong) +//! - Message counts (received/dropped) +//! - Connection uptime +//! - Reconnection count +//! +//! # Error Handling +//! +//! Errors are classified into recovery strategies via [`WsErrorRecovery`]: +//! - `Retry` - Transient errors, retry with backoff +//! - `Resync` - State out of sync, need to refresh (e.g., orderbook) +//! - `Reconnect` - Connection lost, need to reconnect +//! - `Fatal` - Unrecoverable, requires user intervention /// Connection manager module pub mod connection_manager; mod handlers; mod listen_key; +/// Stream parsers for WebSocket messages +pub mod parsers; mod streams; mod subscriptions; +/// Types and configuration for WebSocket +pub mod types; pub(crate) mod user_data; // Re-export public types for backward compatibility pub use connection_manager::BinanceConnectionManager; pub use handlers::MessageRouter; pub use listen_key::ListenKeyManager; -pub use streams::*; +pub use parsers::{ + BidAskParser, MarkPriceParser, OhlcvParser, StreamParser, TickerParser, TradeParser, +}; +pub use streams::normalize_symbol; pub use subscriptions::{ReconnectConfig, Subscription, SubscriptionManager, SubscriptionType}; +pub use types::{ + BinanceWsConfig, DepthLevel, UpdateSpeed, WsChannelConfig, WsErrorRecovery, WsHealthSnapshot, +}; use crate::binance::{Binance, parser}; use ccxt_core::error::{Error, Result}; @@ -25,11 +77,14 @@ use ccxt_core::types::{ use serde_json::Value; use std::collections::{HashMap, VecDeque}; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::time::Duration; use tokio::sync::{Mutex, RwLock}; const MAX_TRADES: usize = 1000; const MAX_OHLCVS: usize = 1000; +/// Default shutdown timeout in milliseconds +const DEFAULT_SHUTDOWN_TIMEOUT_MS: u64 = 5000; /// Binance WebSocket client wrapper pub struct BinanceWs { @@ -48,6 +103,16 @@ pub struct BinanceWs { pub(crate) orders: Arc>>>, pub(crate) my_trades: Arc>>>, pub(crate) positions: Arc>>>, + /// Channel capacity configuration + channel_config: WsChannelConfig, + /// Statistics for health monitoring + messages_received: Arc, + messages_dropped: Arc, + last_message_time: Arc, + connection_start_time: Arc, + /// Shutdown state tracking + is_shutting_down: Arc, + shutdown_complete: Arc, } impl std::fmt::Debug for BinanceWs { @@ -58,6 +123,21 @@ impl std::fmt::Debug for BinanceWs { } } +impl Drop for BinanceWs { + fn drop(&mut self) { + // Warn if shutdown() was not called before dropping + if !self.shutdown_complete.load(Ordering::Acquire) + && !self.is_shutting_down.load(Ordering::Acquire) + { + tracing::warn!( + "BinanceWs dropped without calling shutdown(). \ + This may leave resources uncleaned. \ + Consider calling shutdown() before dropping." + ); + } + } +} + impl BinanceWs { /// Creates a new Binance WebSocket client pub fn new(url: String) -> Self { @@ -87,6 +167,72 @@ impl BinanceWs { orders: Arc::new(RwLock::new(HashMap::new())), my_trades: Arc::new(RwLock::new(HashMap::new())), positions: Arc::new(RwLock::new(HashMap::new())), + channel_config: WsChannelConfig::default(), + messages_received: Arc::new(AtomicU64::new(0)), + messages_dropped: Arc::new(AtomicU64::new(0)), + last_message_time: Arc::new(AtomicU64::new(0)), + connection_start_time: Arc::new(AtomicU64::new( + chrono::Utc::now().timestamp_millis() as u64 + )), + is_shutting_down: Arc::new(AtomicBool::new(false)), + shutdown_complete: Arc::new(AtomicBool::new(false)), + } + } + + /// Creates a new Binance WebSocket client with custom configuration. + /// + /// # Arguments + /// * `config` - Configuration options including URL, channel capacities, and backpressure strategy + /// + /// # Example + /// ```rust,ignore + /// use ccxt_exchanges::binance::ws::{BinanceWs, BinanceWsConfig, WsChannelConfig}; + /// use ccxt_core::ws_client::BackpressureStrategy; + /// + /// let config = BinanceWsConfig::new("wss://stream.binance.com:9443/ws".to_string()) + /// .with_backpressure(BackpressureStrategy::DropOldest); + /// let ws = BinanceWs::new_with_config(config); + /// ``` + pub fn new_with_config(config: BinanceWsConfig) -> Self { + let subscription_manager = Arc::new(SubscriptionManager::new()); + let message_router = Arc::new(MessageRouter::new( + config.url, + subscription_manager.clone(), + None, + )); + + // Start the router immediately + let router_clone = message_router.clone(); + tokio::spawn(async move { + if let Err(e) = router_clone.start(None).await { + tracing::error!("Failed to start MessageRouter: {}", e); + } + }); + + Self { + message_router, + subscription_manager, + listen_key: Arc::new(RwLock::new(None)), + listen_key_manager: None, + tickers: Arc::new(Mutex::new(HashMap::new())), + bids_asks: Arc::new(Mutex::new(HashMap::new())), + mark_prices: Arc::new(Mutex::new(HashMap::new())), + orderbooks: Arc::new(Mutex::new(HashMap::new())), + trades: Arc::new(Mutex::new(HashMap::new())), + ohlcvs: Arc::new(Mutex::new(HashMap::new())), + balances: Arc::new(RwLock::new(HashMap::new())), + orders: Arc::new(RwLock::new(HashMap::new())), + my_trades: Arc::new(RwLock::new(HashMap::new())), + positions: Arc::new(RwLock::new(HashMap::new())), + channel_config: config.channel_config, + messages_received: Arc::new(AtomicU64::new(0)), + messages_dropped: Arc::new(AtomicU64::new(0)), + last_message_time: Arc::new(AtomicU64::new(0)), + connection_start_time: Arc::new(AtomicU64::new( + chrono::Utc::now().timestamp_millis() as u64 + )), + is_shutting_down: Arc::new(AtomicBool::new(false)), + shutdown_complete: Arc::new(AtomicBool::new(false)), } } @@ -123,6 +269,32 @@ impl BinanceWs { orders: Arc::new(RwLock::new(HashMap::new())), my_trades: Arc::new(RwLock::new(HashMap::new())), positions: Arc::new(RwLock::new(HashMap::new())), + channel_config: WsChannelConfig::default(), + messages_received: Arc::new(AtomicU64::new(0)), + messages_dropped: Arc::new(AtomicU64::new(0)), + last_message_time: Arc::new(AtomicU64::new(0)), + connection_start_time: Arc::new(AtomicU64::new( + chrono::Utc::now().timestamp_millis() as u64 + )), + is_shutting_down: Arc::new(AtomicBool::new(false)), + shutdown_complete: Arc::new(AtomicBool::new(false)), + } + } + + /// Returns the channel capacity for a given subscription type + fn channel_capacity_for(&self, sub_type: &SubscriptionType) -> usize { + match sub_type { + SubscriptionType::Ticker | SubscriptionType::BookTicker => { + self.channel_config.ticker_capacity + } + SubscriptionType::OrderBook => self.channel_config.orderbook_capacity, + SubscriptionType::Trades | SubscriptionType::Kline(_) | SubscriptionType::MarkPrice => { + self.channel_config.trades_capacity + } + SubscriptionType::Balance + | SubscriptionType::Orders + | SubscriptionType::MyTrades + | SubscriptionType::Positions => self.channel_config.user_data_capacity, } } @@ -151,6 +323,88 @@ impl BinanceWs { Ok(()) } + /// Gracefully shuts down the WebSocket client. + /// + /// This method performs a complete shutdown sequence: + /// 1. Sets the shutting down flag to reject new subscriptions + /// 2. Stops the message router + /// 3. Stops listen key auto-refresh + /// 4. Clears all subscriptions + /// 5. Clears cached data + /// + /// After calling this method, the client cannot be reused. + /// This method is idempotent - calling it multiple times is safe. + pub async fn shutdown(&self) -> Result<()> { + // Check if already shutting down or complete + if self.shutdown_complete.load(Ordering::Acquire) { + return Ok(()); + } + + // Set shutting down flag to reject new subscriptions + if self.is_shutting_down.swap(true, Ordering::AcqRel) { + // Another shutdown is in progress, wait for it + while !self.shutdown_complete.load(Ordering::Acquire) { + tokio::time::sleep(Duration::from_millis(10)).await; + } + return Ok(()); + } + + tracing::info!("Initiating graceful shutdown of BinanceWs"); + + // Stop the message router (sends close frame) + let shutdown_result = tokio::time::timeout( + Duration::from_millis(DEFAULT_SHUTDOWN_TIMEOUT_MS), + self.message_router.stop(), + ) + .await; + + if shutdown_result.is_err() { + tracing::warn!("Shutdown timeout exceeded, forcing close"); + } + + // Stop listen key auto-refresh + if let Some(manager) = &self.listen_key_manager { + manager.stop_auto_refresh().await; + } + + // Clear all subscriptions + self.subscription_manager.clear().await; + + // Clear cached data + self.tickers.lock().await.clear(); + self.bids_asks.lock().await.clear(); + self.mark_prices.lock().await.clear(); + self.orderbooks.lock().await.clear(); + self.trades.lock().await.clear(); + self.ohlcvs.lock().await.clear(); + self.balances.write().await.clear(); + self.orders.write().await.clear(); + self.my_trades.write().await.clear(); + self.positions.write().await.clear(); + + // Mark shutdown as complete + self.shutdown_complete.store(true, Ordering::Release); + + tracing::info!("BinanceWs shutdown complete"); + Ok(()) + } + + /// Returns true if the client is shutting down or has shut down. + #[inline] + pub fn is_shutting_down(&self) -> bool { + self.is_shutting_down.load(Ordering::Acquire) + } + + /// Checks if shutdown is allowed and returns an error if shutting down. + #[inline] + #[allow(dead_code)] + fn check_not_shutting_down(&self) -> Result<()> { + if self.is_shutting_down.load(Ordering::Acquire) { + return Err(Error::invalid_request("WebSocket client is shutting down")); + } + Ok(()) + } + /// Connects to the user data stream pub async fn connect_user_stream(&self) -> Result<()> { let manager = self.listen_key_manager.as_ref() @@ -194,10 +448,18 @@ impl BinanceWs { } } - /// Subscribes to the ticker stream for a symbol - pub async fn subscribe_ticker(&self, symbol: &str) -> Result<()> { - let stream = format!("{}@ticker", symbol.to_lowercase()); - let (tx, _rx) = tokio::sync::mpsc::channel(1024); + /// Subscribes to the ticker stream for a symbol. + /// + /// Returns a receiver that will receive ticker updates as JSON values. + /// The caller is responsible for consuming messages from the receiver. + pub async fn subscribe_ticker( + &self, + symbol: &str, + ) -> Result> { + let normalized = normalize_symbol(symbol); + let stream = format!("{}@ticker", normalized); + let capacity = self.channel_capacity_for(&SubscriptionType::Ticker); + let (tx, rx) = tokio::sync::mpsc::channel(capacity); self.subscription_manager .add_subscription( @@ -208,13 +470,17 @@ impl BinanceWs { ) .await?; - self.message_router.subscribe(vec![stream]).await + self.message_router.subscribe(vec![stream]).await?; + Ok(rx) } - /// Subscribes to the 24-hour ticker stream for all symbols - pub async fn subscribe_all_tickers(&self) -> Result<()> { + /// Subscribes to the 24-hour ticker stream for all symbols. + /// + /// Returns a receiver that will receive ticker updates as JSON values. + pub async fn subscribe_all_tickers(&self) -> Result> { let stream = "!ticker@arr".to_string(); - let (tx, _rx) = tokio::sync::mpsc::channel(1024); + let capacity = self.channel_capacity_for(&SubscriptionType::Ticker); + let (tx, rx) = tokio::sync::mpsc::channel(capacity); self.subscription_manager .add_subscription( @@ -225,13 +491,21 @@ impl BinanceWs { ) .await?; - self.message_router.subscribe(vec![stream]).await + self.message_router.subscribe(vec![stream]).await?; + Ok(rx) } - /// Subscribes to real-time trade executions for a symbol - pub async fn subscribe_trades(&self, symbol: &str) -> Result<()> { - let stream = format!("{}@trade", symbol.to_lowercase()); - let (tx, _rx) = tokio::sync::mpsc::channel(1024); + /// Subscribes to real-time trade executions for a symbol. + /// + /// Returns a receiver that will receive trade updates as JSON values. + pub async fn subscribe_trades( + &self, + symbol: &str, + ) -> Result> { + let normalized = normalize_symbol(symbol); + let stream = format!("{}@trade", normalized); + let capacity = self.channel_capacity_for(&SubscriptionType::Trades); + let (tx, rx) = tokio::sync::mpsc::channel(capacity); self.subscription_manager .add_subscription( @@ -242,13 +516,21 @@ impl BinanceWs { ) .await?; - self.message_router.subscribe(vec![stream]).await + self.message_router.subscribe(vec![stream]).await?; + Ok(rx) } - /// Subscribes to the aggregated trade stream for a symbol - pub async fn subscribe_agg_trades(&self, symbol: &str) -> Result<()> { - let stream = format!("{}@aggTrade", symbol.to_lowercase()); - let (tx, _rx) = tokio::sync::mpsc::channel(1024); + /// Subscribes to the aggregated trade stream for a symbol. + /// + /// Returns a receiver that will receive aggregated trade updates as JSON values. + pub async fn subscribe_agg_trades( + &self, + symbol: &str, + ) -> Result> { + let normalized = normalize_symbol(symbol); + let stream = format!("{}@aggTrade", normalized); + let capacity = self.channel_capacity_for(&SubscriptionType::Trades); + let (tx, rx) = tokio::sync::mpsc::channel(capacity); self.subscription_manager .add_subscription( @@ -259,22 +541,40 @@ impl BinanceWs { ) .await?; - self.message_router.subscribe(vec![stream]).await + self.message_router.subscribe(vec![stream]).await?; + Ok(rx) } - /// Subscribes to the order book depth stream + /// Subscribes to the order book depth stream. + /// + /// Returns a receiver that will receive order book updates as JSON values. + /// + /// # Arguments + /// * `symbol` - Trading pair symbol (e.g., "BTCUSDT" or "BTC/USDT") + /// * `levels` - Depth level (L5, L10, or L20) + /// * `update_speed` - Update frequency (Ms100 or Ms1000) + /// + /// # Example + /// ```rust,ignore + /// use ccxt_exchanges::binance::ws::{DepthLevel, UpdateSpeed}; + /// let mut rx = ws.subscribe_orderbook("BTCUSDT", DepthLevel::L20, UpdateSpeed::Ms100).await?; + /// while let Some(msg) = rx.recv().await { + /// println!("Order book update: {:?}", msg); + /// } + /// ``` pub async fn subscribe_orderbook( &self, symbol: &str, - levels: u32, - update_speed: &str, - ) -> Result<()> { - let stream = if update_speed == "100ms" { - format!("{}@depth{}@100ms", symbol.to_lowercase(), levels) - } else { - format!("{}@depth{}", symbol.to_lowercase(), levels) + levels: DepthLevel, + update_speed: UpdateSpeed, + ) -> Result> { + let normalized = normalize_symbol(symbol); + let stream = match update_speed { + UpdateSpeed::Ms100 => format!("{}@depth{}@100ms", normalized, levels.as_u32()), + UpdateSpeed::Ms1000 => format!("{}@depth{}", normalized, levels.as_u32()), }; - let (tx, _rx) = tokio::sync::mpsc::channel(1024); + let capacity = self.channel_capacity_for(&SubscriptionType::OrderBook); + let (tx, rx) = tokio::sync::mpsc::channel(capacity); self.subscription_manager .add_subscription( @@ -285,25 +585,29 @@ impl BinanceWs { ) .await?; - self.message_router.subscribe(vec![stream]).await + self.message_router.subscribe(vec![stream]).await?; + Ok(rx) } - /// Subscribes to the diff order book stream + /// Subscribes to the diff order book stream. + /// + /// Returns a receiver that will receive order book diff updates as JSON values. + /// + /// # Arguments + /// * `symbol` - Trading pair symbol + /// * `update_speed` - Optional update frequency pub async fn subscribe_orderbook_diff( &self, symbol: &str, - update_speed: Option<&str>, - ) -> Result<()> { - let stream = if let Some(speed) = update_speed { - if speed == "100ms" { - format!("{}@depth@100ms", symbol.to_lowercase()) - } else { - format!("{}@depth", symbol.to_lowercase()) - } - } else { - format!("{}@depth", symbol.to_lowercase()) + update_speed: Option, + ) -> Result> { + let normalized = normalize_symbol(symbol); + let stream = match update_speed { + Some(UpdateSpeed::Ms100) => format!("{}@depth@100ms", normalized), + _ => format!("{}@depth", normalized), }; - let (tx, _rx) = tokio::sync::mpsc::channel(1024); + let capacity = self.channel_capacity_for(&SubscriptionType::OrderBook); + let (tx, rx) = tokio::sync::mpsc::channel(capacity); self.subscription_manager .add_subscription( @@ -314,30 +618,43 @@ impl BinanceWs { ) .await?; - self.message_router.subscribe(vec![stream]).await + self.message_router.subscribe(vec![stream]).await?; + Ok(rx) } - /// Subscribes to Kline (candlestick) data for a symbol - pub async fn subscribe_kline(&self, symbol: &str, interval: &str) -> Result<()> { - let stream = format!("{}@kline_{}", symbol.to_lowercase(), interval); - let (tx, _rx) = tokio::sync::mpsc::channel(1024); + /// Subscribes to Kline (candlestick) data for a symbol. + /// + /// Returns a receiver that will receive kline updates as JSON values. + pub async fn subscribe_kline( + &self, + symbol: &str, + interval: &str, + ) -> Result> { + let normalized = normalize_symbol(symbol); + let stream = format!("{}@kline_{}", normalized, interval); + let sub_type = SubscriptionType::Kline(interval.to_string()); + let capacity = self.channel_capacity_for(&sub_type); + let (tx, rx) = tokio::sync::mpsc::channel(capacity); self.subscription_manager - .add_subscription( - stream.clone(), - symbol.to_string(), - SubscriptionType::Kline(interval.to_string()), - tx, - ) + .add_subscription(stream.clone(), symbol.to_string(), sub_type, tx) .await?; - self.message_router.subscribe(vec![stream]).await + self.message_router.subscribe(vec![stream]).await?; + Ok(rx) } - /// Subscribes to the mini ticker stream for a symbol - pub async fn subscribe_mini_ticker(&self, symbol: &str) -> Result<()> { - let stream = format!("{}@miniTicker", symbol.to_lowercase()); - let (tx, _rx) = tokio::sync::mpsc::channel(1024); + /// Subscribes to the mini ticker stream for a symbol. + /// + /// Returns a receiver that will receive mini ticker updates as JSON values. + pub async fn subscribe_mini_ticker( + &self, + symbol: &str, + ) -> Result> { + let normalized = normalize_symbol(symbol); + let stream = format!("{}@miniTicker", normalized); + let capacity = self.channel_capacity_for(&SubscriptionType::Ticker); + let (tx, rx) = tokio::sync::mpsc::channel(capacity); self.subscription_manager .add_subscription( @@ -348,13 +665,17 @@ impl BinanceWs { ) .await?; - self.message_router.subscribe(vec![stream]).await + self.message_router.subscribe(vec![stream]).await?; + Ok(rx) } - /// Subscribes to the mini ticker stream for all symbols - pub async fn subscribe_all_mini_tickers(&self) -> Result<()> { + /// Subscribes to the mini ticker stream for all symbols. + /// + /// Returns a receiver that will receive mini ticker updates as JSON values. + pub async fn subscribe_all_mini_tickers(&self) -> Result> { let stream = "!miniTicker@arr".to_string(); - let (tx, _rx) = tokio::sync::mpsc::channel(1024); + let capacity = self.channel_capacity_for(&SubscriptionType::Ticker); + let (tx, rx) = tokio::sync::mpsc::channel(capacity); self.subscription_manager .add_subscription( @@ -365,7 +686,8 @@ impl BinanceWs { ) .await?; - self.message_router.subscribe(vec![stream]).await + self.message_router.subscribe(vec![stream]).await?; + Ok(rx) } /// Cancels an existing subscription @@ -410,69 +732,156 @@ impl BinanceWs { self.subscription_manager.active_count() } - /// Watches a single mark price stream (internal helper) - async fn watch_mark_price_internal( - &self, - symbol: &str, - channel_name: &str, - ) -> Result { - let stream = format!("{}@{}", symbol.to_lowercase(), channel_name); - tracing::debug!( - "watch_mark_price_internal: stream={}, symbol={}", - stream, - symbol + /// Returns a health snapshot for monitoring the WebSocket connection. + /// + /// This provides metrics useful for monitoring connection health: + /// - Latency from ping/pong + /// - Message counts (received and dropped) + /// - Connection uptime + /// - Reconnection count + pub fn health(&self) -> WsHealthSnapshot { + let now = chrono::Utc::now().timestamp_millis() as u64; + let start_time = self.connection_start_time.load(Ordering::Relaxed); + let last_msg = self.last_message_time.load(Ordering::Relaxed); + + WsHealthSnapshot { + latency_ms: self.message_router.latency(), + messages_received: self.messages_received.load(Ordering::Relaxed), + messages_dropped: self.messages_dropped.load(Ordering::Relaxed), + last_message_time: if last_msg > 0 { + Some(last_msg as i64) + } else { + None + }, + connection_uptime_ms: if start_time > 0 { + now.saturating_sub(start_time) + } else { + 0 + }, + reconnect_count: self.message_router.reconnect_count(), + } + } + + /// Records that a message was received (for health tracking). + #[inline] + #[allow(dead_code)] + pub(crate) fn record_message_received(&self) { + self.messages_received.fetch_add(1, Ordering::Relaxed); + self.last_message_time.store( + chrono::Utc::now().timestamp_millis() as u64, + Ordering::Relaxed, ); + } - let (tx, mut rx) = tokio::sync::mpsc::channel(1024); + /// Records that a message was dropped (for health tracking). + #[inline] + #[allow(dead_code)] + pub(crate) fn record_message_dropped(&self) { + let count = self.messages_dropped.fetch_add(1, Ordering::Relaxed) + 1; + // Log every 100th drop to avoid log spam + if count % 100 == 1 { + tracing::warn!(dropped_count = count, "Message dropped due to backpressure"); + } + } + + /// Generic method for watching a WebSocket stream. + /// + /// This method abstracts the common pattern used by all watch_* methods: + /// 1. Create a channel for receiving messages + /// 2. Add subscription to the manager + /// 3. Subscribe via the message router + /// 4. Wait for and parse messages + /// + /// # Type Parameters + /// * `T` - The output type to parse messages into + /// * `P` - The parser type implementing `StreamParser` + /// + /// # Arguments + /// * `stream` - The stream name (e.g., "btcusdt@ticker") + /// * `symbol` - The symbol for subscription tracking + /// * `sub_type` - The subscription type + /// * `market` - Optional market info for parsing + async fn watch_stream( + &self, + stream: String, + symbol: String, + sub_type: SubscriptionType, + market: Option<&ccxt_core::types::Market>, + ) -> Result + where + P: parsers::StreamParser, + { + let capacity = self.channel_capacity_for(&sub_type); + let (tx, mut rx) = tokio::sync::mpsc::channel(capacity); self.subscription_manager - .add_subscription( - stream.clone(), - symbol.to_string(), - SubscriptionType::MarkPrice, - tx, - ) + .add_subscription(stream.clone(), symbol, sub_type, tx) .await?; self.message_router.subscribe(vec![stream.clone()]).await?; loop { if let Some(message) = rx.recv().await { + // Skip subscription confirmation messages if message.get("result").is_some() { - tracing::debug!("Received subscription result for {}", stream); continue; } - tracing::debug!("Received mark price message for {}", stream); - - match parser::parse_ws_mark_price(&message) { - Ok(mark_price) => { - let mut mark_prices = self.mark_prices.lock().await; - mark_prices.insert(mark_price.symbol.clone(), mark_price.clone()); - return Ok(mark_price); - } + match P::parse(&message, market) { + Ok(data) => return Ok(data), Err(e) => { tracing::warn!( - "Failed to parse mark price message for stream {}: {:?}. Payload: {:?}", + "Failed to parse message for stream {}: {:?}. Payload: {:?}", stream, e, message ); + // Continue waiting for the next message } } } else { - tracing::warn!("Subscription channel closed for {}", stream); return Err(Error::network("Subscription channel closed")); } } } + /// Watches a single mark price stream (internal helper) + async fn watch_mark_price_internal( + &self, + symbol: &str, + channel_name: &str, + ) -> Result { + let normalized = normalize_symbol(symbol); + let stream = format!("{}@{}", normalized, channel_name); + tracing::debug!( + "watch_mark_price_internal: stream={}, symbol={}", + stream, + symbol + ); + + let mark_price = self + .watch_stream::( + stream, + symbol.to_string(), + SubscriptionType::MarkPrice, + None, + ) + .await?; + + // Update cache + let mut mark_prices = self.mark_prices.lock().await; + mark_prices.insert(mark_price.symbol.clone(), mark_price.clone()); + + Ok(mark_price) + } + /// Watches multiple mark price streams (internal helper) async fn watch_mark_prices_internal( &self, symbols: Option>, channel_name: &str, ) -> Result> { - let (tx, mut rx) = tokio::sync::mpsc::channel(1024); + let capacity = self.channel_capacity_for(&SubscriptionType::MarkPrice); + let (tx, mut rx) = tokio::sync::mpsc::channel(capacity); let streams: Vec = if let Some(syms) = symbols.as_ref() { let mut streams = Vec::with_capacity(syms.len()); @@ -573,46 +982,23 @@ impl BinanceWs { /// Watches a single ticker stream (internal helper) async fn watch_ticker_internal(&self, symbol: &str, channel_name: &str) -> Result { - let stream = format!("{}@{}", symbol.to_lowercase(), channel_name); + let normalized = normalize_symbol(symbol); + let stream = format!("{}@{}", normalized, channel_name); - let (tx, mut rx) = tokio::sync::mpsc::channel(1024); - self.subscription_manager - .add_subscription( - stream.clone(), + let ticker = self + .watch_stream::( + stream, symbol.to_string(), SubscriptionType::Ticker, - tx, + None, ) .await?; - self.message_router.subscribe(vec![stream.clone()]).await?; + // Update cache + let mut tickers = self.tickers.lock().await; + tickers.insert(ticker.symbol.clone(), ticker.clone()); - loop { - if let Some(message) = rx.recv().await { - if message.get("result").is_some() { - continue; - } - - match parser::parse_ws_ticker(&message, None) { - Ok(ticker) => { - let mut tickers = self.tickers.lock().await; - tickers.insert(ticker.symbol.clone(), ticker.clone()); - return Ok(ticker); - } - Err(e) => { - tracing::warn!( - "Failed to parse ticker message for stream {}: {:?}. Payload: {:?}", - stream, - e, - message - ); - // Continue waiting for the next message instead of hanging silently - } - } - } else { - return Err(Error::network("Subscription channel closed")); - } - } + Ok(ticker) } /// Watches multiple ticker streams (internal helper) @@ -629,7 +1015,8 @@ impl BinanceWs { vec![format!("!{}@arr", channel_name)] }; - let (tx, mut rx) = tokio::sync::mpsc::channel(1024); + let capacity = self.channel_capacity_for(&SubscriptionType::Ticker); + let (tx, mut rx) = tokio::sync::mpsc::channel(capacity); for stream in &streams { self.subscription_manager @@ -740,16 +1127,16 @@ impl BinanceWs { exchange: &Binance, symbol: &str, limit: Option, - update_speed: i32, + update_speed: UpdateSpeed, is_futures: bool, ) -> Result { - let stream = if update_speed == 100 { - format!("{}@depth@100ms", symbol.to_lowercase()) - } else { - format!("{}@depth", symbol.to_lowercase()) + let stream = match update_speed { + UpdateSpeed::Ms100 => format!("{}@depth@100ms", symbol.to_lowercase()), + UpdateSpeed::Ms1000 => format!("{}@depth", symbol.to_lowercase()), }; - let (tx, mut rx) = tokio::sync::mpsc::channel(1024); + let capacity = self.channel_capacity_for(&SubscriptionType::OrderBook); + let (tx, mut rx) = tokio::sync::mpsc::channel(capacity); self.subscription_manager .add_subscription( stream.clone(), @@ -789,45 +1176,26 @@ impl BinanceWs { } Err(e) => { let err_msg = e.to_string(); + let recovery = WsErrorRecovery::from_error_message(&err_msg); - if err_msg.contains("RESYNC_NEEDED") { - tracing::warn!("Resync needed for {}: {}", symbol, err_msg); - - let current_time = chrono::Utc::now().timestamp_millis(); - let should_resync = { - let orderbooks = self.orderbooks.lock().await; - if let Some(ob) = orderbooks.get(symbol) { - ob.should_resync(current_time) - } else { - true - } - }; - - if should_resync { - tracing::info!("Initiating resync for {}", symbol); - - { - let mut orderbooks = self.orderbooks.lock().await; - if let Some(ob) = orderbooks.get_mut(symbol) { - ob.reset_for_resync(); - ob.mark_resync_initiated(current_time); - } - } - - tokio::time::sleep(Duration::from_millis(500)).await; - + match recovery { + WsErrorRecovery::Resync => { + tracing::warn!("Resync needed for {}: {}", symbol, err_msg); match self - .fetch_orderbook_snapshot( - exchange, symbol, limit, is_futures, - ) + .resync_orderbook(exchange, symbol, limit, is_futures) .await { - Ok(_) => { + Ok(true) => { tracing::info!( "Resync completed successfully for {}", symbol ); - continue; + } + Ok(false) => { + tracing::debug!( + "Resync rate limited for {}, skipping", + symbol + ); } Err(resync_err) => { tracing::error!( @@ -839,9 +1207,20 @@ impl BinanceWs { } } } - tracing::debug!("Resync rate limited for {}, skipping", symbol); + WsErrorRecovery::Fatal => { + tracing::error!( + "Fatal error handling orderbook delta: {}", + err_msg + ); + return Err(e); + } + _ => { + tracing::error!( + "Failed to handle orderbook delta: {}", + err_msg + ); + } } - tracing::error!("Failed to handle orderbook delta: {}", err_msg); } } } @@ -852,13 +1231,58 @@ impl BinanceWs { } } + /// Resyncs the orderbook for a symbol by fetching a fresh snapshot. + /// + /// Returns `Ok(true)` if resync was performed, `Ok(false)` if rate limited. + async fn resync_orderbook( + &self, + exchange: &Binance, + symbol: &str, + limit: Option, + is_futures: bool, + ) -> Result { + let current_time = chrono::Utc::now().timestamp_millis(); + + // Check rate limit + let should_resync = { + let orderbooks = self.orderbooks.lock().await; + if let Some(ob) = orderbooks.get(symbol) { + ob.should_resync(current_time) + } else { + true + } + }; + + if !should_resync { + return Ok(false); + } + + // Reset orderbook state + { + let mut orderbooks = self.orderbooks.lock().await; + if let Some(ob) = orderbooks.get_mut(symbol) { + ob.reset_for_resync(); + ob.mark_resync_initiated(current_time); + } + } + + // Wait before fetching snapshot + tokio::time::sleep(Duration::from_millis(500)).await; + + // Fetch fresh snapshot + self.fetch_orderbook_snapshot(exchange, symbol, limit, is_futures) + .await?; + + Ok(true) + } + /// Watches multiple order book streams (internal helper) async fn watch_orderbooks_internal( &self, exchange: &Binance, symbols: Vec, limit: Option, - update_speed: i32, + update_speed: UpdateSpeed, is_futures: bool, ) -> Result> { if symbols.len() > 200 { @@ -867,14 +1291,14 @@ impl BinanceWs { )); } - let (tx, mut rx) = tokio::sync::mpsc::channel(1024); + let capacity = self.channel_capacity_for(&SubscriptionType::OrderBook); + let (tx, mut rx) = tokio::sync::mpsc::channel(capacity); let mut streams = Vec::new(); for symbol in &symbols { - let stream = if update_speed == 100 { - format!("{}@depth@100ms", symbol.to_lowercase()) - } else { - format!("{}@depth", symbol.to_lowercase()) + let stream = match update_speed { + UpdateSpeed::Ms100 => format!("{}@depth@100ms", symbol.to_lowercase()), + UpdateSpeed::Ms1000 => format!("{}@depth", symbol.to_lowercase()), }; streams.push(stream.clone()); @@ -947,35 +1371,23 @@ impl BinanceWs { /// Watches the best bid/ask data for a unified symbol (internal helper) async fn watch_bids_asks_internal(&self, symbol: &str, market_id: &str) -> Result { - let stream = format!("{}@bookTicker", market_id.to_lowercase()); - let (tx, mut rx) = tokio::sync::mpsc::channel(1024); + let normalized = normalize_symbol(market_id); + let stream = format!("{}@bookTicker", normalized); - self.subscription_manager - .add_subscription( - stream.clone(), + let bid_ask = self + .watch_stream::( + stream, symbol.to_string(), SubscriptionType::BookTicker, - tx, + None, ) .await?; - self.message_router.subscribe(vec![stream.clone()]).await?; + // Update cache + let mut bids_asks_map = self.bids_asks.lock().await; + bids_asks_map.insert(symbol.to_string(), bid_ask.clone()); - loop { - if let Some(message) = rx.recv().await { - if message.get("result").is_some() { - continue; - } - - if let Ok(bid_ask) = parser::parse_ws_bid_ask(&message) { - let mut bids_asks_map = self.bids_asks.lock().await; - bids_asks_map.insert(symbol.to_string(), bid_ask.clone()); - return Ok(bid_ask); - } - } else { - return Err(Error::network("Subscription channel closed")); - } - } + Ok(bid_ask) } /// Streams trade data for a unified symbol (internal helper) @@ -988,7 +1400,8 @@ impl BinanceWs { market: Option<&ccxt_core::types::Market>, ) -> Result> { let stream = format!("{}@trade", market_id.to_lowercase()); - let (tx, mut rx) = tokio::sync::mpsc::channel(1024); + let capacity = self.channel_capacity_for(&SubscriptionType::Trades); + let (tx, mut rx) = tokio::sync::mpsc::channel(capacity); self.subscription_manager .add_subscription( @@ -1051,15 +1464,12 @@ impl BinanceWs { limit: Option, ) -> Result> { let stream = format!("{}@kline_{}", market_id.to_lowercase(), timeframe); - let (tx, mut rx) = tokio::sync::mpsc::channel(1024); + let sub_type = SubscriptionType::Kline(timeframe.to_string()); + let capacity = self.channel_capacity_for(&sub_type); + let (tx, mut rx) = tokio::sync::mpsc::channel(capacity); self.subscription_manager - .add_subscription( - stream.clone(), - symbol.to_string(), - SubscriptionType::Kline(timeframe.to_string()), - tx, - ) + .add_subscription(stream.clone(), symbol.to_string(), sub_type, tx) .await?; self.message_router.subscribe(vec![stream.clone()]).await?; @@ -1121,7 +1531,8 @@ impl BinanceWs { async fn watch_balance_internal(&self, account_type: &str) -> Result { self.connect_user_stream().await?; - let (tx, mut rx) = tokio::sync::mpsc::channel(1024); + let capacity = self.channel_capacity_for(&SubscriptionType::Balance); + let (tx, mut rx) = tokio::sync::mpsc::channel(capacity); self.subscription_manager .add_subscription( @@ -1162,7 +1573,8 @@ impl BinanceWs { ) -> Result> { self.connect_user_stream().await?; - let (tx, mut rx) = tokio::sync::mpsc::channel(1024); + let capacity = self.channel_capacity_for(&SubscriptionType::Orders); + let (tx, mut rx) = tokio::sync::mpsc::channel(capacity); self.subscription_manager .add_subscription( @@ -1226,7 +1638,8 @@ impl BinanceWs { ) -> Result> { self.connect_user_stream().await?; - let (tx, mut rx) = tokio::sync::mpsc::channel(1024); + let capacity = self.channel_capacity_for(&SubscriptionType::MyTrades); + let (tx, mut rx) = tokio::sync::mpsc::channel(capacity); self.subscription_manager .add_subscription( @@ -1273,7 +1686,8 @@ impl BinanceWs { ) -> Result> { self.connect_user_stream().await?; - let (tx, mut rx) = tokio::sync::mpsc::channel(1024); + let capacity = self.channel_capacity_for(&SubscriptionType::Positions); + let (tx, mut rx) = tokio::sync::mpsc::channel(capacity); self.subscription_manager .add_subscription( @@ -1462,6 +1876,11 @@ include!("binance_impl.rs"); #[allow(clippy::disallowed_methods)] mod tests { use super::*; + use streams::WS_BASE_URL; + use types::{ + DEFAULT_ORDERBOOK_CAPACITY, DEFAULT_TICKER_CAPACITY, DEFAULT_TRADES_CAPACITY, + DEFAULT_USER_DATA_CAPACITY, + }; #[tokio::test] async fn test_binance_ws_creation() { @@ -1525,4 +1944,105 @@ mod tests { let binance_symbol = symbol.replace('/', "").to_lowercase(); assert_eq!(binance_symbol, "btcusdt"); } + + #[test] + fn test_ws_health_snapshot_default() { + let health = WsHealthSnapshot::default(); + assert_eq!(health.messages_received, 0); + assert_eq!(health.messages_dropped, 0); + assert!(health.latency_ms.is_none()); + assert!(health.last_message_time.is_none()); + assert_eq!(health.connection_uptime_ms, 0); + assert_eq!(health.reconnect_count, 0); + } + + #[test] + fn test_ws_health_snapshot_is_healthy() { + let mut health = WsHealthSnapshot::default(); + + // Empty snapshot is healthy (no data yet) + assert!(health.is_healthy()); + + // High drop rate is unhealthy + health.messages_received = 100; + health.messages_dropped = 20; // 20% drop rate + assert!(!health.is_healthy()); + + // Low drop rate is healthy + health.messages_dropped = 5; // 5% drop rate + assert!(health.is_healthy()); + } + + #[tokio::test] + async fn test_shutdown_sets_flags() { + let ws = BinanceWs::new("wss://stream.binance.com:9443/ws".to_string()); + + // Initially not shutting down + assert!(!ws.is_shutting_down()); + + // After shutdown, flag should be set + let _ = ws.shutdown().await; + assert!(ws.is_shutting_down()); + } + + #[tokio::test] + async fn test_shutdown_is_idempotent() { + let ws = BinanceWs::new("wss://stream.binance.com:9443/ws".to_string()); + + // Multiple shutdowns should not panic + let result1 = ws.shutdown().await; + let result2 = ws.shutdown().await; + + assert!(result1.is_ok()); + assert!(result2.is_ok()); + } + + #[tokio::test] + async fn test_channel_capacity_configuration() { + // Test default configuration + let ws = BinanceWs::new("wss://stream.binance.com:9443/ws".to_string()); + + assert_eq!( + ws.channel_capacity_for(&SubscriptionType::Ticker), + DEFAULT_TICKER_CAPACITY + ); + assert_eq!( + ws.channel_capacity_for(&SubscriptionType::OrderBook), + DEFAULT_ORDERBOOK_CAPACITY + ); + assert_eq!( + ws.channel_capacity_for(&SubscriptionType::Trades), + DEFAULT_TRADES_CAPACITY + ); + assert_eq!( + ws.channel_capacity_for(&SubscriptionType::Balance), + DEFAULT_USER_DATA_CAPACITY + ); + } + + #[tokio::test] + async fn test_custom_channel_capacity_configuration() { + use ccxt_core::ws_client::BackpressureStrategy; + + let custom_config = WsChannelConfig { + ticker_capacity: 128, + orderbook_capacity: 256, + trades_capacity: 512, + user_data_capacity: 64, + }; + + let config = BinanceWsConfig::new("wss://stream.binance.com:9443/ws".to_string()) + .with_channel_config(custom_config) + .with_backpressure(BackpressureStrategy::DropOldest); + + let ws = BinanceWs::new_with_config(config); + + assert_eq!(ws.channel_capacity_for(&SubscriptionType::Ticker), 128); + assert_eq!(ws.channel_capacity_for(&SubscriptionType::OrderBook), 256); + assert_eq!(ws.channel_capacity_for(&SubscriptionType::Trades), 512); + assert_eq!(ws.channel_capacity_for(&SubscriptionType::Balance), 64); + assert_eq!(ws.channel_capacity_for(&SubscriptionType::Orders), 64); + assert_eq!(ws.channel_capacity_for(&SubscriptionType::MyTrades), 64); + assert_eq!(ws.channel_capacity_for(&SubscriptionType::Positions), 64); + } } diff --git a/ccxt-exchanges/src/binance/ws/parsers.rs b/ccxt-exchanges/src/binance/ws/parsers.rs new file mode 100644 index 0000000..0d57d9e --- /dev/null +++ b/ccxt-exchanges/src/binance/ws/parsers.rs @@ -0,0 +1,126 @@ +//! Stream parsers for WebSocket message parsing +//! +//! This module provides the `StreamParser` trait and implementations for +//! parsing different types of WebSocket messages from Binance. + +use crate::binance::parser; +use ccxt_core::error::Result; +use ccxt_core::types::{BidAsk, MarkPrice, Market, OHLCV, Ticker, Trade}; +use serde_json::Value; + +/// Trait for parsing WebSocket stream messages into typed data. +/// +/// Implementations of this trait handle the conversion of raw JSON messages +/// from Binance WebSocket streams into strongly-typed Rust structures. +pub trait StreamParser { + /// The output type produced by this parser + type Output; + + /// Parse a WebSocket message into the output type. + /// + /// # Arguments + /// * `message` - The raw JSON message from the WebSocket + /// * `market` - Optional market information for symbol resolution + /// + /// # Returns + /// The parsed data or an error if parsing fails + fn parse(message: &Value, market: Option<&Market>) -> Result; +} + +/// Parser for ticker stream messages. +pub struct TickerParser; + +impl StreamParser for TickerParser { + type Output = Ticker; + + fn parse(message: &Value, market: Option<&Market>) -> Result { + parser::parse_ws_ticker(message, market) + } +} + +/// Parser for trade stream messages. +pub struct TradeParser; + +impl StreamParser for TradeParser { + type Output = Trade; + + fn parse(message: &Value, market: Option<&Market>) -> Result { + parser::parse_ws_trade(message, market) + } +} + +/// Parser for OHLCV (kline) stream messages. +pub struct OhlcvParser; + +impl StreamParser for OhlcvParser { + type Output = OHLCV; + + fn parse(message: &Value, _market: Option<&Market>) -> Result { + parser::parse_ws_ohlcv(message) + } +} + +/// Parser for mark price stream messages. +pub struct MarkPriceParser; + +impl StreamParser for MarkPriceParser { + type Output = MarkPrice; + + fn parse(message: &Value, _market: Option<&Market>) -> Result { + parser::parse_ws_mark_price(message) + } +} + +/// Parser for bid/ask (book ticker) stream messages. +pub struct BidAskParser; + +impl StreamParser for BidAskParser { + type Output = BidAsk; + + fn parse(message: &Value, _market: Option<&Market>) -> Result { + parser::parse_ws_bid_ask(message) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn test_ticker_parser() { + let message = json!({ + "e": "24hrTicker", + "s": "BTCUSDT", + "c": "50000.00", + "o": "49000.00", + "h": "51000.00", + "l": "48000.00", + "v": "1000.00", + "q": "50000000.00", + "C": 1234567890000_i64 + }); + + let result = TickerParser::parse(&message, None); + assert!(result.is_ok()); + let ticker = result.unwrap(); + assert_eq!(ticker.symbol, "BTCUSDT"); + } + + #[test] + fn test_bid_ask_parser() { + let message = json!({ + "s": "BTCUSDT", + "b": "50000.00", + "B": "1.5", + "a": "50001.00", + "A": "2.0", + "E": 1234567890000_i64 + }); + + let result = BidAskParser::parse(&message, None); + assert!(result.is_ok()); + let bid_ask = result.unwrap(); + assert_eq!(bid_ask.symbol, "BTCUSDT"); + } +} diff --git a/ccxt-exchanges/src/binance/ws/streams.rs b/ccxt-exchanges/src/binance/ws/streams.rs index c53410b..fb54cff 100644 --- a/ccxt-exchanges/src/binance/ws/streams.rs +++ b/ccxt-exchanges/src/binance/ws/streams.rs @@ -7,6 +7,22 @@ pub const WS_BASE_URL: &str = "wss://stream.binance.com:9443/ws"; /// Binance testnet WebSocket endpoint pub const WS_TESTNET_URL: &str = "wss://testnet.binance.vision/ws"; +/// Normalizes a symbol to Binance's expected format. +/// +/// Converts symbols like "BTC/USDT", "BTC-USDT", "BTC_USDT" to "btcusdt". +/// This function is idempotent - already normalized symbols remain unchanged. +/// +/// # Examples +/// ``` +/// use ccxt_exchanges::binance::ws::normalize_symbol; +/// assert_eq!(normalize_symbol("BTC/USDT"), "btcusdt"); +/// assert_eq!(normalize_symbol("btcusdt"), "btcusdt"); +/// assert_eq!(normalize_symbol("ETH-BTC"), "ethbtc"); +/// ``` +pub fn normalize_symbol(symbol: &str) -> String { + symbol.replace(['/', '-', '_'], "").to_lowercase() +} + /// Builds a ticker stream name pub fn ticker_stream(symbol: &str) -> String { format!("{}@ticker", symbol.to_lowercase()) @@ -82,3 +98,37 @@ pub fn all_mini_tickers_stream() -> String { pub fn user_data_stream_url(listen_key: &str) -> String { format!("wss://stream.binance.com:9443/ws/{}", listen_key) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_normalize_symbol_with_slash() { + assert_eq!(normalize_symbol("BTC/USDT"), "btcusdt"); + assert_eq!(normalize_symbol("ETH/BTC"), "ethbtc"); + } + + #[test] + fn test_normalize_symbol_with_dash() { + assert_eq!(normalize_symbol("BTC-USDT"), "btcusdt"); + assert_eq!(normalize_symbol("ETH-BTC"), "ethbtc"); + } + + #[test] + fn test_normalize_symbol_with_underscore() { + assert_eq!(normalize_symbol("BTC_USDT"), "btcusdt"); + assert_eq!(normalize_symbol("ETH_BTC"), "ethbtc"); + } + + #[test] + fn test_normalize_symbol_already_normalized() { + assert_eq!(normalize_symbol("btcusdt"), "btcusdt"); + assert_eq!(normalize_symbol("BTCUSDT"), "btcusdt"); + } + + #[test] + fn test_normalize_symbol_mixed_case() { + assert_eq!(normalize_symbol("BtC/UsDt"), "btcusdt"); + } +} diff --git a/ccxt-exchanges/src/binance/ws/types.rs b/ccxt-exchanges/src/binance/ws/types.rs new file mode 100644 index 0000000..3fb9e8e --- /dev/null +++ b/ccxt-exchanges/src/binance/ws/types.rs @@ -0,0 +1,431 @@ +//! WebSocket types and configuration for Binance +//! +//! This module contains configuration structures, enums, and types used +//! throughout the Binance WebSocket implementation. + +use ccxt_core::ws_client::BackpressureStrategy; +use std::sync::atomic::{AtomicBool, AtomicU64}; +use std::time::Duration; + +/// Default channel capacity for ticker streams +pub const DEFAULT_TICKER_CAPACITY: usize = 256; +/// Default channel capacity for orderbook streams +pub const DEFAULT_ORDERBOOK_CAPACITY: usize = 512; +/// Default channel capacity for trade streams +pub const DEFAULT_TRADES_CAPACITY: usize = 1024; +/// Default channel capacity for user data streams +pub const DEFAULT_USER_DATA_CAPACITY: usize = 256; + +/// Configuration for WebSocket channel capacities. +/// +/// Different data streams have different message frequencies: +/// - Trades are highest frequency, need larger buffers +/// - Tickers and user data are lower frequency +#[derive(Debug, Clone)] +pub struct WsChannelConfig { + /// Channel capacity for ticker streams (default: 256) + pub ticker_capacity: usize, + /// Channel capacity for orderbook streams (default: 512) + pub orderbook_capacity: usize, + /// Channel capacity for trade streams (default: 1024) + pub trades_capacity: usize, + /// Channel capacity for user data streams (default: 256) + pub user_data_capacity: usize, +} + +impl Default for WsChannelConfig { + fn default() -> Self { + Self { + ticker_capacity: DEFAULT_TICKER_CAPACITY, + orderbook_capacity: DEFAULT_ORDERBOOK_CAPACITY, + trades_capacity: DEFAULT_TRADES_CAPACITY, + user_data_capacity: DEFAULT_USER_DATA_CAPACITY, + } + } +} + +/// Order book depth levels supported by Binance. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)] +pub enum DepthLevel { + /// 5 levels of depth + L5 = 5, + /// 10 levels of depth + L10 = 10, + /// 20 levels of depth + #[default] + L20 = 20, +} + +impl DepthLevel { + /// Returns the numeric value of the depth level + pub fn as_u32(self) -> u32 { + self as u32 + } +} + +/// Update speed for order book streams. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)] +pub enum UpdateSpeed { + /// 100 millisecond updates (faster, more bandwidth) + #[default] + Ms100, + /// 1000 millisecond updates (slower, less bandwidth) + Ms1000, +} + +impl UpdateSpeed { + /// Returns the string representation for Binance API + pub fn as_str(&self) -> &'static str { + match self { + Self::Ms100 => "100ms", + Self::Ms1000 => "1000ms", + } + } + + /// Returns the millisecond value + pub fn as_millis(&self) -> u64 { + match self { + Self::Ms100 => 100, + Self::Ms1000 => 1000, + } + } +} + +/// Error recovery strategy for WebSocket errors. +/// +/// This enum helps callers understand how to handle different error types. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum WsErrorRecovery { + /// Retry the operation with exponential backoff + Retry { + /// Maximum number of retry attempts + max_attempts: u32, + /// Base backoff duration + backoff: Duration, + }, + /// Resync state (e.g., fetch fresh orderbook snapshot) + Resync, + /// Reconnect the WebSocket connection + Reconnect, + /// Fatal error - requires user intervention + Fatal, +} + +impl WsErrorRecovery { + /// Creates a retry recovery with default settings + pub fn retry_default() -> Self { + Self::Retry { + max_attempts: 3, + backoff: Duration::from_secs(1), + } + } + + /// Returns true if this is a recoverable error + pub fn is_recoverable(&self) -> bool { + !matches!(self, Self::Fatal) + } + + /// Determines the recovery strategy for a given error message. + /// + /// This classifies errors into transient (retryable), resync-needed, + /// reconnect-needed, or fatal categories. + pub fn from_error_message(error_msg: &str) -> Self { + let lower = error_msg.to_lowercase(); + + // Fatal errors - require user intervention + if is_fatal_error(&lower) { + return Self::Fatal; + } + + // Resync errors - need to refresh state + if is_resync_error(&lower) { + return Self::Resync; + } + + // Reconnect errors - need new connection + if is_reconnect_error(&lower) { + return Self::Reconnect; + } + + // Default to retry for transient errors + Self::retry_default() + } +} + +/// Checks if an error message indicates a fatal (non-recoverable) error. +/// +/// Fatal errors include: +/// - Authentication failures +/// - Invalid API keys +/// - Permission denied +/// - Account issues +fn is_fatal_error(error_msg: &str) -> bool { + const FATAL_PATTERNS: &[&str] = &[ + "invalid api", + "api key", + "authentication", + "unauthorized", + "permission denied", + "forbidden", + "account", + "banned", + "ip banned", + "invalid signature", + "signature", + ]; + + FATAL_PATTERNS.iter().any(|p| error_msg.contains(p)) +} + +/// Checks if an error message indicates a resync is needed. +/// +/// Resync errors include: +/// - Sequence gaps in orderbook +/// - Stale data +/// - Out of sync +fn is_resync_error(error_msg: &str) -> bool { + const RESYNC_PATTERNS: &[&str] = &[ + "resync", + "sequence", + "out of sync", + "stale", + "gap", + "missing update", + ]; + + RESYNC_PATTERNS.iter().any(|p| error_msg.contains(p)) +} + +/// Checks if an error message indicates a reconnection is needed. +/// +/// Reconnect errors include: +/// - Connection closed +/// - Connection reset +/// - Server disconnect +fn is_reconnect_error(error_msg: &str) -> bool { + const RECONNECT_PATTERNS: &[&str] = &[ + "connection closed", + "connection reset", + "disconnected", + "eof", + "broken pipe", + "connection refused", + "server closed", + ]; + + RECONNECT_PATTERNS.iter().any(|p| error_msg.contains(p)) +} + +/// Health snapshot for WebSocket connection monitoring. +/// +/// Provides metrics for monitoring connection health and performance. +#[derive(Debug, Clone, Default)] +pub struct WsHealthSnapshot { + /// Round-trip latency in milliseconds (from ping/pong) + pub latency_ms: Option, + /// Total number of messages received + pub messages_received: u64, + /// Number of messages dropped due to backpressure + pub messages_dropped: u64, + /// Timestamp of the last received message (milliseconds since epoch) + pub last_message_time: Option, + /// Connection uptime in milliseconds + pub connection_uptime_ms: u64, + /// Number of reconnection attempts + pub reconnect_count: u32, +} + +impl WsHealthSnapshot { + /// Returns true if the connection appears healthy + pub fn is_healthy(&self) -> bool { + // Consider unhealthy if: + // - No messages received in last 60 seconds + // - Drop rate > 10% + if let Some(last_time) = self.last_message_time { + let now = chrono::Utc::now().timestamp_millis(); + if now - last_time > 60_000 { + return false; + } + } + + if self.messages_received > 0 { + let drop_rate = self.messages_dropped as f64 / self.messages_received as f64; + if drop_rate > 0.1 { + return false; + } + } + + true + } +} + +/// Internal statistics tracking for WebSocket connections. +#[derive(Debug, Default)] +pub struct WsStats { + /// Total messages received + pub messages_received: AtomicU64, + /// Total messages dropped + pub messages_dropped: AtomicU64, + /// Last message timestamp + pub last_message_time: AtomicU64, + /// Connection start time + pub connection_start_time: AtomicU64, + /// Reconnect count + pub reconnect_count: std::sync::atomic::AtomicU32, +} + +/// Configuration for Binance WebSocket client. +/// +/// Combines all configuration options for creating a BinanceWs instance. +#[derive(Debug, Clone)] +pub struct BinanceWsConfig { + /// WebSocket endpoint URL + pub url: String, + /// Channel capacity configuration + pub channel_config: WsChannelConfig, + /// Backpressure strategy for handling channel overflow + pub backpressure_strategy: BackpressureStrategy, + /// Shutdown timeout in milliseconds + pub shutdown_timeout_ms: u64, +} + +impl BinanceWsConfig { + /// Creates a new configuration with the given URL + pub fn new(url: String) -> Self { + Self { + url, + channel_config: WsChannelConfig::default(), + backpressure_strategy: BackpressureStrategy::DropOldest, + shutdown_timeout_ms: 5000, + } + } + + /// Sets the channel configuration + pub fn with_channel_config(mut self, config: WsChannelConfig) -> Self { + self.channel_config = config; + self + } + + /// Sets the backpressure strategy + pub fn with_backpressure(mut self, strategy: BackpressureStrategy) -> Self { + self.backpressure_strategy = strategy; + self + } + + /// Sets the shutdown timeout + pub fn with_shutdown_timeout(mut self, timeout_ms: u64) -> Self { + self.shutdown_timeout_ms = timeout_ms; + self + } +} + +/// Shutdown state tracking +#[derive(Debug, Default)] +pub struct ShutdownState { + /// Whether shutdown has been initiated + pub is_shutting_down: AtomicBool, + /// Whether shutdown has completed + pub shutdown_complete: AtomicBool, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_depth_level_values() { + assert_eq!(DepthLevel::L5.as_u32(), 5); + assert_eq!(DepthLevel::L10.as_u32(), 10); + assert_eq!(DepthLevel::L20.as_u32(), 20); + } + + #[test] + fn test_update_speed_str() { + assert_eq!(UpdateSpeed::Ms100.as_str(), "100ms"); + assert_eq!(UpdateSpeed::Ms1000.as_str(), "1000ms"); + } + + #[test] + fn test_update_speed_millis() { + assert_eq!(UpdateSpeed::Ms100.as_millis(), 100); + assert_eq!(UpdateSpeed::Ms1000.as_millis(), 1000); + } + + #[test] + fn test_ws_error_recovery_is_recoverable() { + assert!(WsErrorRecovery::retry_default().is_recoverable()); + assert!(WsErrorRecovery::Resync.is_recoverable()); + assert!(WsErrorRecovery::Reconnect.is_recoverable()); + assert!(!WsErrorRecovery::Fatal.is_recoverable()); + } + + #[test] + fn test_ws_channel_config_default() { + let config = WsChannelConfig::default(); + assert_eq!(config.ticker_capacity, DEFAULT_TICKER_CAPACITY); + assert_eq!(config.orderbook_capacity, DEFAULT_ORDERBOOK_CAPACITY); + assert_eq!(config.trades_capacity, DEFAULT_TRADES_CAPACITY); + assert_eq!(config.user_data_capacity, DEFAULT_USER_DATA_CAPACITY); + } + + #[test] + fn test_binance_ws_config_builder() { + let config = BinanceWsConfig::new("wss://test.com".to_string()) + .with_backpressure(BackpressureStrategy::DropNewest) + .with_shutdown_timeout(10000); + + assert_eq!(config.url, "wss://test.com"); + assert_eq!( + config.backpressure_strategy, + BackpressureStrategy::DropNewest + ); + assert_eq!(config.shutdown_timeout_ms, 10000); + } + + #[test] + fn test_ws_error_recovery_fatal_errors() { + assert_eq!( + WsErrorRecovery::from_error_message("Invalid API key"), + WsErrorRecovery::Fatal + ); + assert_eq!( + WsErrorRecovery::from_error_message("Authentication failed"), + WsErrorRecovery::Fatal + ); + assert_eq!( + WsErrorRecovery::from_error_message("Permission denied"), + WsErrorRecovery::Fatal + ); + } + + #[test] + fn test_ws_error_recovery_resync_errors() { + assert_eq!( + WsErrorRecovery::from_error_message("RESYNC_NEEDED: sequence gap"), + WsErrorRecovery::Resync + ); + assert_eq!( + WsErrorRecovery::from_error_message("Out of sync with server"), + WsErrorRecovery::Resync + ); + } + + #[test] + fn test_ws_error_recovery_reconnect_errors() { + assert_eq!( + WsErrorRecovery::from_error_message("Connection closed by server"), + WsErrorRecovery::Reconnect + ); + assert_eq!( + WsErrorRecovery::from_error_message("Connection reset"), + WsErrorRecovery::Reconnect + ); + } + + #[test] + fn test_ws_error_recovery_transient_errors() { + // Unknown errors default to retry + let recovery = WsErrorRecovery::from_error_message("Network timeout"); + assert!(matches!(recovery, WsErrorRecovery::Retry { .. })); + } +} diff --git a/ccxt-exchanges/src/binance/ws_exchange_impl.rs b/ccxt-exchanges/src/binance/ws_exchange_impl.rs index e20a6d1..52da4f2 100644 --- a/ccxt-exchanges/src/binance/ws_exchange_impl.rs +++ b/ccxt-exchanges/src/binance/ws_exchange_impl.rs @@ -568,19 +568,25 @@ impl WsExchange for Binance { let ws = self.connection_manager.get_public_connection().await?; // Use the appropriate subscription method based on channel + // Note: The receiver is intentionally dropped here as this is a fire-and-forget subscription. + // For proper message handling, use the specific watch_* methods instead. match channel { "ticker" => { if let Some(sym) = symbol { let market = self.base.market(sym).await?; - ws.subscribe_ticker(&market.id.to_lowercase()).await + ws.subscribe_ticker(&market.id.to_lowercase()) + .await + .map(|_| ()) } else { - ws.subscribe_all_tickers().await + ws.subscribe_all_tickers().await.map(|_| ()) } } "trade" | "trades" => { if let Some(sym) = symbol { let market = self.base.market(sym).await?; - ws.subscribe_trades(&market.id.to_lowercase()).await + ws.subscribe_trades(&market.id.to_lowercase()) + .await + .map(|_| ()) } else { Err(Error::invalid_request( "Symbol required for trades subscription",