[http client]: refactor with "syncronous-like" design

benches/benches.rs

@@ -6,8 +6,9 @@ use jsonrpsee::http::HttpServer;
 use jsonrpsee::types::jsonrpc::{JsonValue, Params};
 use jsonrpsee::ws::WsServer;
 use std::net::SocketAddr;
+use std::sync::Arc;
 
-criterion_group!(benches, http, ws);
+criterion_group!(benches, http_requests, websocket_requests);
 criterion_main!(benches);
 
 async fn http_server(tx: Sender<SocketAddr>) {
@@ -30,36 +31,75 @@ async fn ws_server(tx: Sender<SocketAddr>) {
 	}
 }
 
-pub fn http(c: &mut criterion::Criterion) {
-	c.bench_function("http 100 requests", |b| {
-		let (tx_addr, rx_addr) = oneshot::channel::<SocketAddr>();
-		async_std::task::spawn(http_server(tx_addr));
-		let server_addr = block_on(rx_addr).unwrap();
-		let client = HttpClient::new(&format!("http://{}", server_addr));
+pub fn http_requests(c: &mut criterion::Criterion) {
+	let mut rt = tokio::runtime::Runtime::new().unwrap();
+	let (tx_addr, rx_addr) = oneshot::channel::<SocketAddr>();
+	async_std::task::spawn(http_server(tx_addr));
+	let server_addr = block_on(rx_addr).unwrap();
+	let client = Arc::new(HttpClient::new(&format!("http://{}", server_addr), Default::default()));
 
+	c.bench_function("synchronous http round trip", |b| {
 		b.iter(|| {
-			block_on(async {
-				for _ in 0..100 {
-					let _: JsonValue = black_box(client.request("say_hello", Params::None).await.unwrap());
-				}
+			rt.block_on(async {
+				let _: JsonValue = black_box(client.request("say_hello", Params::None).await.unwrap());
 			})
 		})
 	});
+
+	c.bench_function_over_inputs(
+		"concurrent http round trip",
+		move |b: &mut Bencher, size: &usize| {
+			b.iter(|| {
+				let mut tasks = Vec::new();
+				for _ in 0..*size {
+					let client_rc = client.clone();
+					let task = rt.spawn(async move {
+						let _: Result<JsonValue, _> = black_box(client_rc.request("say_hello", Params::None)).await;
+					});
+					tasks.push(task);
+				}
+				for task in tasks {
+					rt.block_on(task).unwrap();
+				}
+			})
+		},
+		vec![2, 4, 8, 16, 32, 64, 128],
+	);
 }
 
-pub fn ws(c: &mut criterion::Criterion) {
-	c.bench_function("ws 100 request", |b| {
-		let (tx_addr, rx_addr) = oneshot::channel::<SocketAddr>();
-		async_std::task::spawn(ws_server(tx_addr));
-		let server_addr = block_on(rx_addr).unwrap();
-		let client = block_on(WsClient::new(&format!("ws://{}", server_addr))).unwrap();
+pub fn websocket_requests(c: &mut criterion::Criterion) {
+	let mut rt = tokio::runtime::Runtime::new().unwrap();
+	let (tx_addr, rx_addr) = oneshot::channel::<SocketAddr>();
+	async_std::task::spawn(ws_server(tx_addr));
+	let server_addr = block_on(rx_addr).unwrap();
+	let client = Arc::new(block_on(WsClient::new(&format!("ws://{}", server_addr))).unwrap());
 
+	c.bench_function("synchronous WebSocket round trip", |b| {
 		b.iter(|| {
-			block_on(async {
-				for _ in 0..100 {
-					let _: JsonValue = black_box(client.request("say_hello", Params::None).await.unwrap());
-				}
+			rt.block_on(async {
+				let _: JsonValue = black_box(client.request("say_hello", Params::None).await.unwrap());
 			})
 		})
 	});
+
+	c.bench_function_over_inputs(
+		"concurrent WebSocket round trip",
+		move |b: &mut Bencher, size: &usize| {
+			b.iter(|| {
+				let mut tasks = Vec::new();
+				for _ in 0..*size {
+					let client_rc = client.clone();
+					let task = rt.spawn(async move {
+						let _: Result<JsonValue, _> = black_box(client_rc.request("say_hello", Params::None)).await;
+					});
+					tasks.push(task);
+				}
+				for task in tasks {
+					rt.block_on(task).unwrap();
+				}
+			})
+		},
+		// TODO(niklasad1): investigate why it only works up to 8 concurrent requests.
+		vec![2, 4, 8],
+	);
 }

examples/http.rs

@@ -33,7 +33,7 @@ use jsonrpsee::types::jsonrpc::{JsonValue, Params};
 const SOCK_ADDR: &str = "127.0.0.1:9933";
 const SERVER_URI: &str = "http://localhost:9933";
 
-#[async_std::main]
+#[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
 	env_logger::init();
 
@@ -44,7 +44,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
 	server_started_rx.await?;
 
-	let client = HttpClient::new(SERVER_URI);
+	let client = HttpClient::new(SERVER_URI, Default::default());
 	let response: Result<JsonValue, _> = client.request("say_hello", Params::None).await;
 	println!("r: {:?}", response);
 

examples/ws.rs

@@ -43,9 +43,11 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 	});
 
 	server_started_rx.await?;
-	let client = WsClient::new(SERVER_URI).await?;
-	let response: JsonValue = client.request("say_hello", Params::None).await?;
-	println!("r: {:?}", response);
+	let c1 = WsClient::new(SERVER_URI).await?;
+
+	for _ in 0..10 {
+		let _: Result<String, _> = c1.request("say_hello", Params::None).await;
+	}
 
 	Ok(())
 }

src/client/http/client.rs

@@ -1,54 +1,35 @@
-use std::collections::HashMap;
-use std::io;
-
-use crate::client::http::raw::*;
 use crate::client::http::transport::HttpTransportClient;
-use crate::types::client::Error;
+use crate::types::client::{Error, Mismatch};
 use crate::types::jsonrpc::{self, JsonValue};
+use std::sync::atomic::{AtomicU64, Ordering};
 
-use futures::{channel::mpsc, channel::oneshot, future::Either, pin_mut, prelude::*};
+/// Default maximum request body size (10 MB).
+const DEFAULT_MAX_BODY_SIZE: usize = 10 * 1024 * 1024;
 
-/// Client that wraps a `RawClient` where the `RawClient` is spawned in a background worker tasks.
-///
-/// The communication is performed via a `mpsc` channel where the `Client` acts as simple frontend
-/// and just passes requests along to the backend (worker thread)
-#[derive(Clone)]
-pub struct Client {
-	backend: mpsc::Sender<FrontToBack>,
+/// HTTP configuration.
+#[derive(Copy, Clone)]
+pub struct HttpConfig {
+	/// Maximum request body size in bytes.
+	pub max_request_body_size: usize,
 }
 
-/// Message that the [`Client`] can send to the background task.
-enum FrontToBack {
-	/// Send a one-shot notification to the server. The server doesn't give back any feedback.
-	Notification {
-		/// Method for the notification.
-		method: String,
-		/// Parameters to send to the server.
-		params: jsonrpc::Params,
-	},
+/// HTTP Client.
+pub struct HttpClient {
+	transport: HttpTransportClient,
+	request_id: AtomicU64,
+}
 
-	/// Send a request to the server.
-	StartRequest {
-		/// Method for the request.
-		method: String,
-		/// Parameters of the request.
-		params: jsonrpc::Params,
-		/// One-shot channel where to send back the outcome of that request.
-		send_back: oneshot::Sender<Result<JsonValue, Error>>,
-	},
+impl Default for HttpConfig {
+	fn default() -> Self {
+		Self { max_request_body_size: DEFAULT_MAX_BODY_SIZE }
+	}
 }
 
-impl Client {
+impl HttpClient {
 	/// Create a client to connect to the server at address `endpoint`
-	pub fn new(endpoint: &str) -> Self {
-		let client = RawClient::new(HttpTransportClient::new(endpoint));
-
-		let (to_back, from_front) = mpsc::channel(16);
-		async_std::task::spawn(async move {
-			background_task(client, from_front).await;
-		});
-
-		Self { backend: to_back }
+	pub fn new(endpoint: &str, config: HttpConfig) -> Self {
+		let transport = HttpTransportClient::new(endpoint, config.max_request_body_size);
+		Self { transport, request_id: AtomicU64::new(0) }
 	}
 
 	/// Send a notification to the server.
@@ -57,99 +38,66 @@ impl Client {
 		method: impl Into<String>,
 		params: impl Into<jsonrpc::Params>,
 	) -> Result<(), Error> {
-		let method = method.into();
-		let params = params.into();
-		log::trace!("[frontend]: client send notification: method={:?}, params={:?}", method, params);
-		self.backend.clone().send(FrontToBack::Notification { method, params }).await.map_err(Error::InternalChannel)
+		let request = jsonrpc::Request::Single(jsonrpc::Call::Notification(jsonrpc::Notification {
+			jsonrpc: jsonrpc::Version::V2,
+			method: method.into(),
+			params: params.into(),
+		}));
+
+		self.transport.send_notification(request).await.map_err(|e| Error::TransportError(Box::new(e)))
 	}
 
 	/// Perform a request towards the server.
-	pub async fn request<Ret>(
+	pub async fn request(
 		&self,
 		method: impl Into<String>,
 		params: impl Into<jsonrpc::Params>,
-	) -> Result<Ret, Error>
-	where
-		Ret: jsonrpc::DeserializeOwned,
-	{
-		let method = method.into();
-		let params = params.into();
-		log::trace!("[frontend]: send request: method={:?}, params={:?}", method, params);
-		let (send_back_tx, send_back_rx) = oneshot::channel();
-
-		// TODO: send a `ChannelClosed` message if we close the channel unexpectedly
-
-		self.backend.clone().send(FrontToBack::StartRequest { method, params, send_back: send_back_tx }).await?;
-		let json_value = match send_back_rx.await {
-			Ok(Ok(v)) => v,
-			Ok(Err(err)) => return Err(err),
-			Err(_) => {
-				let err = io::Error::new(io::ErrorKind::Other, "background task closed");
-				return Err(Error::TransportError(Box::new(err)));
+	) -> Result<JsonValue, Error> {
+		// NOTE: `fetch_add` wraps on overflow which is intended.
+		let id = self.request_id.fetch_add(1, Ordering::SeqCst);
+		let request = jsonrpc::Request::Single(jsonrpc::Call::MethodCall(jsonrpc::MethodCall {
+			jsonrpc: jsonrpc::Version::V2,
+			method: method.into(),
+			params: params.into(),
+			id: jsonrpc::Id::Num(id),
+		}));
+
+		let response = self
+			.transport
+			.send_request_and_wait_for_response(request)
+			.await
+			.map_err(|e| Error::TransportError(Box::new(e)))?;
+
+		match response {
+			jsonrpc::Response::Single(rp) => Self::process_response(rp, id),
+			jsonrpc::Response::Batch(_rps) => {
+				todo!("batch request not supported");
+				// for rp in rps {
+				//     // TODO: if an error happens, we throw away the entire batch
+				//     self.process_response(rp)?;
+				// }
 			}
-		};
-		jsonrpc::from_value(json_value).map_err(Error::ParseError)
-	}
-}
-
-/// Function being run in the background that processes messages from the frontend.
-async fn background_task(mut client: RawClient, mut from_front: mpsc::Receiver<FrontToBack>) {
-	// List of requests that the server must answer.
-	let mut ongoing_requests: HashMap<RawClientRequestId, oneshot::Sender<Result<_, _>>> = HashMap::new();
-
-	loop {
-		// We need to do a little transformation in order to destroy the borrow to `client`
-		// and `from_front`.
-		let outcome = {
-			let next_message = from_front.next();
-			let next_event = client.next_event();
-			pin_mut!(next_message);
-			pin_mut!(next_event);
-			match future::select(next_message, next_event).await {
-				Either::Left((v, _)) => Either::Left(v),
-				Either::Right((v, _)) => Either::Right(v),
-			}
-		};
-
-		match outcome {
-			// If the channel is closed, then the `Client` has been destroyed and we
-			// stop this task.
-			Either::Left(None) => {
-				log::trace!("[backend]: background task terminated");
-				if !ongoing_requests.is_empty() {
-					log::warn!("client was dropped with {} pending requests", ongoing_requests.len());
-				}
-				return;
+			// Server MUST NOT reply to a Notification.
+			jsonrpc::Response::Notif(_notif) => {
+				Err(Error::Custom(format!("Server replied with notification response to request ID: {}", id)))
 			}
+		}
+	}
 
-			// User called `notification` on the front-end.
-			Either::Left(Some(FrontToBack::Notification { method, params })) => {
-				log::trace!("[backend]: send notification");
-				let _ = client.send_notification(method, params).await;
+	fn process_response(response: jsonrpc::Output, expected_id: u64) -> Result<JsonValue, Error> {
+		match response.id() {
+			jsonrpc::Id::Num(n) if n == &expected_id => {
+				let ret: Result<JsonValue, _> = response.into();
+				ret.map_err(|e| Error::Request(e))
 			}
-
-			// User called `request` on the front-end.
-			Either::Left(Some(FrontToBack::StartRequest { method, params, send_back })) => {
-				match client.start_request(&method, params).await {
-					Ok(id) => {
-						log::trace!("[backend]; send request: {:?} id: {:?}", method, id);
-						ongoing_requests.insert(id, send_back);
-					}
-					Err(err) => {
-						let _ = send_back.send(Err(Error::TransportError(Box::new(err))));
-					}
-				}
+			jsonrpc::Id::Num(n) => {
+				Err(Error::InvalidRequestId(Mismatch { expected: expected_id.into(), got: (*n).into() }))
 			}
-
-			// Received a response to a request from the server.
-			Either::Right(Ok(RawClientEvent::Response { request_id, result })) => {
-				log::trace!("[backend] received response to req={:?}, result={:?}", request_id, result);
-				let _ = ongoing_requests.remove(&request_id).unwrap().send(result.map_err(Error::Request));
+			jsonrpc::Id::Str(s) => {
+				Err(Error::InvalidRequestId(Mismatch { expected: expected_id.into(), got: s.to_string().into() }))
 			}
-
-			Either::Right(Err(e)) => {
-				// TODO: https://github.com/paritytech/jsonrpsee/issues/67
-				log::error!("Client Error: {:?}", e);
+			jsonrpc::Id::Null => {
+				Err(Error::InvalidRequestId(Mismatch { expected: expected_id.into(), got: JsonValue::Null }))
 			}
 		}
 	}

src/client/http/mod.rs

@@ -1,7 +1,8 @@
 mod client;
-mod raw;
 mod transport;
 
-pub use client::Client;
-pub use raw::RawClient;
+#[cfg(test)]
+mod tests;
+
+pub use client::{HttpClient, HttpConfig};
 pub use transport::HttpTransportClient;