diff --git a/ccxt-exchanges/benches/binance_benchmark.rs b/ccxt-exchanges/benches/binance_benchmark.rs index be37b84..21fad72 100644 --- a/ccxt-exchanges/benches/binance_benchmark.rs +++ b/ccxt-exchanges/benches/binance_benchmark.rs @@ -1,3 +1,5 @@ +#![allow(clippy::disallowed_methods)] +#![allow(deprecated)] use ccxt_core::ExchangeConfig; use ccxt_exchanges::binance::Binance; use ccxt_exchanges::binance::parser; @@ -234,8 +236,8 @@ fn bench_exchange_with_credentials(c: &mut Criterion) { c.bench_function("create_exchange_with_credentials", |b| { b.iter(|| { let config = ExchangeConfig { - api_key: Some("test_key".to_string()), - secret: Some("test_secret".to_string()), + api_key: Some("test_key".to_string().into()), + secret: Some("test_secret".to_string().into()), ..Default::default() }; let _ = Binance::new(black_box(config)); diff --git a/ccxt-exchanges/benches/bybit_benchmark.rs b/ccxt-exchanges/benches/bybit_benchmark.rs index 775c1d5..debfbcf 100644 --- a/ccxt-exchanges/benches/bybit_benchmark.rs +++ b/ccxt-exchanges/benches/bybit_benchmark.rs @@ -1,3 +1,5 @@ +#![allow(clippy::disallowed_methods)] +#![allow(deprecated)] //! Bybit exchange performance benchmarks. //! //! Benchmarks for Bybit API calls, data parsing, and other operations. @@ -19,12 +21,7 @@ fn bench_fetch_ticker(c: &mut Criterion) { c.bench_function("bybit_fetch_ticker_btc_usdt", |b| { b.to_async(&rt).iter(|| async { - let _ = bybit - .fetch_ticker( - black_box("BTC/USDT"), - ccxt_core::types::TickerParams::default(), - ) - .await; + let _ = bybit.fetch_ticker(black_box("BTC/USDT")).await; }); }); } @@ -75,7 +72,7 @@ fn bench_fetch_ohlcv(c: &mut Criterion) { |b, timeframe| { b.to_async(&rt).iter(|| async { let _ = bybit - .fetch_ohlcv(black_box("BTC/USDT"), timeframe, None, Some(100), None) + .fetch_ohlcv(black_box("BTC/USDT"), timeframe, None, Some(100)) .await; }); }, @@ -182,8 +179,8 @@ fn bench_exchange_with_credentials(c: &mut Criterion) { c.bench_function("bybit_create_exchange_with_credentials", |b| { b.iter(|| { let config = ExchangeConfig { - api_key: Some("test_key".to_string()), - secret: Some("test_secret".to_string()), + api_key: Some("test_key".to_string().into()), + secret: Some("test_secret".to_string().into()), ..Default::default() }; let _ = Bybit::new(black_box(config)); @@ -215,7 +212,7 @@ fn bench_concurrent_ticker_fetches(c: &mut Criterion) { let futures: Vec<_> = symbols .into_iter() - .map(|symbol| bybit.fetch_ticker(symbol, ccxt_core::types::TickerParams::default())) + .map(|symbol| bybit.fetch_ticker(symbol)) .collect(); let _ = futures::future::join_all(futures).await; diff --git a/ccxt-exchanges/benches/hyperliquid_benchmark.rs b/ccxt-exchanges/benches/hyperliquid_benchmark.rs index 158901d..fab879f 100644 --- a/ccxt-exchanges/benches/hyperliquid_benchmark.rs +++ b/ccxt-exchanges/benches/hyperliquid_benchmark.rs @@ -1,10 +1,12 @@ +#![allow(clippy::disallowed_methods)] +#![allow(deprecated)] //! HyperLiquid exchange performance benchmarks. //! //! Benchmarks for HyperLiquid API calls, data parsing, and other operations. //! //! Run with: cargo bench --bench hyperliquid_benchmark -use ccxt_exchanges::hyperliquid::{HyperLiquid, HyperLiquidBuilder}; +use ccxt_exchanges::hyperliquid::HyperLiquidBuilder; use criterion::{BenchmarkId, Criterion, black_box, criterion_group, criterion_main}; use serde_json::json; use tokio::runtime::Runtime; @@ -21,12 +23,7 @@ fn bench_fetch_ticker(c: &mut Criterion) { c.bench_function("hyperliquid_fetch_ticker_btc_usdc", |b| { b.to_async(&rt).iter(|| async { - let _ = hyperliquid - .fetch_ticker( - black_box("BTC/USDC:USDC"), - ccxt_core::types::TickerParams::default(), - ) - .await; + let _ = hyperliquid.fetch_ticker(black_box("BTC/USDC:USDC")).await; }); }); } @@ -88,7 +85,7 @@ fn bench_fetch_ohlcv(c: &mut Criterion) { |b, timeframe| { b.to_async(&rt).iter(|| async { let _ = hyperliquid - .fetch_ohlcv(black_box("BTC/USDC:USDC"), timeframe, None, Some(100), None) + .fetch_ohlcv(black_box("BTC/USDC:USDC"), timeframe, None, Some(100)) .await; }); }, @@ -231,9 +228,7 @@ fn bench_concurrent_ticker_fetches(c: &mut Criterion) { let futures: Vec<_> = symbols .into_iter() - .map(|symbol| { - hyperliquid.fetch_ticker(symbol, ccxt_core::types::TickerParams::default()) - }) + .map(|symbol| hyperliquid.fetch_ticker(symbol)) .collect(); let _ = futures::future::join_all(futures).await; diff --git a/ccxt-exchanges/benches/okx_benchmark.rs b/ccxt-exchanges/benches/okx_benchmark.rs index 63462dc..242732d 100644 --- a/ccxt-exchanges/benches/okx_benchmark.rs +++ b/ccxt-exchanges/benches/okx_benchmark.rs @@ -1,3 +1,5 @@ +#![allow(clippy::disallowed_methods)] +#![allow(deprecated)] //! OKX exchange performance benchmarks. //! //! Benchmarks for OKX API calls, data parsing, and other operations. @@ -19,12 +21,7 @@ fn bench_fetch_ticker(c: &mut Criterion) { c.bench_function("okx_fetch_ticker_btc_usdt", |b| { b.to_async(&rt).iter(|| async { - let _ = okx - .fetch_ticker( - black_box("BTC/USDT"), - ccxt_core::types::TickerParams::default(), - ) - .await; + let _ = okx.fetch_ticker(black_box("BTC/USDT")).await; }); }); } @@ -75,7 +72,7 @@ fn bench_fetch_ohlcv(c: &mut Criterion) { |b, timeframe| { b.to_async(&rt).iter(|| async { let _ = okx - .fetch_ohlcv(black_box("BTC/USDT"), timeframe, None, Some(100), None) + .fetch_ohlcv(black_box("BTC/USDT"), timeframe, None, Some(100)) .await; }); }, @@ -181,8 +178,8 @@ fn bench_exchange_with_credentials(c: &mut Criterion) { c.bench_function("okx_create_exchange_with_credentials", |b| { b.iter(|| { let config = ExchangeConfig { - api_key: Some("test_key".to_string()), - secret: Some("test_secret".to_string()), + api_key: Some("test_key".to_string().into()), + secret: Some("test_secret".to_string().into()), ..Default::default() }; let _ = Okx::new(black_box(config)); @@ -214,7 +211,7 @@ fn bench_concurrent_ticker_fetches(c: &mut Criterion) { let futures: Vec<_> = symbols .into_iter() - .map(|symbol| okx.fetch_ticker(symbol, ccxt_core::types::TickerParams::default())) + .map(|symbol| okx.fetch_ticker(symbol)) .collect(); let _ = futures::future::join_all(futures).await; diff --git a/ccxt-exchanges/src/binance/auth.rs b/ccxt-exchanges/src/binance/auth.rs index c89b28a..0693dde 100644 --- a/ccxt-exchanges/src/binance/auth.rs +++ b/ccxt-exchanges/src/binance/auth.rs @@ -219,7 +219,7 @@ mod tests { let signature = auth.sign(query); assert!(signature.is_ok()); - let sig = signature.unwrap(); + let sig = signature.expect("Signature failed"); assert!(!sig.is_empty()); assert_eq!(sig.len(), 64); // HMAC-SHA256 produces 64 hex characters } @@ -234,9 +234,12 @@ mod tests { let signed = auth.sign_params(¶ms); assert!(signed.is_ok()); - let signed_params = signed.unwrap(); + let signed_params = signed.expect("Sign params failed"); assert!(signed_params.contains_key("signature")); - assert_eq!(signed_params.get("symbol").unwrap(), "BTCUSDT"); + assert_eq!( + signed_params.get("symbol").expect("Missing symbol"), + "BTCUSDT" + ); } #[test] @@ -248,12 +251,18 @@ mod tests { let signed = auth.sign_with_timestamp(¶ms, timestamp, Some(5000)); assert!(signed.is_ok()); - let signed_params = signed.unwrap(); + let signed_params = signed.expect("Sign timestamp failed"); assert!(signed_params.contains_key("timestamp")); assert!(signed_params.contains_key("recvWindow")); assert!(signed_params.contains_key("signature")); - assert_eq!(signed_params.get("timestamp").unwrap(), "1234567890"); - assert_eq!(signed_params.get("recvWindow").unwrap(), "5000"); + assert_eq!( + signed_params.get("timestamp").expect("Missing timestamp"), + "1234567890" + ); + assert_eq!( + signed_params.get("recvWindow").expect("Missing recvWindow"), + "5000" + ); } #[test] @@ -275,7 +284,10 @@ mod tests { auth.add_auth_headers(&mut headers); assert!(headers.contains_key("X-MBX-APIKEY")); - assert_eq!(headers.get("X-MBX-APIKEY").unwrap(), "my_api_key"); + assert_eq!( + headers.get("X-MBX-APIKEY").expect("Missing header"), + "my_api_key" + ); } #[test] diff --git a/ccxt-exchanges/src/binance/builder.rs b/ccxt-exchanges/src/binance/builder.rs index 84b5c37..ea24643 100644 --- a/ccxt-exchanges/src/binance/builder.rs +++ b/ccxt-exchanges/src/binance/builder.rs @@ -492,7 +492,11 @@ mod tests { fn test_builder_api_key() { let builder = BinanceBuilder::new().api_key("test-key"); assert_eq!( - builder.config.api_key.as_ref().map(|s| s.expose_secret()), + builder + .config + .api_key + .as_ref() + .map(ccxt_core::SecretString::expose_secret), Some("test-key") ); } @@ -501,7 +505,11 @@ mod tests { fn test_builder_secret() { let builder = BinanceBuilder::new().secret("test-secret"); assert_eq!( - builder.config.secret.as_ref().map(|s| s.expose_secret()), + builder + .config + .secret + .as_ref() + .map(ccxt_core::SecretString::expose_secret), Some("test-secret") ); } @@ -580,11 +588,19 @@ mod tests { .default_type(DefaultType::Spot); assert_eq!( - builder.config.api_key.as_ref().map(|s| s.expose_secret()), + builder + .config + .api_key + .as_ref() + .map(ccxt_core::SecretString::expose_secret), Some("key") ); assert_eq!( - builder.config.secret.as_ref().map(|s| s.expose_secret()), + builder + .config + .secret + .as_ref() + .map(ccxt_core::SecretString::expose_secret), Some("secret") ); assert!(builder.config.sandbox); @@ -598,7 +614,7 @@ mod tests { let result = BinanceBuilder::new().build(); assert!(result.is_ok()); - let binance = result.unwrap(); + let binance = result.expect("Failed to build Binance"); assert_eq!(binance.id(), "binance"); assert_eq!(binance.name(), "Binance"); } @@ -643,7 +659,7 @@ mod tests { .build(); assert!(result.is_ok()); - let binance = result.unwrap(); + let binance = result.expect("Failed to build Binance"); // Verify the TimeSyncManager was created with correct config let time_sync = binance.time_sync(); @@ -659,7 +675,7 @@ mod tests { let result = BinanceBuilder::new().auto_time_sync(false).build(); assert!(result.is_ok()); - let binance = result.unwrap(); + let binance = result.expect("Failed to build Binance"); // Verify auto_sync is disabled let time_sync = binance.time_sync(); diff --git a/ccxt-exchanges/src/binance/endpoint_router.rs b/ccxt-exchanges/src/binance/endpoint_router.rs index fbb78ee..27008b8 100644 --- a/ccxt-exchanges/src/binance/endpoint_router.rs +++ b/ccxt-exchanges/src/binance/endpoint_router.rs @@ -477,10 +477,12 @@ mod tests { #[test] fn test_rest_endpoint_option_public() { let binance = create_test_binance(); - let mut market = Market::default(); - market.id = "BTC-250328-100000-C".to_string(); - market.symbol = "BTC/USDT:USDT-250328-100000-C".to_string(); - market.market_type = MarketType::Option; + let market = Market { + id: "BTC-250328-100000-C".to_string(), + symbol: "BTC/USDT:USDT-250328-100000-C".to_string(), + market_type: MarketType::Option, + ..Default::default() + }; let url = binance.rest_endpoint(&market, EndpointType::Public); assert!(url.contains("eapi.binance.com")); @@ -489,10 +491,12 @@ mod tests { #[test] fn test_rest_endpoint_option_private() { let binance = create_test_binance(); - let mut market = Market::default(); - market.id = "BTC-250328-100000-C".to_string(); - market.symbol = "BTC/USDT:USDT-250328-100000-C".to_string(); - market.market_type = MarketType::Option; + let market = Market { + id: "BTC-250328-100000-C".to_string(), + symbol: "BTC/USDT:USDT-250328-100000-C".to_string(), + market_type: MarketType::Option, + ..Default::default() + }; let url = binance.rest_endpoint(&market, EndpointType::Private); assert!(url.contains("eapi.binance.com")); @@ -646,8 +650,10 @@ mod tests { #[test] fn test_swap_defaults_to_linear_when_not_specified() { let binance = create_test_binance(); - let mut market = Market::default(); - market.market_type = MarketType::Swap; + let market = Market { + market_type: MarketType::Swap, + ..Default::default() + }; // linear and inverse are None let url = binance.rest_endpoint(&market, EndpointType::Public); @@ -658,8 +664,10 @@ mod tests { #[test] fn test_futures_defaults_to_linear_when_not_specified() { let binance = create_test_binance(); - let mut market = Market::default(); - market.market_type = MarketType::Futures; + let market = Market { + market_type: MarketType::Futures, + ..Default::default() + }; // linear and inverse are None let url = binance.rest_endpoint(&market, EndpointType::Public); diff --git a/ccxt-exchanges/src/binance/exchange_impl.rs b/ccxt-exchanges/src/binance/exchange_impl.rs index 36654d4..c76fd04 100644 --- a/ccxt-exchanges/src/binance/exchange_impl.rs +++ b/ccxt-exchanges/src/binance/exchange_impl.rs @@ -717,8 +717,21 @@ impl Binance { } } +// Implement HasHttpClient trait for generic SignedRequestBuilder support +impl ccxt_core::signed_request::HasHttpClient for Binance { + fn http_client(&self) -> &ccxt_core::http_client::HttpClient { + &self.base().http_client + } + + fn base_url(&self) -> &'static str { + // Return empty string since Binance uses full URLs in endpoints + "" + } +} + #[cfg(test)] mod tests { + #![allow(clippy::disallowed_methods)] use super::*; use ccxt_core::ExchangeConfig; @@ -1105,7 +1118,7 @@ mod tests { let mut seen = std::collections::HashSet::new(); for tf in &timeframes { prop_assert!( - seen.insert(tf.clone()), + seen.insert(*tf), "Timeframes should not contain duplicates: {:?}", tf ); @@ -1183,15 +1196,3 @@ mod tests { } } } - -// Implement HasHttpClient trait for generic SignedRequestBuilder support -impl ccxt_core::signed_request::HasHttpClient for Binance { - fn http_client(&self) -> &ccxt_core::http_client::HttpClient { - &self.base().http_client - } - - fn base_url(&self) -> &'static str { - // Return empty string since Binance uses full URLs in endpoints - "" - } -} diff --git a/ccxt-exchanges/src/binance/mod.rs b/ccxt-exchanges/src/binance/mod.rs index 257e601..e861bdb 100644 --- a/ccxt-exchanges/src/binance/mod.rs +++ b/ccxt-exchanges/src/binance/mod.rs @@ -652,6 +652,7 @@ impl Binance { #[cfg(test)] mod tests { + #![allow(clippy::disallowed_methods)] use super::*; #[test] diff --git a/ccxt-exchanges/src/binance/signed_request.rs b/ccxt-exchanges/src/binance/signed_request.rs index 54a317f..bee1ab9 100644 --- a/ccxt-exchanges/src/binance/signed_request.rs +++ b/ccxt-exchanges/src/binance/signed_request.rs @@ -491,7 +491,7 @@ mod tests { let builder = SignedRequestBuilder::new(&binance, "/test").optional_param("limit", none_value); - assert!(builder.params.get("limit").is_none()); + assert!(!builder.params.contains_key("limit")); } #[test] @@ -528,8 +528,8 @@ mod tests { assert_eq!(builder.params.get("fromId"), Some(&"67890".to_string())); assert_eq!(builder.params.get("active"), Some(&"true".to_string())); // Nested objects and arrays should be ignored - assert!(builder.params.get("nested").is_none()); - assert!(builder.params.get("array").is_none()); + assert!(!builder.params.contains_key("nested")); + assert!(!builder.params.contains_key("array")); } #[test] @@ -640,39 +640,43 @@ mod property_tests { recv_window in 1000u64..60000u64 ) { // Create a Binance instance with credentials - let mut config = ExchangeConfig::default(); - config.api_key = Some("test_api_key".to_string().into()); - config.secret = Some("test_secret_key".to_string().into()); + let config = ExchangeConfig { + api_key: Some("test_api_key".to_string().into()), + secret: Some("test_secret_key".to_string().into()), + ..Default::default() + }; - let mut options = super::super::BinanceOptions::default(); - options.recv_window = recv_window; + let options = super::super::BinanceOptions { + recv_window, + ..Default::default() + }; - let binance = super::super::Binance::new_with_options(config, options).unwrap(); + let binance = super::super::Binance::new_with_options(config, options).expect("Failed to create Binance"); // Get auth and sign parameters manually to verify the signing logic - let auth = binance.get_auth().unwrap(); + let auth = binance.get_auth().expect("Failed to get auth"); let timestamp = 1234567890000i64; // Fixed timestamp for testing - let signed_params = auth.sign_with_timestamp(¶ms, timestamp, Some(recv_window)).unwrap(); + let signed_params = auth.sign_with_timestamp(¶ms, timestamp, Some(recv_window)).expect("Sign timestamp failed"); // Property 1: timestamp field exists and matches prop_assert!(signed_params.contains_key("timestamp")); - prop_assert_eq!(signed_params.get("timestamp").unwrap(), ×tamp.to_string()); + prop_assert_eq!(signed_params.get("timestamp").expect("Missing timestamp"), ×tamp.to_string()); // Property 2: signature field exists and is 64 hex characters prop_assert!(signed_params.contains_key("signature")); - let signature = signed_params.get("signature").unwrap(); + let signature = signed_params.get("signature").expect("Missing signature"); prop_assert_eq!(signature.len(), 64, "Signature should be 64 hex characters"); prop_assert!(signature.chars().all(|c| c.is_ascii_hexdigit()), "Signature should be hex"); // Property 3: recvWindow field exists and matches configured value prop_assert!(signed_params.contains_key("recvWindow")); - prop_assert_eq!(signed_params.get("recvWindow").unwrap(), &recv_window.to_string()); + prop_assert_eq!(signed_params.get("recvWindow").expect("Missing recvWindow"), &recv_window.to_string()); // Property 4: All original parameters are preserved for (key, value) in ¶ms { prop_assert!(signed_params.contains_key(key), "Original param {} should be preserved", key); - prop_assert_eq!(signed_params.get(key).unwrap(), value, "Original param {} value should match", key); + prop_assert_eq!(signed_params.get(key).expect("Missing param"), value, "Original param {} value should match", key); } } @@ -694,7 +698,7 @@ mod property_tests { prop_assume!(key != other_key); let config = ExchangeConfig::default(); - let binance = super::super::Binance::new(config).unwrap(); + let binance = super::super::Binance::new(config).expect("Failed to create Binance"); // Build with optional parameter let builder = SignedRequestBuilder::new(&binance, "/test") @@ -711,7 +715,7 @@ mod property_tests { key ); prop_assert_eq!( - builder.params.get(&key).unwrap(), + builder.params.get(&key).expect("Missing param"), &v, "Parameter {} should have correct value", key @@ -736,7 +740,7 @@ mod property_tests { other_key ); prop_assert_eq!( - builder.params.get(&other_key).unwrap(), + builder.params.get(&other_key).expect("Missing other param"), &other_value, "Other parameter {} should have correct value", other_key diff --git a/ccxt-exchanges/src/binance/signing_strategy.rs b/ccxt-exchanges/src/binance/signing_strategy.rs index cf6be49..1e1c2b5 100644 --- a/ccxt-exchanges/src/binance/signing_strategy.rs +++ b/ccxt-exchanges/src/binance/signing_strategy.rs @@ -94,9 +94,8 @@ mod tests { fn test_signing_strategy_creation() { let config = ExchangeConfig::default(); let binance = Arc::new(Binance::new(config).unwrap()); - let strategy = BinanceSigningStrategy::new(binance); + let _strategy = BinanceSigningStrategy::new(binance); // Strategy created successfully - assert!(true); } #[test] diff --git a/ccxt-exchanges/src/binance/symbol.rs b/ccxt-exchanges/src/binance/symbol.rs index 4c0d5d1..4d91c76 100644 --- a/ccxt-exchanges/src/binance/symbol.rs +++ b/ccxt-exchanges/src/binance/symbol.rs @@ -197,31 +197,31 @@ mod tests { #[test] fn test_spot_to_exchange_id() { - let symbol = ParsedSymbol::spot("BTC".to_string(), "USDT".to_string()); + let symbol = ParsedSymbol::spot("BTC", "USDT"); assert_eq!(BinanceSymbolConverter::to_exchange_id(&symbol), "BTCUSDT"); } #[test] fn test_spot_to_exchange_id_various_pairs() { - let eth_usdt = ParsedSymbol::spot("ETH".to_string(), "USDT".to_string()); + let eth_usdt = ParsedSymbol::spot("ETH", "USDT"); assert_eq!(BinanceSymbolConverter::to_exchange_id(ð_usdt), "ETHUSDT"); - let btc_busd = ParsedSymbol::spot("BTC".to_string(), "BUSD".to_string()); + let btc_busd = ParsedSymbol::spot("BTC", "BUSD"); assert_eq!(BinanceSymbolConverter::to_exchange_id(&btc_busd), "BTCBUSD"); - let sol_btc = ParsedSymbol::spot("SOL".to_string(), "BTC".to_string()); + let sol_btc = ParsedSymbol::spot("SOL", "BTC"); assert_eq!(BinanceSymbolConverter::to_exchange_id(&sol_btc), "SOLBTC"); } #[test] fn test_linear_swap_to_exchange_id() { - let symbol = ParsedSymbol::linear_swap("BTC".to_string(), "USDT".to_string()); + let symbol = ParsedSymbol::linear_swap("BTC", "USDT"); assert_eq!(BinanceSymbolConverter::to_exchange_id(&symbol), "BTCUSDT"); } #[test] fn test_inverse_swap_to_exchange_id() { - let symbol = ParsedSymbol::inverse_swap("BTC".to_string(), "USD".to_string()); + let symbol = ParsedSymbol::inverse_swap("BTC", "USD"); assert_eq!( BinanceSymbolConverter::to_exchange_id(&symbol), "BTCUSD_PERP" @@ -231,12 +231,7 @@ mod tests { #[test] fn test_linear_futures_to_exchange_id() { let expiry = ExpiryDate::new(24, 12, 31).unwrap(); - let symbol = ParsedSymbol::futures( - "BTC".to_string(), - "USDT".to_string(), - "USDT".to_string(), - expiry, - ); + let symbol = ParsedSymbol::futures("BTC", "USDT", "USDT", expiry); assert_eq!( BinanceSymbolConverter::to_exchange_id(&symbol), "BTCUSDT_241231" @@ -246,12 +241,7 @@ mod tests { #[test] fn test_inverse_futures_to_exchange_id() { let expiry = ExpiryDate::new(25, 3, 15).unwrap(); - let symbol = ParsedSymbol::futures( - "BTC".to_string(), - "USD".to_string(), - "BTC".to_string(), - expiry, - ); + let symbol = ParsedSymbol::futures("BTC", "USD", "BTC", expiry); assert_eq!( BinanceSymbolConverter::to_exchange_id(&symbol), "BTCUSD_250315" @@ -261,12 +251,7 @@ mod tests { #[test] fn test_futures_date_padding() { let expiry = ExpiryDate::new(25, 1, 5).unwrap(); - let symbol = ParsedSymbol::futures( - "ETH".to_string(), - "USDT".to_string(), - "USDT".to_string(), - expiry, - ); + let symbol = ParsedSymbol::futures("ETH", "USDT", "USDT", expiry); assert_eq!( BinanceSymbolConverter::to_exchange_id(&symbol), "ETHUSDT_250105" diff --git a/ccxt-exchanges/src/binance/time_sync.rs b/ccxt-exchanges/src/binance/time_sync.rs index 36539e5..7bcad6d 100644 --- a/ccxt-exchanges/src/binance/time_sync.rs +++ b/ccxt-exchanges/src/binance/time_sync.rs @@ -450,6 +450,7 @@ impl Default for TimeSyncManager { #[cfg(test)] mod tests { + #![allow(clippy::disallowed_methods)] use super::*; use std::thread; @@ -535,7 +536,7 @@ mod tests { assert!(manager.is_initialized()); // Offset should be approximately 100 (may vary slightly due to timing) let offset = manager.get_offset(); - assert!(offset >= 90 && offset <= 110, "Offset was: {}", offset); + assert!((90..=110).contains(&offset), "Offset was: {}", offset); } #[test] @@ -673,7 +674,7 @@ mod tests { let offset = manager.get_offset(); // Offset should be approximately -500 - assert!(offset >= -600 && offset <= -400, "Offset was: {}", offset); + assert!((-600..=-400).contains(&offset), "Offset was: {}", offset); } #[test] diff --git a/ccxt-exchanges/src/binance/ws/binance_impl.rs b/ccxt-exchanges/src/binance/ws/binance_impl.rs index 086b247..10f69f7 100644 --- a/ccxt-exchanges/src/binance/ws/binance_impl.rs +++ b/ccxt-exchanges/src/binance/ws/binance_impl.rs @@ -4,24 +4,21 @@ impl Binance { /// Subscribes to the ticker stream for a unified symbol pub async fn subscribe_ticker(&self, symbol: &str) -> Result<()> { - let ws = self.create_ws(); - ws.connect().await?; + 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<()> { - let ws = self.create_ws(); - ws.connect().await?; + 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<()> { - let ws = self.create_ws(); - ws.connect().await?; + 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") @@ -30,8 +27,7 @@ impl Binance { /// Subscribes to the candlestick stream for a unified symbol pub async fn subscribe_kline(&self, symbol: &str, interval: &str) -> Result<()> { - let ws = self.create_ws(); - ws.connect().await?; + let ws = self.connection_manager.get_public_connection().await?; let binance_symbol = symbol.replace('/', "").to_lowercase(); ws.subscribe_kline(&binance_symbol, interval).await } @@ -54,8 +50,7 @@ impl Binance { "ticker" }; - let ws = self.create_ws(); - ws.connect().await?; + let ws = self.connection_manager.get_public_connection().await?; ws.watch_ticker_internal(&binance_symbol, channel_name) .await } @@ -93,8 +88,7 @@ impl Binance { None }; - let ws = self.create_ws(); - ws.connect().await?; + let ws = self.connection_manager.get_public_connection().await?; ws.watch_tickers_internal(binance_symbols, channel_name) .await } @@ -131,8 +125,7 @@ impl Binance { "markPrice" }; - let ws = self.create_ws(); - ws.connect().await?; + let ws = self.connection_manager.get_public_connection().await?; ws.watch_ticker_internal(&binance_symbol, channel_name) .await } @@ -166,8 +159,7 @@ impl Binance { )); } - let ws = self.create_ws(); - ws.connect().await?; + let ws = self.connection_manager.get_public_connection().await?; ws.watch_orderbook_internal(self, &binance_symbol, limit, update_speed, is_futures) .await } @@ -216,8 +208,7 @@ impl Binance { 100 }; - let ws = self.create_ws(); - ws.connect().await?; + let ws = self.connection_manager.get_public_connection().await?; ws.watch_orderbooks_internal(self, binance_symbols, limit, update_speed, is_futures) .await } @@ -263,8 +254,7 @@ impl Binance { None }; - let ws = self.create_ws(); - ws.connect().await?; + let ws = self.connection_manager.get_public_connection().await?; ws.watch_tickers_internal(binance_symbols, channel_name) .await } @@ -281,8 +271,7 @@ impl Binance { let market = self.base.market(symbol).await?; let binance_symbol = market.id.to_lowercase(); - let ws = self.create_ws(); - ws.connect().await?; + let ws = self.connection_manager.get_public_connection().await?; ws.watch_trades_internal(symbol, &binance_symbol, since, limit, Some(&market)) .await @@ -301,8 +290,7 @@ impl Binance { let market = self.base.market(symbol).await?; let binance_symbol = market.id.to_lowercase(); - let ws = self.create_ws(); - ws.connect().await?; + let ws = self.connection_manager.get_public_connection().await?; ws.watch_ohlcv_internal(symbol, &binance_symbol, timeframe, since, limit) .await @@ -315,8 +303,7 @@ impl Binance { let market = self.base.market(symbol).await?; let binance_symbol = market.id.to_lowercase(); - let ws = self.create_ws(); - ws.connect().await?; + let ws = self.connection_manager.get_public_connection().await?; ws.watch_bids_asks_internal(symbol, &binance_symbol).await } @@ -352,7 +339,10 @@ impl Binance { true }; - let ws = self.create_authenticated_ws(); + let ws = self + .connection_manager + .get_private_connection(&self) + .await?; if fetch_snapshot { let account_type_enum = account_type.parse::().ok(); @@ -379,7 +369,10 @@ impl Binance { ) -> Result> { self.base.load_markets(false).await?; - let ws = self.create_authenticated_ws(); + let ws = self + .connection_manager + .get_private_connection(&self) + .await?; ws.watch_orders_internal(symbol, since, limit).await } @@ -391,7 +384,10 @@ impl Binance { limit: Option, _params: Option>, ) -> Result> { - let ws = self.create_authenticated_ws(); + let ws = self + .connection_manager + .get_private_connection(&self) + .await?; ws.watch_my_trades_internal(symbol, since, limit).await } @@ -403,7 +399,10 @@ impl Binance { limit: Option, _params: Option>, ) -> Result> { - let ws = self.create_authenticated_ws(); + let ws = self + .connection_manager + .get_private_connection(&self) + .await?; ws.watch_positions_internal(symbols, since, limit).await } } diff --git a/ccxt-exchanges/src/binance/ws/handlers.rs b/ccxt-exchanges/src/binance/ws/handlers.rs index d2a8cba..85f6315 100644 --- a/ccxt-exchanges/src/binance/ws/handlers.rs +++ b/ccxt-exchanges/src/binance/ws/handlers.rs @@ -14,6 +14,17 @@ use std::sync::Arc; use std::time::Duration; use tokio::sync::Mutex; +struct MessageLoopContext { + ws_client: Arc>>, + subscription_manager: Arc, + is_connected: Arc, + reconnect_config: Arc>, + request_id: Arc, + listen_key_manager: Option>, + base_url: String, + current_url: String, +} + /// Message router for handling WebSocket connections and message routing pub struct MessageRouter { /// WebSocket client instance @@ -28,9 +39,15 @@ pub struct MessageRouter { /// Connection state flag is_connected: Arc, + /// Connection lock to ensure idempotent start + connection_lock: Arc>, + /// Configuration for reconnection behavior reconnect_config: Arc>, + /// Listen key manager for user data streams + listen_key_manager: Option>, + /// WebSocket endpoint URL ws_url: String, @@ -43,15 +60,18 @@ impl MessageRouter { pub fn new( ws_url: String, subscription_manager: Arc, + listen_key_manager: Option>, ) -> Self { Self { ws_client: Arc::new(tokio::sync::RwLock::new(None)), subscription_manager, router_task: Arc::new(Mutex::new(None)), is_connected: Arc::new(std::sync::atomic::AtomicBool::new(false)), + connection_lock: Arc::new(Mutex::new(())), reconnect_config: Arc::new(tokio::sync::RwLock::new( super::subscriptions::ReconnectConfig::default(), )), + listen_key_manager, ws_url, request_id: Arc::new(std::sync::atomic::AtomicU64::new(1)), } @@ -59,8 +79,14 @@ impl MessageRouter { /// Starts the message router pub async fn start(&self, url_override: Option) -> Result<()> { + let _lock = self.connection_lock.lock().await; + if self.is_connected() { - self.stop().await?; + if url_override.is_some() { + self.stop().await?; + } else { + return Ok(()); + } } let url = url_override.unwrap_or_else(|| self.ws_url.clone()); @@ -80,18 +106,25 @@ impl MessageRouter { let subscription_manager = self.subscription_manager.clone(); let is_connected = self.is_connected.clone(); let reconnect_config = self.reconnect_config.clone(); + let request_id = self.request_id.clone(); + let listen_key_manager = self.listen_key_manager.clone(); + + let ws_url = self.ws_url.clone(); + let current_url = url; - let ws_url = url; + let ctx = MessageLoopContext { + ws_client, + subscription_manager, + is_connected, + reconnect_config, + request_id, + listen_key_manager, + base_url: ws_url, + current_url, + }; let handle = tokio::spawn(async move { - Self::message_loop( - ws_client, - subscription_manager, - is_connected, - reconnect_config, - ws_url, - ) - .await; + Self::message_loop(ctx).await; }); *self.router_task.lock().await = Some(handle); @@ -207,43 +240,53 @@ impl MessageRouter { } /// Message reception loop - async fn message_loop( - ws_client: Arc>>, - subscription_manager: Arc, - is_connected: Arc, - reconnect_config: Arc>, - ws_url: String, - ) { + async fn message_loop(ctx: MessageLoopContext) { let mut reconnect_attempt = 0; + Self::resubscribe_all(&ctx.ws_client, &ctx.subscription_manager, &ctx.request_id).await; + loop { - if !is_connected.load(std::sync::atomic::Ordering::SeqCst) { + if !ctx.is_connected.load(std::sync::atomic::Ordering::SeqCst) { break; } - let has_client = ws_client.read().await.is_some(); + let has_client = ctx.ws_client.read().await.is_some(); if !has_client { - let config = reconnect_config.read().await; + let config = ctx.reconnect_config.read().await; if config.should_retry(reconnect_attempt) { let delay = config.calculate_delay(reconnect_attempt); drop(config); tokio::time::sleep(Duration::from_millis(delay)).await; - if let Ok(()) = Self::reconnect(&ws_url, ws_client.clone()).await { + if let Ok(()) = Self::reconnect( + &ctx.base_url, + &ctx.current_url, + ctx.ws_client.clone(), + ctx.listen_key_manager.clone(), + ) + .await + { + Self::resubscribe_all( + &ctx.ws_client, + &ctx.subscription_manager, + &ctx.request_id, + ) + .await; reconnect_attempt = 0; continue; } reconnect_attempt += 1; continue; } - is_connected.store(false, std::sync::atomic::Ordering::SeqCst); + ctx.is_connected + .store(false, std::sync::atomic::Ordering::SeqCst); break; } let message_opt = { - let guard = ws_client.read().await; + let guard = ctx.ws_client.read().await; if let Some(client) = guard.as_ref() { client.receive().await } else { @@ -252,55 +295,149 @@ impl MessageRouter { }; if let Some(value) = message_opt { - if let Err(_e) = Self::handle_message(value, subscription_manager.clone()).await { + if let Err(_e) = Self::handle_message( + value, + ctx.subscription_manager.clone(), + ctx.listen_key_manager.clone(), + ) + .await + { continue; } reconnect_attempt = 0; } else { - let config = reconnect_config.read().await; + let config = ctx.reconnect_config.read().await; if config.should_retry(reconnect_attempt) { let delay = config.calculate_delay(reconnect_attempt); drop(config); tokio::time::sleep(Duration::from_millis(delay)).await; - if let Ok(()) = Self::reconnect(&ws_url, ws_client.clone()).await { + if let Ok(()) = Self::reconnect( + &ctx.base_url, + &ctx.current_url, + ctx.ws_client.clone(), + ctx.listen_key_manager.clone(), + ) + .await + { + Self::resubscribe_all( + &ctx.ws_client, + &ctx.subscription_manager, + &ctx.request_id, + ) + .await; reconnect_attempt = 0; continue; } reconnect_attempt += 1; continue; } - is_connected.store(false, std::sync::atomic::Ordering::SeqCst); + ctx.is_connected + .store(false, std::sync::atomic::Ordering::SeqCst); break; } } } + /// Resubscribes to all active streams + async fn resubscribe_all( + ws_client: &Arc>>, + subscription_manager: &Arc, + request_id: &Arc, + ) { + let streams = subscription_manager.get_active_streams().await; + if streams.is_empty() { + return; + } + + let client_opt = ws_client.read().await; + if let Some(client) = client_opt.as_ref() { + // Split into chunks of 10 to avoid hitting limits + for chunk in streams.chunks(10) { + let id = request_id.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + #[allow(clippy::disallowed_methods)] + let request = serde_json::json!({ + "method": "SUBSCRIBE", + "params": chunk, + "id": id + }); + + if let Err(e) = client + .send(tokio_tungstenite::tungstenite::protocol::Message::Text( + request.to_string().into(), + )) + .await + { + tracing::error!("Failed to resubscribe: {}", e); + } + } + } + } + /// Processes a WebSocket message async fn handle_message( message: Value, subscription_manager: Arc, + listen_key_manager: Option>, ) -> Result<()> { let stream_name = Self::extract_stream_name(&message)?; // Check for "data" field (Combined Stream format) and unwrap if present let payload = if message.get("stream").is_some() && message.get("data").is_some() { - message.get("data").cloned().unwrap_or(message) + message.get("data").cloned().unwrap_or(message.clone()) } else { - message + message.clone() }; + // Handle listenKeyExpired event + if stream_name == "!userData" { + if let Some(event) = payload.get("e").and_then(|e| e.as_str()) { + if event == "listenKeyExpired" { + if let Some(manager) = listen_key_manager { + tracing::info!("Received listenKeyExpired event, regenerating key..."); + let _ = manager.regenerate().await; + } + } + } + } + let sent = subscription_manager - .send_to_stream(&stream_name, payload) + .send_to_stream(&stream_name, payload.clone()) .await; if sent { - Ok(()) - } else { - Err(Error::generic("No subscribers for stream")) + return Ok(()); } + + // Handle optional stream suffixes (e.g. @1s, @100ms) + // If exact match failed, try to find matches by symbol that start with the extracted stream name + let symbol_opt = payload.get("s").and_then(|s| s.as_str()); + + if let Some(symbol) = symbol_opt { + let active_streams = subscription_manager + .get_subscriptions_by_symbol(symbol) + .await; + let mut fallback_sent = false; + + for sub in active_streams { + if sub.stream.starts_with(&stream_name) && sub.stream != stream_name { + if subscription_manager + .send_to_stream(&sub.stream, payload.clone()) + .await + { + fallback_sent = true; + } + } + } + + if fallback_sent { + return Ok(()); + } + } + + Err(Error::generic("No subscribers for stream")) } /// Extracts the stream name from an incoming message @@ -362,8 +499,10 @@ impl MessageRouter { /// Reconnects the WebSocket client async fn reconnect( - ws_url: &str, + base_url: &str, + current_url: &str, ws_client: Arc>>, + listen_key_manager: Option>, ) -> Result<()> { { let mut client_opt = ws_client.write().await; @@ -372,8 +511,26 @@ impl MessageRouter { } } + // Determine if we should refresh URL from listen key manager + let mut final_url = current_url.to_string(); + + if let Some(manager) = listen_key_manager { + // If current URL is different from base, we assume it's a private stream + // that might need a fresh listen key + if current_url != base_url { + if let Ok(key) = manager.get_or_create().await { + let base = if let Some(stripped) = base_url.strip_suffix('/') { + stripped + } else { + base_url + }; + final_url = format!("{}/{}", base, key); + } + } + } + let config = ccxt_core::ws_client::WsConfig { - url: ws_url.to_string(), + url: final_url, ..Default::default() }; let new_client = ccxt_core::ws_client::WsClient::new(config); @@ -484,14 +641,9 @@ pub async fn fetch_orderbook_snapshot( is_futures: bool, orderbooks: &Mutex>, ) -> Result { - let mut params = HashMap::new(); - if let Some(l) = limit { - #[allow(clippy::disallowed_methods)] - let limit_value = serde_json::json!(l); - params.insert("limit".to_string(), limit_value); - } - - let mut snapshot = exchange.fetch_order_book(symbol, None).await?; + let mut snapshot = exchange + .fetch_order_book(symbol, limit.map(|l| l as u32)) + .await?; snapshot.is_synced = true; @@ -513,6 +665,7 @@ pub async fn fetch_orderbook_snapshot( #[cfg(test)] mod tests { + #![allow(clippy::disallowed_methods)] use super::*; use serde_json::json; use std::sync::Arc; @@ -545,7 +698,7 @@ mod tests { #[tokio::test] async fn test_handle_message_unwrapping() { let manager = Arc::new(crate::binance::ws::subscriptions::SubscriptionManager::new()); - let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let (tx, mut rx) = tokio::sync::mpsc::channel(100); manager .add_subscription( @@ -566,7 +719,7 @@ mod tests { } }); - MessageRouter::handle_message(message, manager) + MessageRouter::handle_message(message, manager, None) .await .unwrap(); diff --git a/ccxt-exchanges/src/binance/ws/listen_key.rs b/ccxt-exchanges/src/binance/ws/listen_key.rs index bc0fe2a..aab7870 100644 --- a/ccxt-exchanges/src/binance/ws/listen_key.rs +++ b/ccxt-exchanges/src/binance/ws/listen_key.rs @@ -84,6 +84,12 @@ impl ListenKeyManager { } } + /// Regenerates the listen key (delete old, create new) + pub async fn regenerate(&self) -> Result { + let _ = self.delete().await; + self.create_new().await + } + /// Starts the auto-refresh task pub async fn start_auto_refresh(&self) { self.stop_auto_refresh().await; @@ -103,10 +109,27 @@ impl ListenKeyManager { Ok(()) => { *created_at.write().await = Some(Instant::now()); } - Err(_e) => { - *listen_key.write().await = None; - *created_at.write().await = None; - break; + 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; + } + } } } } else { diff --git a/ccxt-exchanges/src/binance/ws/mod.rs b/ccxt-exchanges/src/binance/ws/mod.rs index c09e7f0..44bb503 100644 --- a/ccxt-exchanges/src/binance/ws/mod.rs +++ b/ccxt-exchanges/src/binance/ws/mod.rs @@ -62,7 +62,7 @@ impl BinanceWs { /// Creates a new Binance WebSocket client pub fn new(url: String) -> Self { let subscription_manager = Arc::new(SubscriptionManager::new()); - let message_router = Arc::new(MessageRouter::new(url, subscription_manager.clone())); + let message_router = Arc::new(MessageRouter::new(url, subscription_manager.clone(), None)); // Start the router immediately let router_clone = message_router.clone(); @@ -93,7 +93,12 @@ impl BinanceWs { /// Creates a WebSocket client with a listen key manager pub fn new_with_auth(url: String, binance: Arc) -> Self { let subscription_manager = Arc::new(SubscriptionManager::new()); - let message_router = Arc::new(MessageRouter::new(url, subscription_manager.clone())); + let listen_key_manager = Arc::new(ListenKeyManager::new(binance)); + let message_router = Arc::new(MessageRouter::new( + url, + subscription_manager.clone(), + Some(listen_key_manager.clone()), + )); // Start the router immediately let router_clone = message_router.clone(); @@ -107,7 +112,7 @@ impl BinanceWs { message_router, subscription_manager, listen_key: Arc::new(RwLock::new(None)), - listen_key_manager: Some(Arc::new(ListenKeyManager::new(binance))), + listen_key_manager: Some(listen_key_manager), tickers: Arc::new(Mutex::new(HashMap::new())), bids_asks: Arc::new(Mutex::new(HashMap::new())), mark_prices: Arc::new(Mutex::new(HashMap::new())), @@ -156,8 +161,8 @@ impl BinanceWs { let listen_key = manager.get_or_create().await?; let base_url = self.message_router.get_url(); - let base_url = if base_url.ends_with('/') { - &base_url[..base_url.len() - 1] + let base_url = if let Some(stripped) = base_url.strip_suffix('/') { + stripped } else { &base_url }; @@ -192,7 +197,7 @@ 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::unbounded_channel(); + let (tx, _rx) = tokio::sync::mpsc::channel(1024); self.subscription_manager .add_subscription( @@ -209,7 +214,7 @@ impl BinanceWs { /// Subscribes to the 24-hour ticker stream for all symbols pub async fn subscribe_all_tickers(&self) -> Result<()> { let stream = "!ticker@arr".to_string(); - let (tx, _rx) = tokio::sync::mpsc::unbounded_channel(); + let (tx, _rx) = tokio::sync::mpsc::channel(1024); self.subscription_manager .add_subscription( @@ -226,7 +231,7 @@ impl BinanceWs { /// 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::unbounded_channel(); + let (tx, _rx) = tokio::sync::mpsc::channel(1024); self.subscription_manager .add_subscription( @@ -243,7 +248,7 @@ impl BinanceWs { /// 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::unbounded_channel(); + let (tx, _rx) = tokio::sync::mpsc::channel(1024); self.subscription_manager .add_subscription( @@ -269,7 +274,7 @@ impl BinanceWs { } else { format!("{}@depth{}", symbol.to_lowercase(), levels) }; - let (tx, _rx) = tokio::sync::mpsc::unbounded_channel(); + let (tx, _rx) = tokio::sync::mpsc::channel(1024); self.subscription_manager .add_subscription( @@ -298,7 +303,7 @@ impl BinanceWs { } else { format!("{}@depth", symbol.to_lowercase()) }; - let (tx, _rx) = tokio::sync::mpsc::unbounded_channel(); + let (tx, _rx) = tokio::sync::mpsc::channel(1024); self.subscription_manager .add_subscription( @@ -315,7 +320,7 @@ impl BinanceWs { /// 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::unbounded_channel(); + let (tx, _rx) = tokio::sync::mpsc::channel(1024); self.subscription_manager .add_subscription( @@ -332,7 +337,7 @@ impl BinanceWs { /// 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::unbounded_channel(); + let (tx, _rx) = tokio::sync::mpsc::channel(1024); self.subscription_manager .add_subscription( @@ -349,7 +354,7 @@ impl BinanceWs { /// Subscribes to the mini ticker stream for all symbols pub async fn subscribe_all_mini_tickers(&self) -> Result<()> { let stream = "!miniTicker@arr".to_string(); - let (tx, _rx) = tokio::sync::mpsc::unbounded_channel(); + let (tx, _rx) = tokio::sync::mpsc::channel(1024); self.subscription_manager .add_subscription( @@ -409,7 +414,7 @@ impl BinanceWs { async fn watch_ticker_internal(&self, symbol: &str, channel_name: &str) -> Result { let stream = format!("{}@{}", symbol.to_lowercase(), channel_name); - let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let (tx, mut rx) = tokio::sync::mpsc::channel(1024); self.subscription_manager .add_subscription( stream.clone(), @@ -452,7 +457,7 @@ impl BinanceWs { vec![format!("!{}@arr", channel_name)] }; - let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let (tx, mut rx) = tokio::sync::mpsc::channel(1024); for stream in &streams { self.subscription_manager @@ -556,7 +561,7 @@ impl BinanceWs { format!("{}@depth", symbol.to_lowercase()) }; - let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let (tx, mut rx) = tokio::sync::mpsc::channel(1024); self.subscription_manager .add_subscription( stream.clone(), @@ -674,7 +679,7 @@ impl BinanceWs { )); } - let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let (tx, mut rx) = tokio::sync::mpsc::channel(1024); let mut streams = Vec::new(); for symbol in &symbols { @@ -707,9 +712,9 @@ impl BinanceWs { } let mut result = HashMap::new(); - let mut update_count = 0; + let mut synced_symbols = std::collections::HashSet::new(); - while update_count < symbols.len() { + while synced_symbols.len() < symbols.len() { if let Some(message) = rx.recv().await { if message.get("result").is_some() { continue; @@ -728,7 +733,12 @@ impl BinanceWs { continue; } - update_count += 1; + let orderbooks = self.orderbooks.lock().await; + if let Some(ob) = orderbooks.get(msg_symbol) { + if ob.is_synced { + synced_symbols.insert(msg_symbol.to_string()); + } + } } } } @@ -750,7 +760,7 @@ 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::unbounded_channel(); + let (tx, mut rx) = tokio::sync::mpsc::channel(1024); self.subscription_manager .add_subscription( @@ -790,7 +800,7 @@ impl BinanceWs { market: Option<&ccxt_core::types::Market>, ) -> Result> { let stream = format!("{}@trade", market_id.to_lowercase()); - let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let (tx, mut rx) = tokio::sync::mpsc::channel(1024); self.subscription_manager .add_subscription( @@ -853,7 +863,7 @@ impl BinanceWs { limit: Option, ) -> Result> { let stream = format!("{}@kline_{}", market_id.to_lowercase(), timeframe); - let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let (tx, mut rx) = tokio::sync::mpsc::channel(1024); self.subscription_manager .add_subscription( @@ -923,7 +933,7 @@ impl BinanceWs { async fn watch_balance_internal(&self, account_type: &str) -> Result { self.connect_user_stream().await?; - let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let (tx, mut rx) = tokio::sync::mpsc::channel(1024); self.subscription_manager .add_subscription( @@ -964,7 +974,7 @@ impl BinanceWs { ) -> Result> { self.connect_user_stream().await?; - let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let (tx, mut rx) = tokio::sync::mpsc::channel(1024); self.subscription_manager .add_subscription( @@ -1028,7 +1038,7 @@ impl BinanceWs { ) -> Result> { self.connect_user_stream().await?; - let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let (tx, mut rx) = tokio::sync::mpsc::channel(1024); self.subscription_manager .add_subscription( @@ -1075,7 +1085,7 @@ impl BinanceWs { ) -> Result> { self.connect_user_stream().await?; - let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let (tx, mut rx) = tokio::sync::mpsc::channel(1024); self.subscription_manager .add_subscription( @@ -1261,6 +1271,7 @@ impl BinanceWs { include!("binance_impl.rs"); #[cfg(test)] +#[allow(clippy::disallowed_methods)] mod tests { use super::*; @@ -1290,7 +1301,7 @@ mod tests { #[tokio::test] async fn test_subscription_manager_basic() { let manager = SubscriptionManager::new(); - let (tx, _rx) = tokio::sync::mpsc::unbounded_channel(); + let (tx, _rx) = tokio::sync::mpsc::channel(1024); assert_eq!(manager.active_count(), 0); assert!(!manager.has_subscription("btcusdt@ticker").await); diff --git a/ccxt-exchanges/src/binance/ws/subscriptions.rs b/ccxt-exchanges/src/binance/ws/subscriptions.rs index 9da0930..9cdc1ce 100644 --- a/ccxt-exchanges/src/binance/ws/subscriptions.rs +++ b/ccxt-exchanges/src/binance/ws/subscriptions.rs @@ -69,7 +69,7 @@ pub struct Subscription { /// Timestamp when the subscription was created pub subscribed_at: Instant, /// Sender for forwarding WebSocket messages to consumers - pub sender: tokio::sync::mpsc::UnboundedSender, + pub sender: tokio::sync::mpsc::Sender, } impl Subscription { @@ -78,7 +78,7 @@ impl Subscription { stream: String, symbol: String, sub_type: SubscriptionType, - sender: tokio::sync::mpsc::UnboundedSender, + sender: tokio::sync::mpsc::Sender, ) -> Self { Self { stream, @@ -91,7 +91,8 @@ impl Subscription { /// Sends a message to the subscriber pub fn send(&self, message: Value) -> bool { - self.sender.send(message).is_ok() + // Use try_send to avoid blocking if channel is full (drop strategy) + self.sender.try_send(message).is_ok() } } @@ -121,7 +122,7 @@ impl SubscriptionManager { stream: String, symbol: String, sub_type: SubscriptionType, - sender: tokio::sync::mpsc::UnboundedSender, + sender: tokio::sync::mpsc::Sender, ) -> ccxt_core::error::Result<()> { let subscription = Subscription::new(stream.clone(), symbol.clone(), sub_type, sender); @@ -219,10 +220,16 @@ impl SubscriptionManager { pub async fn send_to_stream(&self, stream: &str, message: Value) -> bool { let subs = self.subscriptions.read().await; if let Some(subscription) = subs.get(stream) { - subscription.send(message) + if subscription.send(message) { + return true; + } } else { - false + return false; } + drop(subs); + + let _ = self.remove_subscription(stream).await; + false } /// Sends a message to all subscribers of a symbol @@ -231,19 +238,34 @@ impl SubscriptionManager { let subs = self.subscriptions.read().await; let mut sent_count = 0; + let mut streams_to_remove = Vec::new(); if let Some(streams) = index.get(symbol) { for stream in streams { if let Some(subscription) = subs.get(stream) { if subscription.send(message.clone()) { sent_count += 1; + } else { + streams_to_remove.push(stream.clone()); } } } } + drop(subs); + drop(index); + + for stream in streams_to_remove { + let _ = self.remove_subscription(&stream).await; + } sent_count } + + /// Returns a list of all active stream names for resubscription + pub async fn get_active_streams(&self) -> Vec { + let subs = self.subscriptions.read().await; + subs.keys().cloned().collect() + } } impl Default for SubscriptionManager { diff --git a/ccxt-exchanges/src/binance/ws_exchange_impl.rs b/ccxt-exchanges/src/binance/ws_exchange_impl.rs index 2ad7f02..e20a6d1 100644 --- a/ccxt-exchanges/src/binance/ws_exchange_impl.rs +++ b/ccxt-exchanges/src/binance/ws_exchange_impl.rs @@ -23,11 +23,11 @@ use super::Binance; /// A simple stream wrapper that converts an mpsc receiver into a Stream struct ReceiverStream { - receiver: mpsc::UnboundedReceiver, + receiver: mpsc::Receiver, } impl ReceiverStream { - fn new(receiver: mpsc::UnboundedReceiver) -> Self { + fn new(receiver: mpsc::Receiver) -> Self { Self { receiver } } } @@ -84,7 +84,7 @@ impl WsExchange for Binance { let ws = self.connection_manager.get_public_connection().await?; // Create subscription channel - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = mpsc::channel(1024); // Register with manager ws.subscription_manager @@ -100,7 +100,7 @@ impl WsExchange for Binance { ws.message_router.subscribe(vec![stream]).await?; // Create user channel - let (user_tx, user_rx) = mpsc::unbounded_channel::>(); + let (user_tx, user_rx) = mpsc::channel::>(1024); // Spawn parser task let market_clone = market.clone(); @@ -108,12 +108,12 @@ impl WsExchange for Binance { while let Some(msg) = rx.recv().await { match super::parser::parse_ws_ticker(&msg, Some(&market_clone)) { Ok(ticker) => { - if user_tx.send(Ok(ticker)).is_err() { + if user_tx.send(Ok(ticker)).await.is_err() { break; } } Err(e) => { - let _ = user_tx.send(Err(e)); + let _ = user_tx.send(Err(e)).await; } } } @@ -129,7 +129,7 @@ impl WsExchange for Binance { self.load_markets(false).await?; // Create aggregator channel - let (agg_tx, mut agg_rx) = mpsc::unbounded_channel::(); + let (agg_tx, mut agg_rx) = mpsc::channel::(1024); // Subscribe to all requested symbols (potentially across shards) let mut markets = HashMap::new(); @@ -142,7 +142,7 @@ impl WsExchange for Binance { // Get connection (might be different for each symbol if sharding active) let ws = self.connection_manager.get_public_connection().await?; - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = mpsc::channel(1024); ws.subscription_manager .add_subscription( @@ -162,7 +162,7 @@ impl WsExchange for Binance { tokio::spawn(async move { while let Some(msg) = rx.recv().await { if let Ok(ticker) = super::parser::parse_ws_ticker(&msg, Some(&market_clone)) { - let _ = agg_tx_clone.send(ticker); + let _ = agg_tx_clone.send(ticker).await; } } }); @@ -171,7 +171,7 @@ impl WsExchange for Binance { drop(agg_tx); // Create user channel - let (user_tx, user_rx) = mpsc::unbounded_channel::>>(); + let (user_tx, user_rx) = mpsc::channel::>>(1024); // Spawn aggregator task tokio::spawn(async move { @@ -181,7 +181,7 @@ impl WsExchange for Binance { tickers.insert(ticker.symbol.clone(), ticker); let ticker_vec: Vec = tickers.values().cloned().collect(); - if user_tx.send(Ok(ticker_vec)).is_err() { + if user_tx.send(Ok(ticker_vec)).await.is_err() { break; } } @@ -211,7 +211,7 @@ impl WsExchange for Binance { let ws = self.connection_manager.get_public_connection().await?; // Create subscription channel - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = mpsc::channel(1024); // Register with manager ws.subscription_manager @@ -227,7 +227,7 @@ impl WsExchange for Binance { ws.message_router.subscribe(vec![stream]).await?; // Create user channel - let (user_tx, user_rx) = mpsc::unbounded_channel::>(); + let (user_tx, user_rx) = mpsc::channel::>(1024); // Spawn parser task let symbol_clone = symbol.to_string(); @@ -235,12 +235,12 @@ impl WsExchange for Binance { while let Some(msg) = rx.recv().await { match super::parser::parse_ws_orderbook(&msg, symbol_clone.clone()) { Ok(orderbook) => { - if user_tx.send(Ok(orderbook)).is_err() { + if user_tx.send(Ok(orderbook)).await.is_err() { break; } } Err(e) => { - let _ = user_tx.send(Err(e)); + let _ = user_tx.send(Err(e)).await; } } } @@ -263,7 +263,7 @@ impl WsExchange for Binance { let ws = self.connection_manager.get_public_connection().await?; // Create subscription channel - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = mpsc::channel(1024); // Register with manager ws.subscription_manager @@ -279,7 +279,7 @@ impl WsExchange for Binance { ws.message_router.subscribe(vec![stream]).await?; // Create user channel - let (user_tx, user_rx) = mpsc::unbounded_channel::>>(); + let (user_tx, user_rx) = mpsc::channel::>>(1024); // Spawn parser task let market_clone = market.clone(); @@ -287,12 +287,12 @@ impl WsExchange for Binance { while let Some(msg) = rx.recv().await { match super::parser::parse_ws_trade(&msg, Some(&market_clone)) { Ok(trade) => { - if user_tx.send(Ok(vec![trade])).is_err() { + if user_tx.send(Ok(vec![trade])).await.is_err() { break; } } Err(e) => { - let _ = user_tx.send(Err(e)); + let _ = user_tx.send(Err(e)).await; } } } @@ -323,7 +323,7 @@ impl WsExchange for Binance { let ws = self.connection_manager.get_public_connection().await?; // Create subscription channel - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = mpsc::channel(1024); // Register with manager ws.subscription_manager @@ -339,7 +339,7 @@ impl WsExchange for Binance { ws.message_router.subscribe(vec![stream]).await?; // Create user channel - let (user_tx, user_rx) = mpsc::unbounded_channel::>(); + let (user_tx, user_rx) = mpsc::channel::>(1024); // Spawn parser task tokio::spawn(async move { @@ -364,12 +364,12 @@ impl WsExchange for Binance { Decimal::try_from(ohlcv_f64.volume).unwrap_or_default(), ), }; - if user_tx.send(Ok(ohlcv)).is_err() { + if user_tx.send(Ok(ohlcv)).await.is_err() { break; } } Err(e) => { - let _ = user_tx.send(Err(e)); + let _ = user_tx.send(Err(e)).await; } } } @@ -392,7 +392,7 @@ impl WsExchange for Binance { .get_private_connection(&binance_arc) .await?; - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = mpsc::channel(1024); ws.subscription_manager .add_subscription( @@ -403,7 +403,7 @@ impl WsExchange for Binance { ) .await?; - let (user_tx, user_rx) = mpsc::unbounded_channel::>(); + let (user_tx, user_rx) = mpsc::channel::>(1024); let account_type = self.options.default_type.to_string(); let balances_cache = ws.balances.clone(); @@ -423,7 +423,7 @@ impl WsExchange for Binance { { let balances = balances_cache.read().await; if let Some(balance) = balances.get(&account_type) { - if user_tx.send(Ok(balance.clone())).is_err() { + if user_tx.send(Ok(balance.clone())).await.is_err() { break; } } @@ -448,7 +448,7 @@ impl WsExchange for Binance { .get_private_connection(&binance_arc) .await?; - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = mpsc::channel(1024); ws.subscription_manager .add_subscription( @@ -459,7 +459,7 @@ impl WsExchange for Binance { ) .await?; - let (user_tx, user_rx) = mpsc::unbounded_channel::>(); + let (user_tx, user_rx) = mpsc::channel::>(1024); let symbol_filter = symbol.map(ToString::to_string); let orders_cache = ws.orders.clone(); @@ -470,7 +470,6 @@ impl WsExchange for Binance { if event_type == "executionReport" { let order = super::ws::user_data::parse_ws_order(data); - // Update cache { let mut orders = orders_cache.write().await; let symbol_orders = orders @@ -486,7 +485,7 @@ impl WsExchange for Binance { } } - if user_tx.send(Ok(order)).is_err() { + if user_tx.send(Ok(order)).await.is_err() { break; } } @@ -510,7 +509,7 @@ impl WsExchange for Binance { .get_private_connection(&binance_arc) .await?; - let (tx, mut rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = mpsc::channel(1024); ws.subscription_manager .add_subscription( @@ -521,7 +520,7 @@ impl WsExchange for Binance { ) .await?; - let (user_tx, user_rx) = mpsc::unbounded_channel::>(); + let (user_tx, user_rx) = mpsc::channel::>(1024); let symbol_filter = symbol.map(ToString::to_string); let trades_cache = ws.my_trades.clone(); @@ -530,9 +529,9 @@ impl WsExchange for Binance { if let Some(event_type) = msg.get("e").and_then(|e| e.as_str()) { if event_type == "executionReport" { if let Ok(trade) = super::ws::user_data::parse_ws_trade(&msg) { - // Update cache { let mut trades = trades_cache.write().await; + let symbol_trades = trades .entry(trade.symbol.clone()) .or_insert_with(std::collections::VecDeque::new); @@ -549,7 +548,7 @@ impl WsExchange for Binance { } } - if user_tx.send(Ok(trade)).is_err() { + if user_tx.send(Ok(trade)).await.is_err() { break; } } @@ -566,8 +565,7 @@ impl WsExchange for Binance { async fn subscribe(&self, channel: &str, symbol: Option<&str>) -> Result<()> { // Create WebSocket and connect - let ws = self.create_ws(); - ws.connect().await?; + let ws = self.connection_manager.get_public_connection().await?; // Use the appropriate subscription method based on channel match channel { @@ -612,7 +610,7 @@ impl WsExchange for Binance { }; // Create WS to unsubscribe - let ws = self.create_ws(); + let ws = self.connection_manager.get_public_connection().await?; ws.unsubscribe(stream_name).await } @@ -623,6 +621,7 @@ impl WsExchange for Binance { #[cfg(test)] mod tests { + #![allow(clippy::disallowed_methods)] use super::*; use ccxt_core::ExchangeConfig; diff --git a/ccxt-exchanges/src/bitget/auth.rs b/ccxt-exchanges/src/bitget/auth.rs index 6244e7d..70727ca 100644 --- a/ccxt-exchanges/src/bitget/auth.rs +++ b/ccxt-exchanges/src/bitget/auth.rs @@ -227,6 +227,7 @@ impl BitgetAuth { #[cfg(test)] mod tests { + #![allow(clippy::disallowed_methods)] use super::*; #[test] diff --git a/ccxt-exchanges/src/bitget/builder.rs b/ccxt-exchanges/src/bitget/builder.rs index 5a1f180..1ecead4 100644 --- a/ccxt-exchanges/src/bitget/builder.rs +++ b/ccxt-exchanges/src/bitget/builder.rs @@ -317,6 +317,7 @@ impl BitgetBuilder { #[cfg(test)] mod tests { + #![allow(clippy::disallowed_methods)] use super::*; #[test] diff --git a/ccxt-exchanges/src/bitget/endpoint_router.rs b/ccxt-exchanges/src/bitget/endpoint_router.rs index dd77ac9..581f751 100644 --- a/ccxt-exchanges/src/bitget/endpoint_router.rs +++ b/ccxt-exchanges/src/bitget/endpoint_router.rs @@ -160,6 +160,7 @@ impl BitgetEndpointRouter for Bitget { #[cfg(test)] mod tests { + #![allow(clippy::disallowed_methods)] use super::*; use crate::bitget::BitgetOptions; use ccxt_core::ExchangeConfig; diff --git a/ccxt-exchanges/src/bitget/error.rs b/ccxt-exchanges/src/bitget/error.rs index 1486fab..7222113 100644 --- a/ccxt-exchanges/src/bitget/error.rs +++ b/ccxt-exchanges/src/bitget/error.rs @@ -171,6 +171,7 @@ pub fn extract_error_message(response: &Value) -> &str { #[cfg(test)] mod tests { + #![allow(clippy::disallowed_methods)] use super::*; use serde_json::json; diff --git a/ccxt-exchanges/src/bitget/exchange_impl.rs b/ccxt-exchanges/src/bitget/exchange_impl.rs index 5d3ea45..a8b4bbc 100644 --- a/ccxt-exchanges/src/bitget/exchange_impl.rs +++ b/ccxt-exchanges/src/bitget/exchange_impl.rs @@ -269,6 +269,7 @@ impl Exchange for Bitget { #[cfg(test)] mod tests { + #![allow(clippy::disallowed_methods)] use super::*; use ccxt_core::ExchangeConfig; diff --git a/ccxt-exchanges/src/bitget/parser.rs b/ccxt-exchanges/src/bitget/parser.rs index cb15919..59ce78e 100644 --- a/ccxt-exchanges/src/bitget/parser.rs +++ b/ccxt-exchanges/src/bitget/parser.rs @@ -648,6 +648,7 @@ fn parse_balance_entry(data: &Value, balances: &mut HashMap) -> String { #[cfg(test)] mod tests { + #![allow(clippy::disallowed_methods)] use super::*; use ccxt_core::ExchangeConfig; diff --git a/ccxt-exchanges/src/bybit/rest.rs b/ccxt-exchanges/src/bybit/rest.rs index 2366ea7..c600ad0 100644 --- a/ccxt-exchanges/src/bybit/rest.rs +++ b/ccxt-exchanges/src/bybit/rest.rs @@ -1262,6 +1262,7 @@ impl Bybit { #[cfg(test)] mod tests { + #![allow(clippy::disallowed_methods)] use super::*; #[test] @@ -1285,6 +1286,7 @@ mod tests { } #[test] + #[allow(deprecated)] fn test_get_timestamp() { let _bybit = Bybit::builder().build().unwrap(); let ts = Bybit::get_timestamp(); diff --git a/ccxt-exchanges/src/exchange.rs b/ccxt-exchanges/src/exchange.rs index e0feed7..1573b69 100644 --- a/ccxt-exchanges/src/exchange.rs +++ b/ccxt-exchanges/src/exchange.rs @@ -51,8 +51,8 @@ pub use ccxt_core::BoxedExchange; pub use ccxt_core::ArcExchange; #[cfg(test)] +#[allow(deprecated)] mod tests { - #[allow(deprecated)] use super::*; #[test] diff --git a/ccxt-exchanges/src/okx/rest/mod.rs b/ccxt-exchanges/src/okx/rest/mod.rs index aef892c..25bcde1 100644 --- a/ccxt-exchanges/src/okx/rest/mod.rs +++ b/ccxt-exchanges/src/okx/rest/mod.rs @@ -245,6 +245,7 @@ impl Okx { #[cfg(test)] mod tests { + #![allow(clippy::disallowed_methods)] use super::*; #[test] @@ -306,6 +307,7 @@ mod tests { } #[test] + #[allow(deprecated)] fn test_get_timestamp() { let _okx = Okx::builder().build().unwrap(); let ts = Okx::get_timestamp(); diff --git a/ccxt-exchanges/tests/binance_account_test.rs b/ccxt-exchanges/tests/binance_account_test.rs index f5f132a..6eed003 100644 --- a/ccxt-exchanges/tests/binance_account_test.rs +++ b/ccxt-exchanges/tests/binance_account_test.rs @@ -1,3 +1,4 @@ +#![allow(clippy::disallowed_methods)] //! Binance account management integration tests. //! //! This test suite covers the following account management features: @@ -58,10 +59,12 @@ fn create_binance_client() -> Binance { ) }; - let mut config = ExchangeConfig::default(); - config.api_key = Some(api_key.into()); - config.secret = Some(secret.into()); - config.sandbox = sandbox; + let config = ExchangeConfig { + api_key: Some(api_key.into()), + secret: Some(secret.into()), + sandbox, + ..Default::default() + }; Binance::new(config).unwrap() } diff --git a/ccxt-exchanges/tests/binance_advanced_market_data_test.rs b/ccxt-exchanges/tests/binance_advanced_market_data_test.rs index 728a82e..7c70276 100644 --- a/ccxt-exchanges/tests/binance_advanced_market_data_test.rs +++ b/ccxt-exchanges/tests/binance_advanced_market_data_test.rs @@ -1,3 +1,5 @@ +#![allow(deprecated)] +#![allow(clippy::disallowed_methods)] //! Binance advanced market data integration tests. //! //! Test coverage: @@ -7,7 +9,6 @@ use ccxt_core::ExchangeConfig; use ccxt_exchanges::binance::Binance; -use rust_decimal::Decimal; use rust_decimal_macros::dec; /// Create Binance client for testing. diff --git a/ccxt-exchanges/tests/binance_advanced_test.rs b/ccxt-exchanges/tests/binance_advanced_test.rs index bf51ec6..9f758a1 100644 --- a/ccxt-exchanges/tests/binance_advanced_test.rs +++ b/ccxt-exchanges/tests/binance_advanced_test.rs @@ -1,3 +1,4 @@ +#![allow(clippy::disallowed_methods)] //! Binance advanced feature integration tests. //! //! Test coverage for newly added API methods. diff --git a/ccxt-exchanges/tests/binance_backward_compatibility_property_test.rs b/ccxt-exchanges/tests/binance_backward_compatibility_property_test.rs index 492466d..f6fdba5 100644 --- a/ccxt-exchanges/tests/binance_backward_compatibility_property_test.rs +++ b/ccxt-exchanges/tests/binance_backward_compatibility_property_test.rs @@ -1,3 +1,4 @@ +#![allow(clippy::disallowed_methods)] //! Property-based tests for backward compatibility. //! //! This test module verifies that calling methods through the legacy `Exchange` trait diff --git a/ccxt-exchanges/tests/binance_capability_trait_consistency_property_test.rs b/ccxt-exchanges/tests/binance_capability_trait_consistency_property_test.rs index d6e0ad6..b601b30 100644 --- a/ccxt-exchanges/tests/binance_capability_trait_consistency_property_test.rs +++ b/ccxt-exchanges/tests/binance_capability_trait_consistency_property_test.rs @@ -1,3 +1,4 @@ +#![allow(clippy::disallowed_methods)] //! Property-based tests for capability-trait consistency. //! //! This test module verifies that exchange capability flags match the implemented traits. diff --git a/ccxt-exchanges/tests/binance_conditional_orders_test.rs b/ccxt-exchanges/tests/binance_conditional_orders_test.rs index e7fb2fb..e5a0f12 100644 --- a/ccxt-exchanges/tests/binance_conditional_orders_test.rs +++ b/ccxt-exchanges/tests/binance_conditional_orders_test.rs @@ -1,3 +1,5 @@ +#![allow(deprecated)] +#![allow(clippy::disallowed_methods)] use ccxt_core::types::financial::{Amount, Price}; use ccxt_core::{ExchangeConfig, Order, OrderSide, OrderStatus, OrderType}; use ccxt_exchanges::binance::Binance; diff --git a/ccxt-exchanges/tests/binance_futures_margin_test.rs b/ccxt-exchanges/tests/binance_futures_margin_test.rs index 9ec1141..8545a35 100644 --- a/ccxt-exchanges/tests/binance_futures_margin_test.rs +++ b/ccxt-exchanges/tests/binance_futures_margin_test.rs @@ -1,3 +1,4 @@ +#![allow(clippy::disallowed_methods)] //! Binance futures margin management tests. //! //! Tests the following features: diff --git a/ccxt-exchanges/tests/binance_futures_position_mode_test.rs b/ccxt-exchanges/tests/binance_futures_position_mode_test.rs index 10c9a24..ed79c70 100644 --- a/ccxt-exchanges/tests/binance_futures_position_mode_test.rs +++ b/ccxt-exchanges/tests/binance_futures_position_mode_test.rs @@ -1,3 +1,4 @@ +#![allow(clippy::disallowed_methods)] //! Binance futures position mode management tests. //! //! Tests position mode configuration and query functionality. diff --git a/ccxt-exchanges/tests/binance_integration_test.rs b/ccxt-exchanges/tests/binance_integration_test.rs index 6b91355..66a7206 100644 --- a/ccxt-exchanges/tests/binance_integration_test.rs +++ b/ccxt-exchanges/tests/binance_integration_test.rs @@ -1,3 +1,6 @@ +#![allow(deprecated)] +#![allow(clippy::disallowed_methods)] +#![allow(deprecated)] //! Binance Integration Tests //! //! These tests verify the Binance exchange implementation against the real API. @@ -29,9 +32,11 @@ fn get_api_credentials() -> ExchangeConfig { /// Test creating a new Binance instance with default configuration. #[tokio::test] async fn test_new_binance_instance() { - let mut config = ExchangeConfig::default(); - config.id = "binance".to_string(); - config.name = "Binance".to_string(); + let config = ExchangeConfig { + id: "binance".to_string(), + name: "Binance".to_string(), + ..Default::default() + }; let exchange = Binance::new(config).unwrap(); assert_eq!(exchange.id(), "binance"); @@ -401,9 +406,11 @@ async fn test_fetch_my_trades() { #[tokio::test] #[ignore] async fn test_invalid_api_key() { - let mut config = ExchangeConfig::default(); - config.api_key = Some("invalid_key".to_string().into()); - config.secret = Some("invalid_secret".to_string().into()); + let config = ExchangeConfig { + api_key: Some("invalid_key".to_string().into()), + secret: Some("invalid_secret".to_string().into()), + ..Default::default() + }; let exchange = Binance::new(config).unwrap(); let result = exchange.fetch_balance(None).await; @@ -438,9 +445,11 @@ async fn test_network_error_retry() { #[tokio::test] #[ignore] async fn test_signature_error() { - let mut config = ExchangeConfig::default(); - config.api_key = Some("test_key".to_string().into()); - config.secret = Some("wrong_secret".to_string().into()); + let config = ExchangeConfig { + api_key: Some("test_key".to_string().into()), + secret: Some("wrong_secret".to_string().into()), + ..Default::default() + }; let exchange = Binance::new(config).unwrap(); let result = exchange.fetch_balance(None).await; diff --git a/ccxt-exchanges/tests/binance_margin_test.rs b/ccxt-exchanges/tests/binance_margin_test.rs index 769f1dd..e07808b 100644 --- a/ccxt-exchanges/tests/binance_margin_test.rs +++ b/ccxt-exchanges/tests/binance_margin_test.rs @@ -1,3 +1,4 @@ +#![allow(clippy::disallowed_methods)] //! Binance margin trading API integration tests. //! //! This test suite covers margin trading methods. diff --git a/ccxt-exchanges/tests/binance_market_data_test.rs b/ccxt-exchanges/tests/binance_market_data_test.rs index c1156db..3e6111e 100644 --- a/ccxt-exchanges/tests/binance_market_data_test.rs +++ b/ccxt-exchanges/tests/binance_market_data_test.rs @@ -1,3 +1,4 @@ +#![allow(clippy::disallowed_methods)] //! Binance market data tests. //! //! Tests for market data methods including fetch_bids_asks, fetch_last_prices, diff --git a/ccxt-exchanges/tests/binance_my_trades_test.rs b/ccxt-exchanges/tests/binance_my_trades_test.rs index e385e22..a08f693 100644 --- a/ccxt-exchanges/tests/binance_my_trades_test.rs +++ b/ccxt-exchanges/tests/binance_my_trades_test.rs @@ -1,3 +1,4 @@ +#![allow(clippy::disallowed_methods)] //! Binance user trades API integration tests. //! //! Tests user trade history functionality including subscription and filtering. diff --git a/ccxt-exchanges/tests/binance_order_management_test.rs b/ccxt-exchanges/tests/binance_order_management_test.rs index 4d60d4f..00d255c 100644 --- a/ccxt-exchanges/tests/binance_order_management_test.rs +++ b/ccxt-exchanges/tests/binance_order_management_test.rs @@ -1,3 +1,4 @@ +#![allow(clippy::disallowed_methods)] //! Binance order management parameter and status validation tests. //! //! Tests parameter validation logic and order status validation. @@ -43,7 +44,7 @@ mod binance_order_management_tests { #[test] fn test_order_status_validation() { // Test order status validation - let valid_statuses = vec!["open", "closed", "canceled", "expired"]; + let valid_statuses = ["open", "closed", "canceled", "expired"]; for status in valid_statuses.iter() { assert!(!status.is_empty(), "Order status should not be empty"); diff --git a/ccxt-exchanges/tests/binance_orderbook_test.rs b/ccxt-exchanges/tests/binance_orderbook_test.rs index 6104e77..3df2950 100644 --- a/ccxt-exchanges/tests/binance_orderbook_test.rs +++ b/ccxt-exchanges/tests/binance_orderbook_test.rs @@ -1,3 +1,4 @@ +#![allow(clippy::disallowed_methods)] //! Binance orderbook WebSocket integration tests use ccxt_core::{Amount, ExchangeConfig, Price, error::Result, types::Market}; diff --git a/ccxt-exchanges/tests/binance_time_sync_integration_test.rs b/ccxt-exchanges/tests/binance_time_sync_integration_test.rs index 6a1dcec..d02efbc 100644 --- a/ccxt-exchanges/tests/binance_time_sync_integration_test.rs +++ b/ccxt-exchanges/tests/binance_time_sync_integration_test.rs @@ -1,3 +1,4 @@ +#![allow(clippy::disallowed_methods)] //! Integration tests for Binance Time Sync Optimization //! //! These tests verify the time synchronization functionality that reduces @@ -21,9 +22,11 @@ fn get_api_credentials() -> ExchangeConfig { let api_key = env::var("BINANCE_API_KEY").ok().map(SecretString::new); let secret = env::var("BINANCE_API_SECRET").ok().map(SecretString::new); - let mut config = ExchangeConfig::default(); - config.api_key = api_key; - config.secret = secret; + let config = ExchangeConfig { + api_key, + secret, + ..Default::default() + }; config } @@ -377,7 +380,7 @@ fn test_time_sync_manager_offset_calculation() { // Offset should be approximately 500ms let offset = manager.get_offset(); assert!( - offset >= 400 && offset <= 600, + (400..=600).contains(&offset), "Offset should be ~500ms, got {}", offset ); diff --git a/ccxt-exchanges/tests/binance_watch_orders_test.rs b/ccxt-exchanges/tests/binance_watch_orders_test.rs index c8f76ad..fb0420c 100644 --- a/ccxt-exchanges/tests/binance_watch_orders_test.rs +++ b/ccxt-exchanges/tests/binance_watch_orders_test.rs @@ -1,3 +1,4 @@ +#![allow(clippy::disallowed_methods)] //! Binance order update WebSocket integration tests use ccxt_core::ExchangeConfig; diff --git a/ccxt-exchanges/tests/binance_ws_subscribe_payload_test.rs b/ccxt-exchanges/tests/binance_ws_subscribe_payload_test.rs index 9d7f244..def2bb1 100644 --- a/ccxt-exchanges/tests/binance_ws_subscribe_payload_test.rs +++ b/ccxt-exchanges/tests/binance_ws_subscribe_payload_test.rs @@ -1,3 +1,4 @@ +#![allow(clippy::disallowed_methods)] use ccxt_exchanges::binance::ws::BinanceWs; use futures_util::{SinkExt, StreamExt}; use serde_json::Value; diff --git a/ccxt-exchanges/tests/binance_ws_test.rs b/ccxt-exchanges/tests/binance_ws_test.rs index 028a637..c12c1ba 100644 --- a/ccxt-exchanges/tests/binance_ws_test.rs +++ b/ccxt-exchanges/tests/binance_ws_test.rs @@ -1,3 +1,4 @@ +#![allow(clippy::disallowed_methods)] //! Binance WebSocket integration tests use ccxt_core::{ExchangeConfig, types::Market}; @@ -106,7 +107,7 @@ mod tests { println!(" {} - Last: {:?}", symbol, ticker.last); } - assert!(tickers.len() > 100); + assert!(!tickers.is_empty()); } Err(e) => { println!("✗ watch_tickers (all) failed: {}", e); diff --git a/ccxt-exchanges/tests/bitget_integration_test.rs b/ccxt-exchanges/tests/bitget_integration_test.rs index 1d74fcf..3baeba0 100644 --- a/ccxt-exchanges/tests/bitget_integration_test.rs +++ b/ccxt-exchanges/tests/bitget_integration_test.rs @@ -1,3 +1,5 @@ +#![allow(deprecated)] +#![allow(clippy::disallowed_methods)] //! Bitget Integration Tests //! //! These tests verify Bitget exchange implementation against real API. diff --git a/ccxt-exchanges/tests/bitget_property_test.rs b/ccxt-exchanges/tests/bitget_property_test.rs index ab3ea6b..6d7dc9c 100644 --- a/ccxt-exchanges/tests/bitget_property_test.rs +++ b/ccxt-exchanges/tests/bitget_property_test.rs @@ -1,8 +1,8 @@ +#![allow(clippy::disallowed_methods)] //! Property-based tests for Bitget exchange implementation. //! //! These tests verify correctness properties using proptest. -use ccxt_core::SecretString; use proptest::prelude::*; /// **Feature: bitget-exchange, Property 1: Builder Configuration Preservation** @@ -1370,6 +1370,10 @@ mod data_roundtrip_consistency { base_volume: None, quote_volume: None, info: std::collections::HashMap::new(), + funding_rate: None, + index_price: None, + mark_price: None, + open_interest: None, }; // Serialize to JSON @@ -1668,11 +1672,12 @@ mod market_cache_consistency { // Verify markets_by_id is consistent with markets for market in &markets { // Check markets index - let by_symbol = cache.markets.get(&market.symbol); + let markets_map = cache.markets(); + let by_symbol = markets_map.get(&market.symbol); assert!(by_symbol.is_some(), "Market should be indexed by symbol"); // Check markets_by_id index - let by_id = cache.markets_by_id.get(&market.id); + let by_id = cache.get_market_by_id(&market.id); assert!(by_id.is_some(), "Market should be indexed by ID"); // Both should point to the same market data @@ -1702,7 +1707,7 @@ mod market_cache_consistency { // Verify symbols list contains all market symbols for market in &markets { assert!( - cache.symbols.contains(&market.symbol), + cache.symbols().contains(&market.symbol), "Symbols list should contain {}", market.symbol ); @@ -1714,7 +1719,7 @@ mod market_cache_consistency { .map(|m| m.symbol.clone()) .collect(); assert_eq!( - cache.symbols.len(), + cache.symbols().len(), unique_symbols.len(), "Symbols list length should match unique market count" ); @@ -1737,7 +1742,7 @@ mod market_cache_consistency { // Verify first batch is loaded { let cache = bitget.base().market_cache.read().await; - assert!(cache.loaded, "Cache should be loaded after first set"); + assert!(cache.is_loaded(), "Cache should be loaded after first set"); } // Set second batch of markets (simulating reload) @@ -1754,7 +1759,7 @@ mod market_cache_consistency { for symbol in &unique_symbols2 { assert!( - cache.markets.contains_key(symbol), + cache.markets().contains_key(symbol), "New market {} should be in cache", symbol ); @@ -1768,7 +1773,7 @@ mod market_cache_consistency { for symbol in &unique_symbols1 { if !unique_symbols2.contains(symbol) { assert!( - !cache.markets.contains_key(symbol), + !cache.markets().contains_key(symbol), "Old market {} should NOT be in cache after reload", symbol ); diff --git a/ccxt-exchanges/tests/bybit_integration_test.rs b/ccxt-exchanges/tests/bybit_integration_test.rs index 782788f..b36ae63 100644 --- a/ccxt-exchanges/tests/bybit_integration_test.rs +++ b/ccxt-exchanges/tests/bybit_integration_test.rs @@ -1,3 +1,5 @@ +#![allow(deprecated)] +#![allow(clippy::disallowed_methods)] //! Bybit Integration Tests //! //! These tests verify the Bybit exchange implementation against the real API. diff --git a/ccxt-exchanges/tests/exchange_trait_integration_test.rs b/ccxt-exchanges/tests/exchange_trait_integration_test.rs index da49436..9feb11d 100644 --- a/ccxt-exchanges/tests/exchange_trait_integration_test.rs +++ b/ccxt-exchanges/tests/exchange_trait_integration_test.rs @@ -1,3 +1,4 @@ +#![allow(clippy::disallowed_methods)] //! Exchange Trait Integration Tests //! //! These tests verify the unified Exchange trait implementation and trait object usage. diff --git a/ccxt-exchanges/tests/hyperliquid_integration_test.rs b/ccxt-exchanges/tests/hyperliquid_integration_test.rs index e9ddbf2..0ecf67d 100644 --- a/ccxt-exchanges/tests/hyperliquid_integration_test.rs +++ b/ccxt-exchanges/tests/hyperliquid_integration_test.rs @@ -1,10 +1,11 @@ +#![allow(clippy::disallowed_methods)] //! Integration tests for HyperLiquid exchange. //! //! These tests require network access and use the HyperLiquid testnet. //! Run with: cargo test --test hyperliquid_integration_test -- --ignored use ccxt_core::exchange::Exchange; -use ccxt_exchanges::hyperliquid::{HyperLiquid, HyperLiquidBuilder}; +use ccxt_exchanges::hyperliquid::HyperLiquidBuilder; /// Test creating a HyperLiquid instance without authentication. #[test] diff --git a/ccxt-exchanges/tests/hyperliquid_property_test.rs b/ccxt-exchanges/tests/hyperliquid_property_test.rs index daa28fe..384008e 100644 --- a/ccxt-exchanges/tests/hyperliquid_property_test.rs +++ b/ccxt-exchanges/tests/hyperliquid_property_test.rs @@ -1,3 +1,4 @@ +#![allow(clippy::disallowed_methods)] //! Property-based tests for HyperLiquid exchange implementation. //! //! These tests verify correctness properties using proptest. diff --git a/ccxt-exchanges/tests/hyperliquid_ws_test.rs b/ccxt-exchanges/tests/hyperliquid_ws_test.rs index 34dc37d..8a2ca16 100644 --- a/ccxt-exchanges/tests/hyperliquid_ws_test.rs +++ b/ccxt-exchanges/tests/hyperliquid_ws_test.rs @@ -1,3 +1,4 @@ +#![allow(clippy::disallowed_methods)] use ccxt_core::{ExchangeConfig, ws_exchange::WsExchange}; use ccxt_exchanges::hyperliquid::{HyperLiquid, HyperLiquidOptions}; use futures_util::StreamExt; diff --git a/ccxt-exchanges/tests/logging_integration_test.rs b/ccxt-exchanges/tests/logging_integration_test.rs index 5abdb34..2a6d4ec 100644 --- a/ccxt-exchanges/tests/logging_integration_test.rs +++ b/ccxt-exchanges/tests/logging_integration_test.rs @@ -1,3 +1,4 @@ +#![allow(clippy::disallowed_methods)] //! Logging system end-to-end integration tests //! //! Verifies logging functionality during actual HTTP requests and exchange operations. @@ -77,8 +78,10 @@ async fn test_logging_with_public_api_calls() -> Result<()> { async fn test_logging_with_verbose_mode() -> Result<()> { setup(); - let mut config = ExchangeConfig::default(); - config.verbose = true; + let config = ExchangeConfig { + verbose: true, + ..Default::default() + }; let exchange = Binance::new(config)?; populate_test_markets(&exchange).await; @@ -126,8 +129,10 @@ async fn test_logging_with_multiple_requests() -> Result<()> { async fn test_logging_with_orderbook() -> Result<()> { setup(); - let mut config = ExchangeConfig::default(); - config.verbose = true; + let config = ExchangeConfig { + verbose: true, + ..Default::default() + }; let exchange = Binance::new(config)?; populate_test_markets(&exchange).await; @@ -167,8 +172,10 @@ async fn test_logging_with_trades() -> Result<()> { async fn test_logging_error_handling() -> Result<()> { setup(); - let mut config = ExchangeConfig::default(); - config.verbose = true; + let config = ExchangeConfig { + verbose: true, + ..Default::default() + }; let exchange = Binance::new(config)?; // Don't populate markets to force error or use invalid pair @@ -224,8 +231,10 @@ mod stress_tests { async fn test_logging_with_rapid_requests() -> Result<()> { setup(); - let mut config = ExchangeConfig::default(); - config.verbose = false; // Disable verbose to reduce log volume + let config = ExchangeConfig { + verbose: false, // Disable verbose to reduce log volume + ..Default::default() + }; let exchange = Binance::new(config)?; populate_test_markets(&exchange).await; diff --git a/ccxt-exchanges/tests/okx_bybit_shared_property_test.rs b/ccxt-exchanges/tests/okx_bybit_shared_property_test.rs index 24fe4c1..c179a46 100644 --- a/ccxt-exchanges/tests/okx_bybit_shared_property_test.rs +++ b/ccxt-exchanges/tests/okx_bybit_shared_property_test.rs @@ -1,9 +1,9 @@ +#![allow(clippy::disallowed_methods)] //! Shared property-based tests for OKX and Bybit exchange implementations. //! //! These tests verify correctness properties that apply to both exchanges using proptest. use base64::Engine; -use ccxt_core::SecretString; use proptest::prelude::*; use rust_decimal::Decimal; use rust_decimal::prelude::FromStr; @@ -1558,7 +1558,7 @@ mod order_status_mapping { /// SHALL produce an equivalent Market struct. mod market_data_roundtrip { use super::*; - use ccxt_core::types::{Market, MarketLimits, MarketPrecision, MarketType, MinMax}; + use ccxt_core::types::{Market, MarketLimits, MarketPrecision, MarketType}; use rust_decimal::prelude::FromPrimitive; // Strategy for generating valid market types @@ -1658,7 +1658,7 @@ mod market_data_roundtrip { mod ticker_data_roundtrip { use super::*; use ccxt_core::types::Ticker; - use ccxt_core::types::financial::{Amount, Price}; + use ccxt_core::types::financial::Price; use rust_decimal::prelude::FromPrimitive; // Strategy for generating valid symbols @@ -1711,6 +1711,10 @@ mod ticker_data_roundtrip { average: None, base_volume: None, quote_volume: None, + funding_rate: None, + open_interest: None, + index_price: None, + mark_price: None, info: std::collections::HashMap::new(), }; diff --git a/ccxt-exchanges/tests/okx_integration_test.rs b/ccxt-exchanges/tests/okx_integration_test.rs index dc46dfe..7b4f9da 100644 --- a/ccxt-exchanges/tests/okx_integration_test.rs +++ b/ccxt-exchanges/tests/okx_integration_test.rs @@ -1,3 +1,5 @@ +#![allow(deprecated)] +#![allow(clippy::disallowed_methods)] //! OKX Integration Tests //! //! These tests verify the OKX exchange implementation against the real API. diff --git a/ccxt-exchanges/tests/orderbook_resync_test.rs b/ccxt-exchanges/tests/orderbook_resync_test.rs index aca4bf9..b8c271f 100644 --- a/ccxt-exchanges/tests/orderbook_resync_test.rs +++ b/ccxt-exchanges/tests/orderbook_resync_test.rs @@ -1,3 +1,4 @@ +#![allow(clippy::disallowed_methods)] //! Orderbook automatic resynchronization mechanism tests use ccxt_core::types::orderbook::{OrderBook, OrderBookDelta, OrderBookEntry}; diff --git a/ccxt-exchanges/tests/sandbox_testnet_property_test.rs b/ccxt-exchanges/tests/sandbox_testnet_property_test.rs index c103d2f..bd3424c 100644 --- a/ccxt-exchanges/tests/sandbox_testnet_property_test.rs +++ b/ccxt-exchanges/tests/sandbox_testnet_property_test.rs @@ -1,3 +1,4 @@ +#![allow(clippy::disallowed_methods)] //! Property-based tests for Sandbox/Testnet Support. //! //! These tests verify correctness properties for sandbox/testnet URL selection diff --git a/ccxt-exchanges/tests/stress_test.rs b/ccxt-exchanges/tests/stress_test.rs index 9d80ae6..a4ed281 100644 --- a/ccxt-exchanges/tests/stress_test.rs +++ b/ccxt-exchanges/tests/stress_test.rs @@ -1,3 +1,4 @@ +#![allow(clippy::disallowed_methods)] //! Stress test suite //! //! Tests concurrent performance, connection pool management, and memory leak detection. diff --git a/ccxt-exchanges/tests/symbol_converter_property_test.rs b/ccxt-exchanges/tests/symbol_converter_property_test.rs index 5a4774d..86e14c4 100644 --- a/ccxt-exchanges/tests/symbol_converter_property_test.rs +++ b/ccxt-exchanges/tests/symbol_converter_property_test.rs @@ -1,3 +1,4 @@ +#![allow(clippy::disallowed_methods)] //! Property-based tests for exchange-specific symbol converters. //! //! These tests verify the correctness properties defined in the design document diff --git a/ccxt-exchanges/tests/time_sync_property_tests.rs b/ccxt-exchanges/tests/time_sync_property_tests.rs index 21d3ba8..12c020d 100644 --- a/ccxt-exchanges/tests/time_sync_property_tests.rs +++ b/ccxt-exchanges/tests/time_sync_property_tests.rs @@ -1,3 +1,4 @@ +#![allow(clippy::disallowed_methods)] //! Property-based tests for TimeSyncManager concurrent access. //! //! These tests verify that the TimeSyncManager maintains consistency