feat: WASM client via web-sys transport

.github/workflows/ci.yml

@@ -12,6 +12,11 @@ on:
     branches:
       - master
 
+env:
+  CARGO_TERM_COLOR: always
+  # Download a RPC server to run wasm tests against.
+  SUBSTRATE_URL: https://releases.parity.io/substrate/x86_64-debian:stretch/latest/substrate/substrate
+
 jobs:
   check-style:
     name: Check style
@@ -149,3 +154,27 @@ jobs:
 
       - name: Cargo nextest
         run: cargo nextest run --workspace
+
+  wasm_tests:
+    name: Test wasm
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@master
+
+    - name: Install
+      run: curl https://rustwasm.github.io/wasm-pack/installer/init.sh -sSf | sh
+
+    - name: Download Substrate
+      run: |
+        curl $SUBSTRATE_URL --output substrate --location
+        chmod +x substrate
+        mkdir -p ~/.local/bin
+        mv substrate ~/.local/bin
+    - name: Run WASM tests
+      run: |
+        substrate --dev --tmp > /dev/null 2>&1 &
+        wasm-pack test --headless --firefox
+        wasm-pack test --headless --chrome
+        pkill substrate
+      working-directory: wasm-tests

Cargo.toml

@@ -12,6 +12,8 @@ members = [
 	"client/ws-client",
 	"client/http-client",
 	"client/transport",
+	"client/wasm-client",
 	"proc-macros",
+	"wasm-tests",
 ]
 resolver = "2"

client/transport/Cargo.toml

@@ -12,30 +12,49 @@ documentation = "https://docs.rs/jsonrpsee-ws-client"
 [dependencies]
 jsonrpsee-types = { path = "../../types", version = "0.10.1", optional = true }
 jsonrpsee-core = { path = "../../core", version = "0.10.1", features = ["client"] }
-tracing = { version = "0.1", optional = true }
+tracing = "0.1"
+
+# optional
 thiserror = { version = "1", optional = true }
-futures = { version = "0.3.14", default-features = false, features = ["std"], optional = true }
+anyhow = { version = "1", optional = true }
+futures-channel = { version = "0.3.14", default-features = false, optional = true }
+futures-util = { version = "0.3.14", default-features = false, features = ["alloc"], optional = true }
 http = { version = "0.2", optional = true }
 tokio-util = { version = "0.7", features = ["compat"], optional = true }
 tokio = { version = "1", features = ["net", "time", "macros"], optional = true }
 pin-project = { version = "1", optional = true }
 rustls-native-certs = { version = "0.6", optional = true }
 webpki-roots = { version = "0.22", optional = true }
 tokio-rustls = { version = "0.23", optional = true }
+futures-timer = { version = "3", optional = true }
 
 # ws
 soketto = { version = "0.7.1", optional = true }
 
+# web-sys
+wasm-bindgen = { version = "0.2.69", optional = true }
+wasm-bindgen-futures = { version = "0.4.19", optional = true }
+gloo-net = { version = "0.1.0", default-features = false, features = ["json", "websocket"], optional = true }
+
 [features]
 tls = ["tokio-rustls", "webpki-roots", "rustls-native-certs"]
 ws = [
-    "futures",
+    "futures-util",
     "http",
     "tokio",
     "tokio-util",
     "soketto",
     "pin-project",
     "jsonrpsee-types",
     "thiserror",
-    "tracing"
+]
+web = [
+    "wasm-bindgen",
+    "wasm-bindgen-futures",
+    "gloo-net",
+    "futures-channel",
+    "futures-timer",
+    "futures-util",
+    "anyhow",
+    "thiserror",
 ]

client/transport/src/lib.rs

@@ -32,3 +32,7 @@
 /// Websocket transport
 #[cfg(feature = "ws")]
 pub mod ws;
+
+/// Websocket transport via web-sys.
+#[cfg(all(feature = "web", target_arch = "wasm32"))]
+pub mod web;

client/transport/src/web.rs

@@ -0,0 +1,88 @@
+use core::fmt;
+
+use futures_channel::mpsc;
+use futures_util::sink::SinkExt;
+use futures_util::stream::{SplitSink, SplitStream, StreamExt};
+use gloo_net::websocket::{futures::WebSocket, Message, WebSocketError};
+use jsonrpsee_core::async_trait;
+use jsonrpsee_core::client::{TransportReceiverT, TransportSenderT};
+
+/// Web-sys transport error that can occur.
+#[derive(Debug, thiserror::Error)]
+pub enum Error {
+	/// Internal send error
+	#[error("Could not send message: {0}")]
+	SendError(#[from] mpsc::SendError),
+	/// Sender went away
+	#[error("Sender went away couldn't receive the message")]
+	SenderDisconnected,
+	/// Error that occurred in `JS context`.
+	#[error("JS Error: {0:?}")]
+	Js(String),
+	/// WebSocket error
+	#[error("WebSocket Error: {0:?}")]
+	WebSocket(WebSocketError),
+}
+
+/// Sender.
+pub struct Sender(SplitSink<WebSocket, Message>);
+
+impl fmt::Debug for Sender {
+	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+		f.debug_struct("Sender").finish()
+	}
+}
+
+/// Receiver.
+pub struct Receiver(SplitStream<WebSocket>);
+
+impl fmt::Debug for Receiver {
+	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+		f.debug_struct("Receiver").finish()
+	}
+}
+
+#[async_trait(?Send)]
+impl TransportSenderT for Sender {
+	type Error = Error;
+
+	async fn send(&mut self, msg: String) -> Result<(), Self::Error> {
+		tracing::trace!("tx: {:?}", msg);
+		self.0.send(Message::Text(msg)).await.map_err(|e| Error::WebSocket(e))?;
+		Ok(())
+	}
+
+	async fn close(&mut self) -> Result<(), Error> {
+		Ok(())
+	}
+}
+
+#[async_trait(?Send)]
+impl TransportReceiverT for Receiver {
+	type Error = Error;
+
+	async fn receive(&mut self) -> Result<String, Self::Error> {
+		match self.0.next().await {
+			Some(Ok(msg)) => {
+				tracing::trace!("rx: {:?}", msg);
+
+				let txt = match msg {
+					Message::Bytes(bytes) => String::from_utf8(bytes).expect("WebSocket message is valid utf8; qed"),
+					Message::Text(txt) => txt,
+				};
+
+				Ok(txt)
+			}
+			Some(Err(err)) => Err(Error::WebSocket(err)),
+			None => Err(Error::SenderDisconnected),
+		}
+	}
+}
+
+/// Create a transport sender & receiver pair.
+pub async fn connect(url: impl AsRef<str>) -> Result<(Sender, Receiver), Error> {
+	let websocket = WebSocket::open(url.as_ref()).map_err(|e| Error::Js(e.to_string()))?;
+	let (write, read) = websocket.split();
+
+	Ok((Sender(write), Receiver(read)))
+}

client/transport/src/ws/mod.rs

@@ -30,7 +30,7 @@ use std::io;
 use std::net::{SocketAddr, ToSocketAddrs};
 use std::time::Duration;
 
-use futures::io::{BufReader, BufWriter};
+use futures_util::io::{BufReader, BufWriter};
 use jsonrpsee_core::client::{CertificateStore, TransportReceiverT, TransportSenderT};
 use jsonrpsee_core::TEN_MB_SIZE_BYTES;
 use jsonrpsee_core::{async_trait, Cow};
@@ -188,7 +188,7 @@ impl TransportSenderT for Sender {
 
 	/// Sends out a request. Returns a `Future` that finishes when the request has been
 	/// successfully sent.
-	async fn send(&mut self, body: String) -> Result<(), WsError> {
+	async fn send(&mut self, body: String) -> Result<(), Self::Error> {
 		tracing::debug!("send: {}", body);
 		self.inner.send_text(body).await?;
 		self.inner.flush().await?;
@@ -206,7 +206,7 @@ impl TransportReceiverT for Receiver {
 	type Error = WsError;
 
 	/// Returns a `Future` resolving when the server sent us something back.
-	async fn receive(&mut self) -> Result<String, WsError> {
+	async fn receive(&mut self) -> Result<String, Self::Error> {
 		let mut message = Vec::new();
 		self.inner.receive_data(&mut message).await?;
 		let s = String::from_utf8(message).expect("Found invalid UTF-8");

client/transport/src/ws/stream.rs

@@ -31,8 +31,8 @@ use std::pin::Pin;
 use std::task::Context;
 use std::task::Poll;
 
-use futures::io::{IoSlice, IoSliceMut};
-use futures::prelude::*;
+use futures_util::io::{IoSlice, IoSliceMut};
+use futures_util::*;
 use pin_project::pin_project;
 use tokio::net::TcpStream;
 use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};
@@ -53,13 +53,13 @@ impl AsyncRead for EitherStream {
 		match self.project() {
 			EitherStreamProj::Plain(s) => {
 				let compat = s.compat();
-				futures::pin_mut!(compat);
+				futures_util::pin_mut!(compat);
 				AsyncRead::poll_read(compat, cx, buf)
 			}
 			#[cfg(feature = "tls")]
 			EitherStreamProj::Tls(t) => {
 				let compat = t.compat();
-				futures::pin_mut!(compat);
+				futures_util::pin_mut!(compat);
 				AsyncRead::poll_read(compat, cx, buf)
 			}
 		}
@@ -73,13 +73,13 @@ impl AsyncRead for EitherStream {
 		match self.project() {
 			EitherStreamProj::Plain(s) => {
 				let compat = s.compat();
-				futures::pin_mut!(compat);
+				futures_util::pin_mut!(compat);
 				AsyncRead::poll_read_vectored(compat, cx, bufs)
 			}
 			#[cfg(feature = "tls")]
 			EitherStreamProj::Tls(t) => {
 				let compat = t.compat();
-				futures::pin_mut!(compat);
+				futures_util::pin_mut!(compat);
 				AsyncRead::poll_read_vectored(compat, cx, bufs)
 			}
 		}
@@ -91,13 +91,13 @@ impl AsyncWrite for EitherStream {
 		match self.project() {
 			EitherStreamProj::Plain(s) => {
 				let compat = s.compat_write();
-				futures::pin_mut!(compat);
+				futures_util::pin_mut!(compat);
 				AsyncWrite::poll_write(compat, cx, buf)
 			}
 			#[cfg(feature = "tls")]
 			EitherStreamProj::Tls(t) => {
 				let compat = t.compat_write();
-				futures::pin_mut!(compat);
+				futures_util::pin_mut!(compat);
 				AsyncWrite::poll_write(compat, cx, buf)
 			}
 		}
@@ -107,13 +107,13 @@ impl AsyncWrite for EitherStream {
 		match self.project() {
 			EitherStreamProj::Plain(s) => {
 				let compat = s.compat_write();
-				futures::pin_mut!(compat);
+				futures_util::pin_mut!(compat);
 				AsyncWrite::poll_write_vectored(compat, cx, bufs)
 			}
 			#[cfg(feature = "tls")]
 			EitherStreamProj::Tls(t) => {
 				let compat = t.compat_write();
-				futures::pin_mut!(compat);
+				futures_util::pin_mut!(compat);
 				AsyncWrite::poll_write_vectored(compat, cx, bufs)
 			}
 		}
@@ -123,13 +123,13 @@ impl AsyncWrite for EitherStream {
 		match self.project() {
 			EitherStreamProj::Plain(s) => {
 				let compat = s.compat_write();
-				futures::pin_mut!(compat);
+				futures_util::pin_mut!(compat);
 				AsyncWrite::poll_flush(compat, cx)
 			}
 			#[cfg(feature = "tls")]
 			EitherStreamProj::Tls(t) => {
 				let compat = t.compat_write();
-				futures::pin_mut!(compat);
+				futures_util::pin_mut!(compat);
 				AsyncWrite::poll_flush(compat, cx)
 			}
 		}
@@ -139,13 +139,13 @@ impl AsyncWrite for EitherStream {
 		match self.project() {
 			EitherStreamProj::Plain(s) => {
 				let compat = s.compat_write();
-				futures::pin_mut!(compat);
+				futures_util::pin_mut!(compat);
 				AsyncWrite::poll_close(compat, cx)
 			}
 			#[cfg(feature = "tls")]
 			EitherStreamProj::Tls(t) => {
 				let compat = t.compat_write();
-				futures::pin_mut!(compat);
+				futures_util::pin_mut!(compat);
 				AsyncWrite::poll_close(compat, cx)
 			}
 		}

client/wasm-client/Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+name = "jsonrpsee-wasm-client"
+version = "0.10.1"
+authors = ["Parity Technologies <admin@parity.io>", "Pierre Krieger <pierre.krieger1708@gmail.com>"]
+description = "WASM client for JSON-RPC"
+edition = "2021"
+license = "MIT"
+repository = "https://github.com/paritytech/jsonrpsee"
+homepage = "https://github.com/paritytech/jsonrpsee"
+documentation = "https://docs.rs/jsonrpsee-ws-client"
+
+[dependencies]
+jsonrpsee-types = { path = "../../types", version = "0.10.1" }
+jsonrpsee-client-transport = { path = "../transport", version = "0.10.1", features = ["web"] }
+jsonrpsee-core = { path = "../../core", version = "0.10.1", features = ["async-wasm-client"] }
+futures-channel = "0.3"
+
+[dev-dependencies]
+env_logger = "0.9"
+jsonrpsee-test-utils = { path = "../../test-utils" }
+tokio = { version = "1", features = ["macros"] }
+serde_json = "1"