feat: add WebSocket client support for WASM (#1)

.github/workflows/kdf.yaml

@@ -0,0 +1,24 @@
+name: "kdf"
+
+on:
+  push:
+    branches: ["kdf"]
+    paths-ignore:
+      - ".github/**"
+      - "docs/**"
+      - "README.md"
+
+  workflow_dispatch:
+
+jobs:
+  run-tests:
+    uses: ./.github/workflows/ci.yaml
+    secrets: inherit
+
+  release:
+    needs: [run-tests]
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+        with:
+          fetch-depth: 0

Cargo.toml

@@ -11,8 +11,9 @@ members = ["blockchain_api", "relay_client", "relay_rpc"]
 
 [features]
 default = ["full"]
-full = ["client", "rpc"]
+full = ["client", "rpc", "http"]
 client = ["dep:relay_client"]
+http = ["relay_client/http"]
 rpc = ["dep:relay_rpc"]
 
 [dependencies]
@@ -36,6 +37,10 @@ required-features = ["client", "rpc"]
 
 [[example]]
 name = "http_client"
+required-features = ["client", "rpc", "http"]
+
+[[example]]
+name = "ws_unmanaged"
 required-features = ["client", "rpc"]
 
 [[example]]

examples/websocket_client.rs

@@ -19,7 +19,7 @@ struct Args {
     address: String,
 
     /// Specify WalletConnect project ID.
-    #[structopt(short, long, default_value = "3cbaa32f8fbf3cdcc87d27ca1fa68069")]
+    #[structopt(short, long, default_value = "1979a8326eb123238e633655924f0a78")]
     project_id: String,
 }
 

examples/ws_unmanaged.rs

@@ -0,0 +1,140 @@
+use {
+    futures_util::StreamExt,
+    relay_client::{
+        websocket::{Client, Connection, ConnectionControl, PublishedMessage, StreamEvent},
+        ConnectionOptions,
+    },
+    relay_rpc::{
+        auth::{ed25519_dalek::SigningKey, AuthToken},
+        domain::Topic,
+    },
+    std::{sync::Arc, time::Duration},
+    structopt::StructOpt,
+    tokio::spawn,
+};
+
+#[derive(StructOpt)]
+struct Args {
+    /// Specify WebSocket address.
+    #[structopt(short, long, default_value = "wss://relay.walletconnect.org")]
+    address: String,
+
+    /// Specify WalletConnect project ID.
+    #[structopt(short, long, default_value = "86e916bcbacee7f98225dde86b697f5b")]
+    project_id: String,
+}
+
+fn create_conn_opts(address: &str, project_id: &str) -> ConnectionOptions {
+    let key = SigningKey::generate(&mut rand::thread_rng());
+
+    let auth = AuthToken::new("http://127.0.0.1:8000")
+        .aud(address)
+        .ttl(Duration::from_secs(60 * 60))
+        .as_jwt(&key)
+        .unwrap();
+
+    ConnectionOptions::new(project_id, auth).with_address(address)
+}
+
+async fn client_event_loop(client: Arc<Client>) {
+    let mut conn = Connection::new();
+    if let Some(control_rx) = client.control_rx() {
+        let mut control_rx = control_rx.lock().await;
+
+        loop {
+            tokio::select! {
+                event = control_rx.recv() => {
+                    match event {
+                        Some(event) => match event {
+                            ConnectionControl::Connect { request, tx } => {
+                                let result = conn.connect(request).await;
+                                if result.is_ok() {
+                                    println!("Client connected");
+                                }
+                                tx.send(result).ok();
+                            }
+                            ConnectionControl::Disconnect { tx } => {
+                                tx.send(conn.disconnect().await).ok();
+                            }
+                            ConnectionControl::OutboundRequest(request) => {
+                                conn.request(request);
+                            }
+                        }
+                        // Control TX has been dropped, shutting down.
+                        None => {
+                            conn.disconnect().await.ok();
+                            println!("Client disconnected");
+                            break;
+                        }
+                    }
+                }
+                event = conn.select_next_some() => {
+                    match event {
+                        StreamEvent::InboundSubscriptionRequest(request) => {
+                            println!("messaged: received: {:?}", PublishedMessage::from_request(&request));
+                            request.respond(Ok(true)).ok();
+                        }
+                        StreamEvent::InboundError(error) => {
+                            println!("Inbound error: {:?}", error);
+                        }
+                        StreamEvent::OutboundError(error) => {
+                            println!("Outbound error: {:?}", error);
+                        }
+                        StreamEvent::ConnectionClosed(frame) => {
+                            println!("connection closed: frame={frame:?}");
+                            conn.reset();
+                        }
+                    }
+                }
+            }
+        }
+    }
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    let args = Args::from_args();
+
+    let client1 = Arc::new(Client::new_unmanaged());
+    spawn(client_event_loop(client1.clone()));
+
+    client1
+        .connect(&create_conn_opts(&args.address, &args.project_id))
+        .await?;
+
+    let client2 = Arc::new(Client::new_unmanaged());
+    spawn(client_event_loop(client2.clone()));
+
+    client2
+        .connect(&create_conn_opts(&args.address, &args.project_id))
+        .await?;
+
+    let topic = Topic::generate();
+
+    let subscription_id = client1.subscribe(topic.clone()).await?;
+    println!("[client1] subscribed: topic={topic} subscription_id={subscription_id}");
+
+    client2
+        .publish(
+            topic.clone(),
+            Arc::from("Hello WalletConnect!"),
+            None,
+            0,
+            Duration::from_secs(60),
+            false,
+        )
+        .await?;
+
+    println!("[client2] published message with topic: {topic}",);
+
+    tokio::time::sleep(Duration::from_millis(500)).await;
+
+    drop(client1);
+    drop(client2);
+
+    tokio::time::sleep(Duration::from_millis(100)).await;
+
+    println!("clients disconnected");
+
+    Ok(())
+}

relay_client/Cargo.toml

@@ -5,30 +5,48 @@ edition = "2021"
 license = "Apache-2.0"
 
 [features]
-default = ["tokio-tungstenite/native-tls"]
-rustls = ["tokio-tungstenite/rustls-tls-native-roots"]
+default = ["tokio-tungstenite-wasm/native-tls"]
+rustls = ["tokio-tungstenite-wasm/rustls-tls-native-roots"]
+http = ["dep:reqwest"]
 
 [dependencies]
+chrono = { version = "0.4", default-features = false, features = [
+    "alloc",
+    "std",
+] }
+data-encoding = "2.6.0"
+futures-util = { version = "0.3", default-features = false, features = [
+    "sink",
+    "std",
+] }
+http = "1.0.0"
+pin-project = "1.0"
 relay_rpc = { path = "../relay_rpc" }
-futures-util = { version = "0.3", default-features = false, features = ["sink", "std"] }
-thiserror = "1.0"
+reqwest = { version = "0.12.2", optional = true, features = ["json"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 serde_qs = "0.10"
-pin-project = "1.0"
-chrono = { version = "0.4", default-features = false, features = ["alloc", "std"] }
+thiserror = "1.0"
+tokio = { version = "1.22", features = ["sync", "macros"] }
+tokio-tungstenite-wasm = { git = "https://github.com/KomodoPlatform/tokio-tungstenite-wasm.git", rev = "8fc7e2f" }
+tokio-util = "0.7"
 url = "2.3"
-http = "1.0.0"
 
-# HTTP client dependencies.
-reqwest = { version = "0.12.2", features = ["json"] }
+[target.'cfg(not(target_arch = "wasm32"))'.dependencies]
+rand = { version = "0.7", features = ["std", "small_rng"] }
 
-# WebSocket client dependencies.
-tokio = { version = "1.22", features = ["rt", "time", "sync", "macros", "rt-multi-thread"] }
-tokio-tungstenite = "0.21.0"
-futures-channel = "0.3"
-tokio-stream = "0.1"
-tokio-util = "0.7"
+[target.'cfg(target_arch = "wasm32")'.dependencies]
+js-sys = "0.3.27"
+rand = { version = "0.7", default-features = false, features = [
+    "std",
+    "small_rng",
+    "wasm-bindgen",
+] }
+getrandom = { version = "0.2", features = ["js"] }
+wasm-bindgen = "0.2.86"
+wasm-bindgen-test = { version = "0.3.2" }
+wasm-bindgen-futures = "0.4.21"
+web-sys = { version = "0.3.55", features = ["WebSocket"] }
 
 [lints.clippy]
 indexing_slicing = "deny"

relay_client/src/error.rs

@@ -14,11 +14,12 @@ pub enum RequestBuildError {
     Headers,
 
     #[error("Failed to parse connection URL: {0}")]
-    Url(#[from] url::ParseError),
+    Url(String),
 
     #[error("Failed to create websocket request: {0}")]
     WebsocketClient(#[from] crate::websocket::WebsocketClientError),
 
+    #[cfg(feature = "http")]
     #[error("Failed to create HTTP request: {0}")]
     HttpClient(#[from] crate::http::HttpClientError),
 }
@@ -32,6 +33,7 @@ pub enum ClientError {
     #[error("Websocket client error: {0}")]
     WebsocketClient(#[from] crate::websocket::WebsocketClientError),
 
+    #[cfg(feature = "http")]
     #[error("HTTP client error: {0}")]
     HttpClient(#[from] crate::http::HttpClientError),