Ensure IPC server can handle requests concurrently.

Author: pinkisemils

ipc/src/server.rs

@@ -6,7 +6,7 @@ use std::sync::Arc;
 use tokio_service::{self, Service as TokioService};
 use jsonrpc::futures::{future, Future, Stream, Sink};
 use jsonrpc::futures::sync::{mpsc, oneshot};
-use jsonrpc::{FutureResult, Metadata, MetaIoHandler, Middleware, NoopMiddleware};
+use jsonrpc::{Metadata, MetaIoHandler, Middleware, NoopMiddleware};
 use server_utils::{
 	tokio_codec::Framed,
 	tokio::{self, runtime::TaskExecutor, reactor::Handle},
@@ -36,13 +36,25 @@ impl<M: Metadata, S: Middleware<M>> tokio_service::Service for Service<M, S> {
 	type Request = String;
 	type Response = Option<String>;
 
-	type Error = ();
+	type Error = std::io::Error;
 
-	type Future = FutureResult<S::Future>;
+	type Future = Box<Future<Item=Self::Response,Error=Self::Error> + Send>;
 
 	fn call(&self, req: Self::Request) -> Self::Future {
 		trace!(target: "ipc", "Received request: {}", req);
-		self.handler.handle_request(&req, self.meta.clone())
+		let fut = self.handler
+			.handle_request(&req, self.meta.clone())
+			.then(|result| {
+				match result {
+					Err(_) => {
+						future::ok(None)
+					}
+					Ok(some_result) => future::ok(some_result),
+				}
+			})
+			.map_err(|_:()| std::io::ErrorKind::Other.into());
+
+		Box::new(fut)
 	}
 }
 
@@ -55,6 +67,7 @@ pub struct ServerBuilder<M: Metadata = (), S: Middleware<M> = NoopMiddleware> {
 	incoming_separator: codecs::Separator,
 	outgoing_separator: codecs::Separator,
 	security_attributes: SecurityAttributes,
+	client_buffer_size: usize,
 }
 
 impl<M: Metadata + Default, S: Middleware<M>> ServerBuilder<M, S> {
@@ -80,6 +93,7 @@ impl<M: Metadata, S: Middleware<M>> ServerBuilder<M, S> {
 			incoming_separator: codecs::Separator::Empty,
 			outgoing_separator: codecs::Separator::default(),
 			security_attributes: SecurityAttributes::empty(),
+			client_buffer_size: 5,
 		}
 	}
 
@@ -116,6 +130,12 @@ impl<M: Metadata, S: Middleware<M>> ServerBuilder<M, S> {
 		self
 	}
 
+	/// Sets how many concurrent requests per client can be processed at any one time. Set to 5 by default.
+	pub fn set_client_buffer_size(mut self, buffer_size: usize) -> Self {
+		self.client_buffer_size = buffer_size;
+		self
+	}
+
 	/// Creates a new server from the given endpoint.
 	pub fn start(self, path: &str) -> std::io::Result<Server> {
 		let executor = self.executor.initialize()?;
@@ -129,6 +149,7 @@ impl<M: Metadata, S: Middleware<M>> ServerBuilder<M, S> {
 		let (start_signal, start_receiver) = oneshot::channel();
 		let (wait_signal, wait_receiver) = oneshot::channel();
 		let security_attributes = self.security_attributes;
+		let client_buffer_size = self.client_buffer_size;
 
 		executor.spawn(future::lazy(move || {
 			let mut endpoint = Endpoint::new(endpoint_addr);
@@ -173,27 +194,17 @@ impl<M: Metadata, S: Middleware<M>> ServerBuilder<M, S> {
 						outgoing_separator.clone(),
 					),
 				).split();
-				let responses = reader.and_then(move |req| {
-					service.call(req).then(move |response| match response {
-						Err(e) => {
-							warn!(target: "ipc", "Error while processing request: {:?}", e);
-							future::ok(None)
-						},
-						Ok(None) => {
-							future::ok(None)
-						},
-						Ok(Some(response_data)) => {
-							trace!(target: "ipc", "Sent response: {}", &response_data);
-							future::ok(Some(response_data))
-						}
+				let responses = reader
+					.map(move |req| {
+						service.call(req)
 					})
-				})
-				.filter_map(|x| x)
-				// we use `select_with_weak` here, instead of `select`, to close the stream
-				// as soon as the ipc pipe is closed
-				.select_with_weak(receiver.map_err(|e| {
-					warn!(target: "ipc", "Notification error: {:?}", e);
-					std::io::ErrorKind::Other.into()
+					.buffer_unordered(client_buffer_size)
+					.filter_map(|x| x)
+					// we use `select_with_weak` here, instead of `select`, to close the stream
+					// as soon as the ipc pipe is closed
+					.select_with_weak(receiver.map_err(|e| {
+						warn!(target: "ipc", "Notification error: {:?}", e);
+						std::io::ErrorKind::Other.into()
 				}));
 
 				let writer = writer.send_all(responses).then(move |_| {