feat(hyperliquid): implement WebSocket functionality for HyperLiquid …

README.md

@@ -512,8 +512,16 @@ This project is licensed under the MIT License - see the [LICENSE](LICENSE) file
 - **Discussions**: [GitHub Discussions](https://github.com/Praying/ccxt-rust/discussions)
 - **Documentation**: [Project Docs](docs/)
 
+## ❤️ Donations
+
+If the project is profitable, consider donating as showing gratitude for its development:
+
+**USDT or USDC Binance Smart Chain BEP20**:
+
+`0x8e5d858f92938b028065d39450421d0e080d15f7`
+
 ---
 
 **Status**: 🚧 Active Development | **Version**: 0.1.2 | **Updated**: 2026-01
 
-⚠️ **Note**: This library is under active development. APIs may change before v1.0. Not recommended for production use yet.
+⚠️ **Note**: This library is under active development. APIs may change before v1.0. Not recommended for production use yet.

ccxt-core/Cargo.toml

@@ -71,6 +71,8 @@ zeroize = { workspace = true }
 # Test configuration (optional, enabled by test-utils feature)
 dotenvy = { workspace = true, optional = true }
 envy = { workspace = true, optional = true }
+async-stream = "0.3.6"
+derive_more = { version = "2.1.1", features = ["debug"] }
 
 [dev-dependencies]
 tokio-test = { workspace = true }
@@ -112,4 +114,4 @@ backtrace = []
 
 [[bench]]
 name = "timestamp_benchmark"
-harness = false
+harness = false

ccxt-core/src/ws_client/mod.rs

@@ -24,6 +24,7 @@ pub use state::{WsConnectionState, WsStats, WsStatsSnapshot};
 pub use subscription::{Subscription, SubscriptionManager};
 
 use crate::error::{Error, Result};
+use derive_more::Debug;
 use futures_util::{SinkExt, StreamExt, stream::SplitSink};
 use serde_json::Value;
 use std::collections::HashMap;
@@ -53,6 +54,7 @@ type WsWriter = SplitSink<WebSocketStream<MaybeTlsStream<TcpStream>>, Message>;
 /// - `DropOldest`: Removes the oldest message to make room (default)
 /// - `DropNewest`: Discards the incoming message
 /// - `Block`: Waits until space is available (may stall the read loop)
+#[derive(Debug)]
 pub struct WsClient {
     config: WsConfig,
     state: Arc<AtomicU8>,
@@ -64,6 +66,7 @@ pub struct WsClient {
     shutdown_tx: Arc<Mutex<Option<mpsc::UnboundedSender<()>>>>,
     stats: Arc<WsStats>,
     cancel_token: Arc<Mutex<Option<CancellationToken>>>,
+    #[debug(skip)]
     event_callback: Arc<Mutex<Option<WsEventCallback>>>,
     /// Counter for dropped messages due to backpressure
     dropped_messages: Arc<AtomicU32>,

ccxt-core/src/ws_client/reconnect.rs

@@ -259,4 +259,31 @@ impl AutoReconnectCoordinator {
 
         info!("Auto-reconnect loop terminated");
     }
+
+    /// Convert into a stream (requires `tokio-stream`).
+    #[cfg(feature = "websocket")]
+    pub fn into_stream(self) -> impl futures::Stream<Item = serde_json::Value> {
+        let client = self.client.clone();
+        let coordinator = Arc::new(self);
+
+        async_stream::stream! {
+            // Start reconnection in background
+            coordinator.start().await;
+
+            loop {
+                if let Some(msg) = client.receive().await {
+                    yield msg;
+                } else {
+                    // Check if we should stop
+                    if !coordinator.is_enabled() {
+                        break;
+                    }
+                    // Wait a bit before retry if channel closed (shouldn't happen with reconnect)
+                    tokio::time::sleep(Duration::from_millis(100)).await;
+                }
+            }
+
+            coordinator.stop().await;
+        }
+    }
 }

ccxt-exchanges/Cargo.toml

@@ -55,6 +55,7 @@ futures-util = { workspace = true }
 
 # Security
 zeroize = { workspace = true }
+tokio-stream = "0.1.18"
 
 [dev-dependencies]
 tokio-test = { workspace = true }
@@ -98,4 +99,4 @@ full = ["binance", "websocket"]
 
 [[bench]]
 name = "binance_benchmark"
-harness = false
+harness = false

ccxt-exchanges/src/hyperliquid/mod.rs

@@ -73,6 +73,8 @@ pub struct HyperLiquid {
     options: HyperLiquidOptions,
     /// Authentication instance (optional, for private API).
     auth: Option<HyperLiquidAuth>,
+    /// Persistent WebSocket connection reference.
+    pub(crate) ws_connection: std::sync::Arc<tokio::sync::RwLock<Option<ws::HyperLiquidWs>>>,
 }
 
 /// HyperLiquid-specific options.
@@ -146,6 +148,7 @@ impl HyperLiquid {
             base,
             options,
             auth,
+            ws_connection: std::sync::Arc::new(tokio::sync::RwLock::new(None)),
         })
     }
 
@@ -289,12 +292,11 @@ impl HyperLiquid {
             .map(auth::HyperLiquidAuth::wallet_address)
     }
 
-    // TODO: Implement in task 11 (WebSocket Implementation)
-    // /// Creates a public WebSocket client.
-    // pub fn create_ws(&self) -> ws::HyperLiquidWs {
-    //     let urls = self.urls();
-    //     ws::HyperLiquidWs::new(urls.ws)
-    // }
+    /// Creates a WebSocket client for public data streams.
+    pub fn create_ws(&self) -> ws::HyperLiquidWs {
+        let urls = self.urls();
+        ws::HyperLiquidWs::new(urls.ws)
+    }
 }
 
 /// HyperLiquid API URLs.