Implement WebSocket transport layer with core types, protocol handling, and client features

- Added core types: RequestId, Topic, MessageKind, and WebSocketConfig.
- Defined ProtocolHandler trait with default implementations and documentation.
- Created PendingRequest and SubscriptionStore using lock-free scc::HashMap.
- Implemented ConnectionActor for managing WebSocket connections and state transitions.
- Developed WebSocketClient for handling requests, subscriptions, and message sending.
- Added JSON-based protocol handlers for generic and JSON-RPC protocols.
- Created mock server and comprehensive tests for stores, integration, and stress scenarios.
- Documented modules, examples, and updated README for clarity and usage guidance.

Author: Akagi201

Cargo.toml

@@ -3,7 +3,7 @@ members = ["crates/*"]
 resolver = "3"
 
 [workspace.package]
-version = "1.0.0"
+version = "1.1.0"
 edition = "2024"
 authors = ["Akagi201 <akagi201@gmail.com>"]
 license = "Apache-2.0"
@@ -182,8 +182,8 @@ all = "warn"
 
 [workspace.dependencies]
 # local crates
-fastwebsockets = { path = "crates/fastwebsockets", package = "hpx-fastwebsockets", version = "1.0.0" }
-hpx = { path = "crates/hpx", version = "1.0.0" }
+fastwebsockets = { path = "crates/fastwebsockets", package = "hpx-fastwebsockets", version = "1.1.0" }
+hpx = { path = "crates/hpx", version = "1.1.0" }
 
 # external crates
 ahash = "0.8.12"
@@ -192,7 +192,7 @@ async-trait = "0.1.80"
 axum = "0.8.8"
 axum-core = "0.5.6"
 base64 = "0.22.1"
-boring = { git = "https://github.com/cloudflare/boring.git", rev = "5f4cf54cc5a7f488a924f48ea06d334ac121e5ac", version = "5.0.0-alpha.3" }
+boring = { git = "https://github.com/cloudflare/boring.git", rev = "06ca1fd7461d3626746c2cd36d1ada07c85c952d", version = "5.0.0-alpha.3" }
 brotli = "8.0.2"
 bytes = "1.11.1"
 cookie = "0.18.1"
@@ -229,8 +229,8 @@ rand = "0.9.2"
 rustls = "0.23.36"
 rustls-pemfile = "2.2.0"
 rustls-pki-types = "1.14.0"
+scc = "3.5.5"
 schnellru = "0.2.4"
-scc = "2"
 serde = "1.0.228"
 serde_json = "1.0.149"
 serde_urlencoded = "0.7.1"
@@ -247,7 +247,7 @@ system-configuration = "0.7.0"
 tempfile = "3.17.1"
 thiserror = "2.0.18"
 tokio = "1.49.0"
-tokio-boring = { git = "https://github.com/cloudflare/boring.git", rev = "5f4cf54cc5a7f488a924f48ea06d334ac121e5ac", version = "5.0.0-alpha.3" }
+tokio-boring = { git = "https://github.com/cloudflare/boring.git", rev = "06ca1fd7461d3626746c2cd36d1ada07c85c952d", version = "5.0.0-alpha.3" }
 tokio-rustls = "0.26"
 tokio-socks = "0.5.2"
 tokio-test = "0.4.5"

crates/hpx-transport/Cargo.toml

@@ -29,9 +29,9 @@ opentelemetry-otlp = { workspace = true, features = ["grpc-tonic", "metrics"] }
 opentelemetry_sdk = { workspace = true, features = ["rt-tokio"] }
 parking_lot.workspace = true
 rand.workspace = true
+scc.workspace = true
 serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true }
-scc.workspace = true
 serde_urlencoded.workspace = true
 sha2.workspace = true
 thiserror.workspace = true

crates/hpx-transport/src/websocket/pending.rs

@@ -65,7 +65,7 @@ impl PendingRequestStore {
         };
 
         // Insert returns Err if key already exists
-        if self.requests.insert(id, pending).is_err() {
+        if self.requests.insert_sync(id, pending).is_err() {
             return None;
         }
 
@@ -76,7 +76,7 @@ impl PendingRequestStore {
     ///
     /// Returns `true` if the request was found and resolved, `false` otherwise.
     pub fn resolve(&self, id: &RequestId, response: TransportResult<String>) -> bool {
-        if let Some((_, pending)) = self.requests.remove(id) {
+        if let Some((_, pending)) = self.requests.remove_sync(id) {
             // Send the response; ignore error (receiver may have dropped)
             let _ = pending.response_tx.send(response);
             return true;
@@ -90,7 +90,7 @@ impl PendingRequestStore {
     pub fn cleanup_stale(&self) {
         let now = Instant::now();
         self.requests
-            .retain(|_, pending| now.duration_since(pending.created_at) < pending.timeout);
+            .retain_sync(|_, pending| now.duration_since(pending.created_at) < pending.timeout);
     }
 
     /// Clean up stale requests and notify them of timeout.
@@ -101,15 +101,16 @@ impl PendingRequestStore {
         let mut expired = Vec::new();
 
         // First, collect expired IDs
-        self.requests.scan(|id, pending| {
+        self.requests.retain_sync(|id, pending| {
             if now.duration_since(pending.created_at) >= pending.timeout {
                 expired.push((id.clone(), pending.timeout));
             }
+            true
         });
 
         // Then remove and notify each one
         for (id, timeout) in expired {
-            if let Some((_, pending)) = self.requests.remove(&id) {
+            if let Some((_, pending)) = self.requests.remove_sync(&id) {
                 let _ = pending
                     .response_tx
                     .send(Err(TransportError::request_timeout(
@@ -140,12 +141,13 @@ impl PendingRequestStore {
     /// This should be called on connection close to notify all waiters.
     pub fn clear_with_error(&self, error_message: &str) {
         let mut ids = Vec::new();
-        self.requests.scan(|id, _| {
+        self.requests.retain_sync(|id, _| {
             ids.push(id.clone());
+            true
         });
 
         for id in ids {
-            if let Some((_, pending)) = self.requests.remove(&id) {
+            if let Some((_, pending)) = self.requests.remove_sync(&id) {
                 let _ = pending
                     .response_tx
                     .send(Err(TransportError::connection_closed(Some(
@@ -157,7 +159,7 @@ impl PendingRequestStore {
 
     /// Clear all pending requests without notification.
     pub fn clear(&self) {
-        self.requests.clear();
+        self.requests.clear_sync();
     }
 }
 

crates/hpx-transport/src/websocket/subscription.rs

@@ -41,16 +41,11 @@ impl SubscriptionStore {
     /// Returns `(receiver, is_new)` where `is_new` is true if this is
     /// a new subscription that needs to be sent to the server.
     pub fn subscribe(&self, topic: Topic) -> (broadcast::Receiver<WsMessage>, bool) {
-        // Try to get existing subscription first
-        if let Some(entry) = self.subscriptions.get(&topic) {
-            let receiver = entry.get().sender.subscribe();
-            drop(entry); // Release read lock before modifying
-
-            // Increment ref count
-            let _ = self.subscriptions.update(&topic, |_, entry| {
-                entry.ref_count += 1;
-            });
-
+        // Try to get existing subscription and increment ref count atomically
+        if let Some(receiver) = self.subscriptions.update_sync(&topic, |_, entry| {
+            entry.ref_count += 1;
+            entry.sender.subscribe()
+        }) {
             return (receiver, false);
         }
 
@@ -62,18 +57,14 @@ impl SubscriptionStore {
         };
 
         // Insert; if another thread beat us, subscribe to their channel
-        if let Err((_, _entry)) = self.subscriptions.insert(topic.clone(), entry) {
+        if let Err((_, _entry)) = self.subscriptions.insert_sync(topic.clone(), entry) {
             // Race: another thread inserted first
             // The entry we tried to insert is returned on error
 
-            if let Some(existing) = self.subscriptions.get(&topic) {
-                let receiver = existing.get().sender.subscribe();
-                drop(existing);
-
-                let _ = self.subscriptions.update(&topic, |_, entry| {
-                    entry.ref_count += 1;
-                });
-
+            if let Some(receiver) = self.subscriptions.update_sync(&topic, |_, entry| {
+                entry.ref_count += 1;
+                entry.sender.subscribe()
+            }) {
                 return (receiver, false);
             }
         }
@@ -85,17 +76,10 @@ impl SubscriptionStore {
     ///
     /// Returns `Some(receiver)` if the topic exists, `None` otherwise.
     pub fn add_subscriber(&self, topic: &Topic) -> Option<broadcast::Receiver<WsMessage>> {
-        if let Some(entry) = self.subscriptions.get(topic) {
-            let receiver = entry.get().sender.subscribe();
-            drop(entry);
-
-            let _ = self.subscriptions.update(topic, |_, entry| {
-                entry.ref_count += 1;
-            });
-
-            return Some(receiver);
-        }
-        None
+        self.subscriptions.update_sync(topic, |_, entry| {
+            entry.ref_count += 1;
+            entry.sender.subscribe()
+        })
     }
 
     /// Unsubscribe from a topic.
@@ -105,15 +89,15 @@ impl SubscriptionStore {
     pub fn unsubscribe(&self, topic: &Topic) -> bool {
         let mut should_remove = false;
 
-        let _ = self.subscriptions.update(topic, |_, entry| {
+        let _ = self.subscriptions.update_sync(topic, |_, entry| {
             entry.ref_count = entry.ref_count.saturating_sub(1);
             if entry.ref_count == 0 {
                 should_remove = true;
             }
         });
 
         if should_remove {
-            self.subscriptions.remove(topic);
+            self.subscriptions.remove_sync(topic);
             return true;
         }
 
@@ -125,28 +109,28 @@ impl SubscriptionStore {
     /// Returns `true` if the topic exists (even if no active receivers),
     /// `false` if the topic doesn't exist.
     pub fn publish(&self, topic: &Topic, message: WsMessage) -> bool {
-        if let Some(entry) = self.subscriptions.get(topic) {
-            // Ignore send errors (no active receivers is fine)
-            let _ = entry.get().sender.send(message);
-            return true;
-        }
-        false
+        self.subscriptions
+            .update_sync(topic, |_, entry| {
+                // Ignore send errors (no active receivers is fine)
+                let _ = entry.sender.send(message.clone());
+            })
+            .is_some()
     }
 
     /// Get all currently subscribed topics.
     pub fn get_all_topics(&self) -> Vec<Topic> {
         let mut topics = Vec::new();
-        self.subscriptions.scan(|topic, _| {
+        self.subscriptions.retain_sync(|topic, _| {
             topics.push(topic.clone());
+            true
         });
         topics
     }
 
     /// Get the subscriber count for a topic.
     pub fn subscriber_count(&self, topic: &Topic) -> usize {
         self.subscriptions
-            .get(topic)
-            .map(|entry| entry.get().ref_count)
+            .update_sync(topic, |_, entry| entry.ref_count)
             .unwrap_or(0)
     }
 
@@ -162,7 +146,7 @@ impl SubscriptionStore {
 
     /// Clear all subscriptions.
     pub fn clear(&self) {
-        self.subscriptions.clear();
+        self.subscriptions.clear_sync();
     }
 }