feat: WASM client via web-sys transport

Cargo.toml

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

client/transport/Cargo.toml

@@ -12,30 +12,48 @@ documentation = "https://docs.rs/jsonrpsee-ws-client"
 [dependencies]
 jsonrpsee-types = { path = "../../types", version = "0.9.0", optional = true }
 jsonrpsee-core = { path = "../../core", version = "0.9.0", features = ["client"] }
-tracing = { version = "0.1", optional = true }
+tracing = "0.1"
 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, optional = true }
 http = { version = "0.2", optional = true }
 tokio-util = { version = "0.6", 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 }
 
+# wasm
+wasm-bindgen = { version = "0.2.69", optional = true }
+wasm-bindgen-futures = { version = "0.4.19", optional = true }
+js-sys = { version = "0.3.46", optional = true }
+web-sys = { version = "0.3.46", features = ["BinaryType", "Blob", "ErrorEvent", "FileReader", "MessageEvent", "CloseEvent", "ProgressEvent", "WebSocket", "console"], 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"
+]
+wasm = [
+    "wasm-bindgen",
+    "wasm-bindgen-futures",
+    "js-sys",
+    "web-sys",
+    "futures-channel",
+    "futures-timer",
+    "futures-util",
+    "anyhow",
 ]

client/transport/src/lib.rs

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

client/transport/src/web_sys.rs

@@ -0,0 +1,124 @@
+use std::time::Duration;
+
+use anyhow::anyhow;
+use futures_channel::{mpsc, oneshot};
+use futures_timer::Delay;
+use futures_util::future::{self, Either};
+use futures_util::StreamExt;
+use jsonrpsee_core::client::{TransportReceiverT, TransportSenderT};
+use jsonrpsee_core::{async_trait, Error};
+use wasm_bindgen::closure::Closure;
+use wasm_bindgen::{JsCast, JsValue};
+use web_sys::{CloseEvent, MessageEvent, WebSocket};
+
+#[derive(Debug)]
+enum WebSocketMessage {
+	Data(String),
+	Close,
+}
+
+/// Sender.
+#[derive(Debug)]
+pub struct Sender(mpsc::UnboundedSender<WebSocketMessage>);
+
+/// Receiver.
+#[derive(Debug)]
+pub struct Receiver(mpsc::UnboundedReceiver<String>);
+
+#[async_trait]
+impl TransportSenderT for Sender {
+	type Error = Error;
+
+	async fn send(&mut self, msg: String) -> Result<(), Self::Error> {
+		tracing::trace!("tx: {:?}", msg);
+		self.0.unbounded_send(WebSocketMessage::Data(msg)).map_err(|e| Error::Transport(anyhow!("{:?}", e)))
+	}
+
+	async fn close(&mut self) -> Result<(), Error> {
+		self.0.unbounded_send(WebSocketMessage::Close).map_err(|e| Error::Transport(anyhow!("{:?}", e)))
+	}
+}
+
+#[async_trait]
+impl TransportReceiverT for Receiver {
+	type Error = Error;
+
+	async fn receive(&mut self) -> Result<String, Self::Error> {
+		match self.0.next().await {
+			Some(msg) => {
+				tracing::trace!("rx: {:?}", msg);
+				Ok(msg)
+			}
+			None => Err(Error::Transport(anyhow!("Connection closed"))),
+		}
+	}
+}
+
+/// Create a transport sender & receiver pair.
+pub async fn connect(url: impl AsRef<str>, connection_timeout: Duration) -> Result<(Sender, Receiver), Error> {
+	let (from_back, rx) = mpsc::unbounded();
+	let (tx, mut to_back) = mpsc::unbounded();
+
+	let websocket =
+		WebSocket::new(url.as_ref()).map_err(|e| Error::Transport(anyhow!("Connection failed: {:?}", e)))?;
+	websocket.set_binary_type(web_sys::BinaryType::Arraybuffer);
+
+	let tx1 = tx.clone();
+
+	let from_back1 = from_back.clone();
+	let on_msg_callback = Closure::wrap(Box::new(move |e: MessageEvent| {
+		// Binary message.
+		if let Ok(abuf) = e.data().dyn_into::<js_sys::ArrayBuffer>() {
+			let msg = abuf.to_string();
+			let _ = from_back1.unbounded_send(msg.into());
+		// Text message.
+		} else if let Some(txt) = e.data().as_string() {
+			let _ = from_back1.unbounded_send(txt);
+		} else {
+			tracing::warn!("Received unsupported message");
+		}
+	}) as Box<dyn FnMut(MessageEvent)>);
+
+	// Close event.
+	let on_close_callback = Closure::once(move |_e: CloseEvent| {
+		tracing::info!("Connection closed");
+		tx1.close_channel();
+		from_back.close_channel();
+	});
+
+	let (conn_tx, conn_rx) = oneshot::channel();
+
+	let on_open_callback = Closure::once(move |_: JsValue| {
+		tracing::info!("Connection established");
+		conn_tx.send(()).expect("rx still alive; qed");
+	});
+
+	websocket.set_onopen(Some(on_open_callback.as_ref().unchecked_ref()));
+	websocket.set_onmessage(Some(on_msg_callback.as_ref().unchecked_ref()));
+	websocket.set_onclose(Some(on_close_callback.as_ref().unchecked_ref()));
+
+	// Prevent for being dropped (this will be leaked intentionally).
+	on_msg_callback.forget();
+	on_open_callback.forget();
+	on_close_callback.forget();
+
+	match future::select(conn_rx, Delay::new(connection_timeout)).await {
+		Either::Left((_, _)) => (),
+		Either::Right((_, _)) => return Err(Error::Transport(anyhow!("Connection timeout exceeded"))),
+	};
+
+	let tx3 = tx.clone();
+	wasm_bindgen_futures::spawn_local(async move {
+		while let Some(WebSocketMessage::Data(msg)) = to_back.next().await {
+			if let Err(e) = websocket.send_with_str(&msg) {
+				tracing::warn!("Failed to send: {:?}", e);
+				break;
+			}
+		}
+
+		let _ = websocket.close();
+		tx3.close_channel();
+	});
+
+	Ok((Sender(tx), Receiver(rx)))
+}

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/ws-client/src/lib.rs

@@ -174,6 +174,6 @@ impl<'a> WsClientBuilder<'a> {
 			.request_timeout(self.request_timeout)
 			.max_concurrent_requests(self.max_concurrent_requests)
 			.id_format(self.id_kind)
-			.build(sender, receiver))
+			.build_with_tokio(sender, receiver))
 	}
 }