Update/refactor to latest tokio_io (#122)

* failing tests for tcp and ipc

* encoder/decoder

* fix remaining

* also tcp

* fix warnings

* review fixes

* fix warnings

Author: NikVolf

ipc/Cargo.toml

@@ -14,6 +14,7 @@ tokio-service = "0.1"
 jsonrpc-core = { version = "7.0", path = "../core" }
 jsonrpc-server-utils = { version = "7.0", path = "../server-utils" }
 parity-tokio-ipc = { git = "https://github.com/nikvolf/parity-tokio-ipc" }
+bytes = "0.4"
 
 [dev-dependencies]
 env_logger = "0.4"

ipc/src/lib.rs

@@ -6,6 +6,7 @@ extern crate jsonrpc_core as jsonrpc;
 extern crate jsonrpc_server_utils as server_utils;
 extern crate parity_tokio_ipc;
 extern crate tokio_service;
+extern crate bytes;
 
 #[macro_use] extern crate log;
 

ipc/src/logger.rs

@@ -1,3 +1,5 @@
+#![allow(dead_code)]
+
 use std::env;
 use log::LogLevelFilter;
 use env_logger::LogBuilder;

ipc/src/server.rs

@@ -6,8 +6,9 @@ use jsonrpc::futures::{future, Future, Stream, Sink};
 use jsonrpc::futures::sync::oneshot;
 use jsonrpc::{Metadata, MetaIoHandler, Middleware, NoopMiddleware};
 use jsonrpc::futures::BoxFuture;
-use server_utils::tokio_core::io::Io;
+
 use server_utils::tokio_core::reactor::Remote;
+use server_utils::tokio_io::AsyncRead;
 use server_utils::reactor;
 
 use meta::{MetaExtractor, NoopExtractor, RequestContext};
@@ -95,12 +96,12 @@ impl<M: Metadata, S: Middleware<M>> ServerBuilder<M, S> {
 			let listener = match Endpoint::new(endpoint_addr, handle) {
 				Ok(l) => l,
 				Err(e) => {
-					start_signal.complete(Err(e));
+					start_signal.send(Err(e)).expect("Cannot fail since receiver never dropped before receiving");
 					return future::ok(()).boxed();
 				}
 			};
 
-			start_signal.complete(Ok(()));
+			start_signal.send(Ok(())).expect("Cannot fail since receiver never dropped before receiving");
 			let remote = handle.remote().clone();
 			let connections = listener.incoming();
 
@@ -128,7 +129,7 @@ impl<M: Metadata, S: Middleware<M>> ServerBuilder<M, S> {
 				.filter_map(|x| x);
 
 
-				let writer = writer.send_all(responses).then(move |_| {
+				let writer = writer.send_all(responses).then(|_| {
 					trace!(target: "ipc", "Peer: service finished");
 					Ok(())
 				});
@@ -160,7 +161,7 @@ pub struct Server {
 impl Server {
 	/// Closes the server (waits for finish)
 	pub fn close(mut self) {
-		self.stop.take().unwrap().complete(());
+		self.stop.take().map(|stop| stop.send(()));
 		self.remote.take().unwrap().close();
 		self.clear_file();
 	}
@@ -172,13 +173,13 @@ impl Server {
 
 	/// Remove socket file
 	fn clear_file(&self) {
-		::std::fs::remove_file(&self.path).unwrap_or_else(|_| {}); // ignore error, file could have been gone somewhere
+		let _ = ::std::fs::remove_file(&self.path); // ignore error, file could have been gone somewhere
 	}
 }
 
 impl Drop for Server {
 	fn drop(&mut self) {
-		self.stop.take().map(|stop| stop.complete(()));
+		let _ = self.stop.take().map(|stop| stop.send(()));
 		self.remote.take().map(|remote| remote.close());
 		self.clear_file();
 	}
@@ -195,7 +196,7 @@ mod tests {
 	use jsonrpc::futures::{Future, future};
 	use self::tokio_uds::UnixStream;
 	use server_utils::tokio_core::reactor::Core;
-	use server_utils::tokio_core::io;
+	use server_utils::tokio_io::io;
 
 	fn server_builder() -> ServerBuilder {
 		let mut io = MetaIoHandler::<()>::default();
@@ -275,6 +276,42 @@ mod tests {
 			);
 	}
 
+	#[test]
+	fn req_parallel() {
+		use std::thread;
+
+		::logger::init_log();
+		let path = "/tmp/test-ipc-45000";
+		let _server = run(path);
+
+		let mut handles = Vec::new();
+		for _ in 0..4 {
+			let path = path.clone();
+			handles.push(
+				thread::spawn(move || {
+					for _ in 0..100 {
+						let result = dummy_request_str(
+							&path,
+							b"{\"jsonrpc\": \"2.0\", \"method\": \"say_hello\", \"params\": [42, 23], \"id\": 1}\n",
+							);
+
+						assert_eq!(
+							result,
+							"{\"jsonrpc\":\"2.0\",\"result\":\"hello\",\"id\":1}\n",
+							"Response does not exactly much the expected response",
+							);				
+
+						::std::thread::sleep(::std::time::Duration::from_millis(10));	
+					}					
+				})
+			);
+		}	
+
+		for handle in handles.drain(..) {
+			handle.join().unwrap();
+		}
+	}
+
 	#[test]
 	fn close() {
 		::logger::init_log();

ipc/src/stream_codec.rs

@@ -1,5 +1,6 @@
 use std::io;
-use server_utils::tokio_core::io::{Codec, EasyBuf};
+use server_utils::tokio_io::codec::{Decoder, Encoder};
+use bytes::{BytesMut, BufMut};
 
 pub struct StreamCodec;
 
@@ -10,11 +11,11 @@ fn is_whitespace(byte: u8) -> bool {
 	}
 }
 
-impl Codec for StreamCodec {
-	type In = String;
-	type Out = String;
+impl Decoder for StreamCodec {
+	type Item = String;
+	type Error = io::Error;
 
-	fn decode(&mut self, buf: &mut EasyBuf) -> io::Result<Option<Self::In>> {
+	fn decode(&mut self, buf: &mut BytesMut) -> io::Result<Option<Self::Item>> {
 		let mut depth = 0;
 		let mut in_str = false;
 		let mut is_escaped = false;
@@ -44,7 +45,7 @@ impl Codec for StreamCodec {
 			}
 
 			if depth == 0 && idx != start_idx && idx - start_idx + 1 > whitespaces {
-				let bts = buf.drain_to(idx + 1);
+				let bts = buf.split_to(idx + 1);
 				match String::from_utf8(bts.as_ref().to_vec()) {
 					Ok(val) => { return Ok(Some(val)) },
 					Err(_) => { return Ok(None); } // skip non-utf requests (TODO: log error?)
@@ -54,10 +55,15 @@ impl Codec for StreamCodec {
 
 		Ok(None)
 	}
+}
 
-	fn encode(&mut self, msg: String, buf: &mut Vec<u8>) -> io::Result<()> {
-		buf.extend_from_slice(msg.as_bytes());
-		buf.push(b'\n'); // for #4750
+impl Encoder for StreamCodec {
+	type Item = String;
+	type Error = io::Error;
+	
+	fn encode(&mut self, msg: String, buf: &mut BytesMut) -> io::Result<()> {
+		buf.put_slice(msg.as_bytes());
+		buf.put(b'\n'); // for #4750
 		Ok(())
 	}
 }
@@ -66,12 +72,13 @@ impl Codec for StreamCodec {
 mod tests {
 
 	use super::StreamCodec;
-	use server_utils::tokio_core::io::{Codec, EasyBuf};
+	use server_utils::tokio_io::codec::Decoder;
+	use bytes::{BytesMut, BufMut};
 
 	#[test]
 	fn simple_encode() {
-		let mut buf = EasyBuf::new();
-		buf.get_mut().extend_from_slice(b"{ test: 1 }{ test: 2 }{ test: 3 }");
+		let mut buf = BytesMut::with_capacity(2048);
+		buf.put_slice(b"{ test: 1 }{ test: 2 }{ test: 3 }");
 
 		let mut codec = StreamCodec;
 
@@ -84,8 +91,8 @@ mod tests {
 
 	#[test]
 	fn whitespace() {
-		let mut buf = EasyBuf::new();
-		buf.get_mut().extend_from_slice(b"{ test: 1 }\n\n\n\n{ test: 2 }\n\r{\n test: 3 }  ");
+		let mut buf = BytesMut::with_capacity(2048);
+		buf.put_slice(b"{ test: 1 }\n\n\n\n{ test: 2 }\n\r{\n test: 3 }  ");
 
 		let mut codec = StreamCodec;
 
@@ -113,8 +120,8 @@ mod tests {
 
 	#[test]
 	fn fragmented_encode() {
-		let mut buf = EasyBuf::new();
-		buf.get_mut().extend_from_slice(b"{ test: 1 }{ test: 2 }{ tes");
+		let mut buf = BytesMut::with_capacity(2048);
+		buf.put_slice(b"{ test: 1 }{ test: 2 }{ tes");
 
 		let mut codec = StreamCodec;
 		let request = codec.decode(&mut buf)
@@ -126,7 +133,7 @@ mod tests {
 			.expect("There should be at least one request in second fragmented test");
 		assert_eq!(String::from_utf8(buf.as_ref().to_vec()).unwrap(), "{ tes");
 
-		buf.get_mut().extend_from_slice(b"t: 3 }");
+		buf.put_slice(b"t: 3 }");
 		let request = codec.decode(&mut buf)
 			.expect("There should be no error in third fragmented test")
 			.expect("There should be at least one request in third fragmented test");
@@ -150,8 +157,8 @@ mod tests {
 			]
 		}"#;
 
-		let mut buf = EasyBuf::new();
-		buf.get_mut().extend_from_slice(request.as_bytes());
+		let mut buf = BytesMut::with_capacity(65536);
+		buf.put_slice(request.as_bytes());
 
 		let mut codec = StreamCodec;
 		let parsed_request = codec.decode(&mut buf)

tcp/Cargo.toml

@@ -13,9 +13,9 @@ log = "0.3"
 parking_lot = "0.4"
 serde_json = "0.9"
 tokio-service = "0.1"
-
 jsonrpc-core = { version = "7.0", path = "../core" }
 jsonrpc-server-utils = { version = "7.0", path = "../server-utils" }
+bytes = "0.4"
 
 [dev-dependencies]
 lazy_static = "0.2"

tcp/src/lib.rs

@@ -27,6 +27,7 @@ extern crate jsonrpc_server_utils as server_utils;
 extern crate parking_lot;
 extern crate serde_json;
 extern crate tokio_service;
+extern crate bytes;
 
 #[macro_use] extern crate log;
 

tcp/src/line_codec.rs

@@ -1,16 +1,17 @@
 use std::{io, str};
-use server_utils::tokio_core::io::{Codec, EasyBuf};
+use server_utils::tokio_io::codec::{Decoder, Encoder};
+use bytes::{BytesMut, BufMut};
 
 pub struct LineCodec;
 
-impl Codec for LineCodec {
-	type In = String;
-	type Out = String;
+impl Decoder for LineCodec {
+	type Item = String;
+	type Error = io::Error;
 
-	fn decode(&mut self, buf: &mut EasyBuf) -> io::Result<Option<Self::In>> {
-		if let Some(i) = buf.as_slice().iter().position(|&b| b == b'\n') {
-			let line = buf.drain_to(i);
-			buf.drain_to(1);
+	fn decode(&mut self, buf: &mut BytesMut) -> io::Result<Option<Self::Item>> {
+		if let Some(i) = buf.as_ref().iter().position(|&b| b == b'\n') {
+			let line = buf.split_to(i);
+			buf.split_to(1);
 
 			match str::from_utf8(&line.as_ref()) {
 				Ok(s) => Ok(Some(s.to_string())),
@@ -20,10 +21,15 @@ impl Codec for LineCodec {
 			Ok(None)
 		}
 	}
+}
+
+impl Encoder for LineCodec {
+	type Item = String;
+	type Error = io::Error;
 
-	fn encode(&mut self, msg: String, buf: &mut Vec<u8>) -> io::Result<()> {
-		buf.extend_from_slice(msg.as_bytes());
-		buf.push(b'\n');
+	fn encode(&mut self, msg: String, buf: &mut BytesMut) -> io::Result<()> {
+		buf.put_slice(msg.as_bytes());
+		buf.put(b'\n');
 		Ok(())
 	}
 }
@@ -32,12 +38,14 @@ impl Codec for LineCodec {
 mod tests {
 
 	use super::LineCodec;
-	use server_utils::tokio_core::io::{Codec, EasyBuf};
+
+	use server_utils::tokio_io::codec::Decoder;
+	use bytes::{BytesMut, BufMut};
 
 	#[test]
 	fn simple_encode() {
-		let mut buf = EasyBuf::new();
-		buf.get_mut().extend_from_slice(b"{ test: 1 }\n{ test: 2 }\n{ test: 3 }");
+		let mut buf = BytesMut::with_capacity(2048);
+		buf.put_slice(b"{ test: 1 }\n{ test: 2 }\n{ test: 3 }");
 
 		let mut codec = LineCodec;
 

tcp/src/server.rs

@@ -8,7 +8,7 @@ use jsonrpc::{MetaIoHandler, Metadata, Middleware, NoopMiddleware};
 use jsonrpc::futures::{future, Future, Stream, Sink};
 use jsonrpc::futures::sync::{mpsc, oneshot};
 use server_utils::{reactor, tokio_core};
-use server_utils::tokio_core::io::Io;
+use server_utils::tokio_io::AsyncRead;
 
 use dispatch::{Dispatcher, SenderChannels, PeerMessageQueue};
 use line_codec::LineCodec;
@@ -164,19 +164,19 @@ pub struct Server {
 impl Server {
 	/// Closes the server (waits for finish)
 	pub fn close(mut self) {
-		self.stop.take().unwrap().complete(());
+		let _ = self.stop.take().map(|sg| sg.send(()));
 		self.remote.take().unwrap().close();
 	}
 
 	/// Wait for the server to finish
 	pub fn wait(mut self) {
-		self.remote.take().unwrap().wait();
+		self.remote.take().unwrap().close();
 	}
 }
 
 impl Drop for Server {
 	fn drop(&mut self) {
-		self.stop.take().map(|stop| stop.complete(()));
+		let _ = self.stop.take().map(|sg| sg.send(()));
 		self.remote.take().map(|remote| remote.close());
 	}
 }