Merge branch 'sync' of github.com:ethcore/parity into state

cov.sh

@@ -15,7 +15,7 @@ if ! type kcov > /dev/null; then
 	exit 1
 fi
 
-cargo test -p ethcore --no-run || exit $?
+cargo test --features ethcore/json-tests -p ethcore --no-run || exit $?
 mkdir -p target/coverage
 kcov --exclude-pattern ~/.multirust,rocksdb,secp256k1 --include-pattern src --verify target/coverage target/debug/deps/ethcore*
 xdg-open target/coverage/index.html

ethcore/src/block_queue.rs

@@ -79,6 +79,8 @@ struct Verification {
 	bad: HashSet<H256>,
 }
 
+const MAX_UNVERIFIED_QUEUE_SIZE: usize = 50000;
+
 impl BlockQueue {
 	/// Creates a new queue instance.
 	pub fn new(engine: Arc<Box<Engine>>, message_channel: IoChannel<NetSyncMessage>) -> BlockQueue {
@@ -290,7 +292,7 @@ impl BlockQueue {
 	pub fn queue_info(&self) -> BlockQueueInfo {
 		let verification = self.verification.lock().unwrap();
 		BlockQueueInfo {
-			full: false,
+			full: verification.unverified.len() + verification.verifying.len() + verification.verified.len() >= MAX_UNVERIFIED_QUEUE_SIZE,
 			verified_queue_size: verification.verified.len(),
 			unverified_queue_size: verification.unverified.len(),
 			verifying_queue_size: verification.verifying.len(),

ethcore/src/transaction.rs

@@ -67,6 +67,24 @@ impl Transaction {
 		}
 	}
 
+	/// Create a new message-call transaction.
+	#[allow(dead_code)]
+	pub fn new_call(to: Address, value: U256, data: Bytes, gas: U256, gas_price: U256, nonce: U256) -> Transaction {
+		Transaction {
+			nonce: nonce,
+			gas_price: gas_price,
+			gas: gas,
+			action: Action::Call(to),
+			value: value,
+			data: data,
+			v: 0,
+			r: x!(0),
+			s: x!(0),
+			hash: RefCell::new(None),
+			sender: RefCell::new(None),
+		}
+	}
+
 	/// Create a new contract-creation transaction.
 	#[cfg(test)]
 	pub fn new_create(value: U256, data: Bytes, gas: U256, gas_price: U256, nonce: U256) -> Transaction {

parity/main.rs

@@ -37,10 +37,10 @@ Usage:
 Options:
   -l --logging LOGGING     Specify the logging level.
   -j --jsonrpc             Enable the JSON-RPC API sever.
-  --jsonrpc-url URL        Specify URL for JSON-RPC API server (default: 127.0.0.1:8545).
+  --jsonrpc-url URL        Specify URL for JSON-RPC API server [default: 127.0.0.1:8545].
 
-  --cache-pref-size BYTES  Specify the prefered size of the blockchain cache in bytes (default: 16384).
+  --cache-pref-size BYTES  Specify the prefered size of the blockchain cache in bytes [default: 16384].
-  --cache-max-size BYTES   Specify the maximum size of the blockchain cache in bytes (default: 262144).
+  --cache-max-size BYTES   Specify the maximum size of the blockchain cache in bytes [default: 262144].
 
   -h --help                Show this screen.
 ", flag_cache_pref_size: usize, flag_cache_max_size: usize);

rpc/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 description = "Ethcore jsonrpc"
 name = "ethcore-rpc"
-version = "0.1.0"
+version = "0.9.0"
 license = "GPL-3.0"
 authors = ["Ethcore <admin@ethcore.io"]
 
@@ -17,4 +17,5 @@ ethcore-util = { path = "../util" }
 ethcore = { path = "../ethcore" }
 ethsync = { path = "../sync" }
 clippy = "0.0.37"
+target_info = "0.1.0"
 

rpc/src/lib.rs

@@ -4,6 +4,7 @@
 #![plugin(serde_macros)]
 #![plugin(clippy)]
 
+extern crate target_info;
 extern crate serde;
 extern crate serde_json;
 extern crate jsonrpc_core;

rpc/src/v1/impls/web3.rs

@@ -1,4 +1,5 @@
 //! Web3 rpc implementation.
+use target_info::Target;
 use jsonrpc_core::*;
 use v1::traits::Web3;
 
@@ -13,8 +14,7 @@ impl Web3Client {
 impl Web3 for Web3Client {
 	fn client_version(&self, params: Params) -> Result<Value, Error> {
 		match params {
-			//Params::None => Ok(Value::String("parity/0.1.0/-/rust1.7-nightly".to_owned())),
+			Params::None => Ok(Value::String(format!("Parity/-/{}/{}-{}-{}/rust1.8-nightly", env!("CARGO_PKG_VERSION"), Target::arch(), Target::env(), Target::os()))),
-			Params::None => Ok(Value::String("surprise/0.1.0/surprise/surprise".to_owned())),
 			_ => Err(Error::invalid_params())
 		}
 	}

sync/src/chain.rs

@@ -23,6 +23,7 @@ use ethcore::error::*;
 use ethcore::block::Block;
 use io::SyncIo;
 use time;
+use std::option::Option;
 
 impl ToUsize for BlockNumber {
 	fn to_usize(&self) -> usize {
@@ -103,14 +104,14 @@ pub struct SyncStatus {
 	pub protocol_version: u8,
 	/// BlockChain height for the moment the sync started.
 	pub start_block_number: BlockNumber,
-	/// Last fully downloaded and imported block number.
+	/// Last fully downloaded and imported block number (if any).
-	pub last_imported_block_number: BlockNumber,
+	pub last_imported_block_number: Option<BlockNumber>,
-	/// Highest block number in the download queue.
+	/// Highest block number in the download queue (if any).
-	pub highest_block_number: BlockNumber,
+	pub highest_block_number: Option<BlockNumber>,
 	/// Total number of blocks for the sync process.
-	pub blocks_total: usize,
+	pub blocks_total: BlockNumber,
 	/// Number of blocks downloaded so far.
-	pub blocks_received: usize,
+	pub blocks_received: BlockNumber,
 	/// Total number of connected peers
 	pub num_peers: usize,
 	/// Total number of active peers
@@ -153,7 +154,7 @@ pub struct ChainSync {
 	/// Last block number for the start of sync
 	starting_block: BlockNumber,
 	/// Highest block number seen
-	highest_block: BlockNumber,
+	highest_block: Option<BlockNumber>,
 	/// Set of block header numbers being downloaded
 	downloading_headers: HashSet<BlockNumber>,
 	/// Set of block body numbers being downloaded
@@ -167,9 +168,9 @@ pub struct ChainSync {
 	/// Used to map body to header
 	header_ids: HashMap<HeaderId, BlockNumber>,
 	/// Last impoted block number
-	last_imported_block: BlockNumber,
+	last_imported_block: Option<BlockNumber>,
 	/// Last impoted block hash
-	last_imported_hash: H256,
+	last_imported_hash: Option<H256>,
 	/// Syncing total  difficulty
 	syncing_difficulty: U256,
 	/// True if common block for our and remote chain has been found
@@ -183,15 +184,15 @@ impl ChainSync {
 		ChainSync {
 			state: SyncState::NotSynced,
 			starting_block: 0,
-			highest_block: 0,
+			highest_block: None,
 			downloading_headers: HashSet::new(),
 			downloading_bodies: HashSet::new(),
 			headers: Vec::new(),
 			bodies: Vec::new(),
 			peers: HashMap::new(),
 			header_ids: HashMap::new(),
-			last_imported_block: 0,
+			last_imported_block: None,
-			last_imported_hash: H256::new(),
+			last_imported_hash: None,
 			syncing_difficulty: U256::from(0u64),
 			have_common_block: false,
 		}
@@ -205,8 +206,8 @@ impl ChainSync {
 			start_block_number: self.starting_block,
 			last_imported_block_number: self.last_imported_block,
 			highest_block_number: self.highest_block,
-			blocks_received: (self.last_imported_block - self.starting_block) as usize,
+			blocks_received: match self.last_imported_block { None => 0, Some(x) => x - self.starting_block },
-			blocks_total: (self.highest_block - self.starting_block) as usize,
+			blocks_total: match self.highest_block { None => 0, Some(x) => x - self.starting_block },
 			num_peers: self.peers.len(),
 			num_active_peers: self.peers.values().filter(|p| p.asking != PeerAsking::Nothing).count(),
 		}
@@ -235,10 +236,10 @@ impl ChainSync {
 	/// Restart sync
 	pub fn restart(&mut self, io: &mut SyncIo) {
 		self.reset();
-		self.last_imported_block = 0;
+		self.last_imported_block = None;
-		self.last_imported_hash = H256::new();
+		self.last_imported_hash = None;
 		self.starting_block = 0;
-		self.highest_block = 0;
+		self.highest_block = None;
 		self.have_common_block = false;
 		io.chain().clear_queue();
 		self.starting_block = io.chain().chain_info().best_block_number;
@@ -300,32 +301,34 @@ impl ChainSync {
 		for i in 0..item_count {
 			let info: BlockHeader = try!(r.val_at(i));
 			let number = BlockNumber::from(info.number);
-			if number <= self.last_imported_block || self.headers.have_item(&number) {
+			if number <= self.current_base_block() || self.headers.have_item(&number) {
 				trace!(target: "sync", "Skipping existing block header");
 				continue;
 			}
-			if number > self.highest_block {
+
-				self.highest_block = number;
+			if self.highest_block == None || number > self.highest_block.unwrap() {
+				self.highest_block = Some(number);
 			}
 			let hash = info.hash();
 			match io.chain().block_status(&hash) {
 				BlockStatus::InChain => {
 					self.have_common_block = true;
-					self.last_imported_block = number;
+					self.last_imported_block = Some(number);
-					self.last_imported_hash = hash.clone();
+					self.last_imported_hash = Some(hash.clone());
 					trace!(target: "sync", "Found common header {} ({})", number, hash);
 				},
 				_ => {
 					if self.have_common_block {
 						//validate chain
-						if self.have_common_block && number == self.last_imported_block + 1 && info.parent_hash != self.last_imported_hash {
+						let base_hash = self.last_imported_hash.clone().unwrap();
-							// TODO: lower peer rating
+						if self.have_common_block && number == self.current_base_block() + 1 && info.parent_hash != base_hash {
-							debug!(target: "sync", "Mismatched block header {} {}", number, hash);
+							// Part of the forked chain. Restart to find common block again
-							continue;
+							debug!(target: "sync", "Mismatched block header {} {}, restarting sync", number, hash);
+							self.restart(io);
+							return Ok(());
 						}
 						if self.headers.find_item(&(number - 1)).map_or(false, |p| p.hash != info.parent_hash) {
 							// mismatching parent id, delete the previous block and don't add this one
-							// TODO: lower peer rating
 							debug!(target: "sync", "Mismatched block header {} {}", number, hash);
 							self.remove_downloaded_blocks(number - 1);
 							continue;
@@ -413,9 +416,9 @@ impl ChainSync {
 
 		trace!(target: "sync", "{} -> NewBlock ({})", peer_id, h);
 		let header_view = HeaderView::new(header_rlp.as_raw());
-		// TODO: Decompose block and add to self.headers and self.bodies instead
  		let mut unknown = false;
-		if header_view.number() == From::from(self.last_imported_block + 1) {
+		// TODO: Decompose block and add to self.headers and self.bodies instead
+		if header_view.number() == From::from(self.current_base_block() + 1) {
 			match io.chain().import_block(block_rlp.as_raw().to_vec()) {
 				Err(ImportError::AlreadyInChain) => {
 					trace!(target: "sync", "New block already in chain {:?}", h);
@@ -423,13 +426,13 @@ impl ChainSync {
 				Err(ImportError::AlreadyQueued) => {
 					trace!(target: "sync", "New block already queued {:?}", h);
 				},
+				Ok(_) => {
+					trace!(target: "sync", "New block queued {:?}", h);
+				},
 				Err(ImportError::UnknownParent) => {
 					unknown = true;
 					trace!(target: "sync", "New block with unknown parent {:?}", h);
 				},
-				Ok(_) => {
-					trace!(target: "sync", "New block queued {:?}", h);
-				},
 				Err(e) => {
 					debug!(target: "sync", "Bad new block {:?} : {:?}", h, e);
 					io.disable_peer(peer_id);
@@ -565,6 +568,10 @@ impl ChainSync {
 		}
 	}
 
+	fn current_base_block(&self) -> BlockNumber {
+		match self.last_imported_block { None => 0, Some(x) => x }
+	}
+
 	/// Find some headers or blocks to download for a peer.
 	fn request_blocks(&mut self, io: &mut SyncIo, peer_id: PeerId) {
 		self.clear_peer_download(peer_id);
@@ -578,7 +585,7 @@ impl ChainSync {
 		let mut needed_bodies: Vec<H256> = Vec::new();
 		let mut needed_numbers: Vec<BlockNumber> = Vec::new();
 
-		if self.have_common_block && !self.headers.is_empty() && self.headers.range_iter().next().unwrap().0 == self.last_imported_block + 1 {
+		if self.have_common_block && !self.headers.is_empty() && self.headers.range_iter().next().unwrap().0 == self.current_base_block() + 1 {
 			for (start, ref items) in self.headers.range_iter() {
 				if needed_bodies.len() > MAX_BODIES_TO_REQUEST {
 					break;
@@ -611,12 +618,12 @@ impl ChainSync {
 				}
 				if start == 0 {
 					self.have_common_block = true; //reached genesis
-					self.last_imported_hash = chain_info.genesis_hash;
+					self.last_imported_hash = Some(chain_info.genesis_hash);
 				}
 			}
 			if self.have_common_block {
 				let mut headers: Vec<BlockNumber> = Vec::new();
-				let mut prev = self.last_imported_block + 1;
+				let mut prev = self.current_base_block() + 1;
 				for (next, ref items) in self.headers.range_iter() {
 					if !headers.is_empty() {
 						break;
@@ -671,7 +678,7 @@ impl ChainSync {
 		{
 			let headers = self.headers.range_iter().next().unwrap();
 			let bodies = self.bodies.range_iter().next().unwrap();
-			if headers.0 != bodies.0 || headers.0 != self.last_imported_block + 1 {
+			if headers.0 != bodies.0 || headers.0 != self.current_base_block() + 1 {
 				return;
 			}
 
@@ -684,27 +691,29 @@ impl ChainSync {
 				block_rlp.append_raw(body.at(0).as_raw(), 1);
 				block_rlp.append_raw(body.at(1).as_raw(), 1);
 				let h = &headers.1[i].hash;
+
 				// Perform basic block verification
 				if !Block::is_good(block_rlp.as_raw()) {
 					debug!(target: "sync", "Bad block rlp {:?} : {:?}", h, block_rlp.as_raw());
 					restart = true;
 					break;
 				}
+
 				match io.chain().import_block(block_rlp.out()) {
 					Err(ImportError::AlreadyInChain) => {
 						trace!(target: "sync", "Block already in chain {:?}", h);
-						self.last_imported_block = headers.0 + i as BlockNumber;
+						self.last_imported_block = Some(headers.0 + i as BlockNumber);
-						self.last_imported_hash = h.clone();
+						self.last_imported_hash = Some(h.clone());
 					},
 					Err(ImportError::AlreadyQueued) => {
 						trace!(target: "sync", "Block already queued {:?}", h);
-						self.last_imported_block = headers.0 + i as BlockNumber;
+						self.last_imported_block = Some(headers.0 + i as BlockNumber);
-						self.last_imported_hash = h.clone();
+						self.last_imported_hash = Some(h.clone());
 					},
 					Ok(_) => {
 						trace!(target: "sync", "Block queued {:?}", h);
-						self.last_imported_block = headers.0 + i as BlockNumber;
+						self.last_imported_block = Some(headers.0 + i as BlockNumber);
-						self.last_imported_hash = h.clone();
+						self.last_imported_hash = Some(h.clone());
 						imported += 1;
 					},
 					Err(e) => {
@@ -721,8 +730,8 @@ impl ChainSync {
 			return;
 		}
 
-		self.headers.remove_head(&(self.last_imported_block + 1));
+		self.headers.remove_head(&(self.last_imported_block.unwrap() + 1));
-		self.bodies.remove_head(&(self.last_imported_block + 1));
+		self.bodies.remove_head(&(self.last_imported_block.unwrap() + 1));
 
 		if self.headers.is_empty() {
 			assert!(self.bodies.is_empty());
@@ -984,7 +993,10 @@ impl ChainSync {
 		})
 	}
 
-	/// Handle peer timeouts
+	/// Maintain other peers. Send out any new blocks and transactions
+	pub fn _maintain_sync(&mut self, _io: &mut SyncIo) {
+	}
+
 	pub fn maintain_peers(&self, io: &mut SyncIo) {
 		let tick = time::precise_time_s();
 		for (peer_id, peer) in &self.peers {
@@ -993,9 +1005,11 @@ impl ChainSync {
 			}
 		}
 	}
-
 	/// Maintain other peers. Send out any new blocks and transactions
-	pub fn _maintain_sync(&mut self, _io: &mut SyncIo) {
+	pub fn maintain_sync(&mut self, io: &mut SyncIo) {
+		if !io.chain().queue_info().full && self.state == SyncState::Waiting {
+			self.state = SyncState::Idle;
+			self.continue_sync(io);
+		}
 	}
 }
-

sync/src/lib.rs

@@ -107,6 +107,7 @@ impl NetworkProtocolHandler<SyncMessage> for EthSync {
 
 	fn timeout(&self, io: &NetworkContext<SyncMessage>, _timer: TimerToken) {
 		self.sync.write().unwrap().maintain_peers(&mut NetSyncIo::new(io, self.chain.deref()));
+		self.sync.write().unwrap().maintain_sync(&mut NetSyncIo::new(io, self.chain.deref()));
 	}
 }
 

sync/src/tests.rs

@@ -4,7 +4,7 @@ use ethcore::block_queue::BlockQueueInfo;
 use ethcore::header::{Header as BlockHeader, BlockNumber};
 use ethcore::error::*;
 use io::SyncIo;
-use chain::ChainSync;
+use chain::{ChainSync, SyncState};
 
 struct TestBlockChainClient {
 	blocks: RwLock<HashMap<H256, Bytes>>,
@@ -248,13 +248,15 @@ struct TestPeer {
 }
 
 struct TestNet {
-	peers: Vec<TestPeer>
+	peers: Vec<TestPeer>,
+	started: bool,
 }
 
 impl TestNet {
 	pub fn new(n: usize) -> TestNet {
 		let mut net = TestNet {
 			peers: Vec::new(),
+			started: false,
 		};
 		for _ in 0..n {
 			net.peers.push(TestPeer {
@@ -294,14 +296,32 @@ impl TestNet {
 				trace!("----------------");
 			}
 			let mut p = self.peers.get_mut(peer).unwrap();
-			p.sync._maintain_sync(&mut TestIo::new(&mut p.chain, &mut p.queue, None));
+			p.sync.maintain_sync(&mut TestIo::new(&mut p.chain, &mut p.queue, None));
 		}
 	}
 
-	pub fn sync(&mut self) {
+	pub fn restart_peer(&mut self, i: usize) {
+		let peer = self.peer_mut(i);
+		peer.sync.restart(&mut TestIo::new(&mut peer.chain, &mut peer.queue, None));
+	}
+
+	pub fn sync(&mut self) -> u32 {
 		self.start();
+		let mut total_steps = 0;
 		while !self.done() {
-			self.sync_step()
+			self.sync_step();
+			total_steps = total_steps + 1;
+		}
+		total_steps
+	}
+
+	pub fn sync_steps(&mut self, count: usize) {
+		if !self.started {
+			self.start();
+			self.started = true;
+		}
+		for _ in 0..count {
+			self.sync_step();
 		}
 	}
 
@@ -310,9 +330,8 @@ impl TestNet {
 	}
 }
 
-
 #[test]
-fn full_sync_two_peers() {
+fn chain_two_peers() {
 	::env_logger::init().ok();
 	let mut net = TestNet::new(3);
 	net.peer_mut(1).chain.add_blocks(1000, false);
@@ -323,7 +342,27 @@ fn full_sync_two_peers() {
 }
 
 #[test]
-fn full_sync_empty_blocks() {
+fn chain_status_after_sync() {
+	::env_logger::init().ok();
+	let mut net = TestNet::new(3);
+	net.peer_mut(1).chain.add_blocks(1000, false);
+	net.peer_mut(2).chain.add_blocks(1000, false);
+	net.sync();
+	let status = net.peer(0).sync.status();
+	assert_eq!(status.state, SyncState::Idle);
+}
+
+#[test]
+fn chain_takes_few_steps() {
+	let mut net = TestNet::new(3);
+	net.peer_mut(1).chain.add_blocks(100, false);
+	net.peer_mut(2).chain.add_blocks(100, false);
+	let total_steps = net.sync();
+	assert!(total_steps < 7);
+}
+
+#[test]
+fn chain_empty_blocks() {
 	::env_logger::init().ok();
 	let mut net = TestNet::new(3);
 	for n in 0..200 {
@@ -336,7 +375,7 @@ fn full_sync_empty_blocks() {
 }
 
 #[test]
-fn forked_sync() {
+fn chain_forked() {
 	::env_logger::init().ok();
 	let mut net = TestNet::new(3);
 	net.peer_mut(0).chain.add_blocks(300, false);
@@ -354,3 +393,25 @@ fn forked_sync() {
 	assert_eq!(net.peer(1).chain.numbers.read().unwrap().deref(), &peer1_chain);
 	assert_eq!(net.peer(2).chain.numbers.read().unwrap().deref(), &peer1_chain);
 }
+
+#[test]
+fn chain_restart() {
+	let mut net = TestNet::new(3);
+	net.peer_mut(1).chain.add_blocks(1000, false);
+	net.peer_mut(2).chain.add_blocks(1000, false);
+
+	net.sync_steps(8);
+
+	// make sure that sync has actually happened
+	assert!(net.peer(0).chain.chain_info().best_block_number > 100);
+	net.restart_peer(0);
+
+	let status = net.peer(0).sync.status();
+	assert_eq!(status.state, SyncState::NotSynced);
+}
+
+#[test]
+fn chain_status_empty() {
+	let net = TestNet::new(2);
+	assert_eq!(net.peer(0).sync.status().state, SyncState::NotSynced);
+}

util/cov.sh

@@ -1 +0,0 @@
-../cov.sh

util/cov.sh

@@ -0,0 +1,9 @@
+if ! type kcov > /dev/null; then
+   	echo "Install kcov first (details inside this file). Aborting."
+	exit 1
+fi
+
+cargo test --no-run || exit $?
+mkdir -p target/coverage
+kcov --exclude-pattern ~/.multirust,rocksdb,secp256k1 --include-pattern src --verify target/coverage target/debug/ethcore_util*
+xdg-open target/coverage/index.html

util/src/network/connection.rs

@@ -6,7 +6,7 @@ use hash::*;
 use sha3::*;
 use bytes::*;
 use rlp::*;
-use std::io::{self, Cursor, Read};
+use std::io::{self, Cursor, Read, Write};
 use error::*;
 use io::{IoContext, StreamToken};
 use network::error::NetworkError;
@@ -22,12 +22,17 @@ use tiny_keccak::Keccak;
 const ENCRYPTED_HEADER_LEN: usize = 32;
 const RECIEVE_PAYLOAD_TIMEOUT: u64 = 30000;
 
-/// Low level tcp connection
+pub trait GenericSocket : Read + Write {
-pub struct Connection {
+}
+
+impl GenericSocket for TcpStream {
+}
+
+pub struct GenericConnection<Socket: GenericSocket> {
 	/// Connection id (token)
 	pub token: StreamToken,
 	/// Network socket
-	pub socket: TcpStream,
+	pub socket: Socket,
 	/// Receive buffer
 	rec_buf: Bytes,
 	/// Expected size
@@ -36,34 +41,11 @@ pub struct Connection {
 	send_queue: VecDeque<Cursor<Bytes>>,
 	/// Event flags this connection expects
 	interest: EventSet,
-	/// Shared network staistics
+	/// Shared network statistics
 	stats: Arc<NetworkStats>,
 }
 
-/// Connection write status.
+impl<Socket: GenericSocket> GenericConnection<Socket> {
-#[derive(PartialEq, Eq)]
-pub enum WriteStatus {
-	/// Some data is still pending for current packet
-	Ongoing,
-	/// All data sent.
-	Complete
-}
-
-impl Connection {
-	/// Create a new connection with given id and socket.
-	pub fn new(token: StreamToken, socket: TcpStream, stats: Arc<NetworkStats>) -> Connection {
-		Connection {
-			token: token,
-			socket: socket,
-			send_queue: VecDeque::new(),
-			rec_buf: Bytes::new(),
-			rec_size: 0,
-			interest: EventSet::hup() | EventSet::readable(),
-			stats: stats,
-		}
-	}
-
-	/// Put a connection into read mode. Receiving up `size` bytes of data.
 	pub fn expect(&mut self, size: usize) {
 		if self.rec_size != self.rec_buf.len() {
 			warn!(target:"net", "Unexpected connection read start");
@@ -79,7 +61,7 @@ impl Connection {
 		}
 		let max = self.rec_size - self.rec_buf.len();
 		// resolve "multiple applicable items in scope [E0034]" error
-		let sock_ref = <TcpStream as Read>::by_ref(&mut self.socket);
+		let sock_ref = <Socket as Read>::by_ref(&mut self.socket);
 		match sock_ref.take(max as u64).try_read_buf(&mut self.rec_buf) {
 			Ok(Some(size)) if size != 0  => {
 				self.stats.inc_recv(size);
@@ -142,6 +124,24 @@ impl Connection {
 			Ok(r)
 		})
 	}
+}
+
+/// Low level tcp connection
+pub type Connection = GenericConnection<TcpStream>;
+
+impl Connection {
+	/// Create a new connection with given id and socket.
+	pub fn new(token: StreamToken, socket: TcpStream, stats: Arc<NetworkStats>) -> Connection {
+		Connection {
+			token: token,
+			socket: socket,
+			send_queue: VecDeque::new(),
+			rec_buf: Bytes::new(),
+			rec_size: 0,
+			interest: EventSet::hup() | EventSet::readable(),
+			stats: stats,
+		}
+	}
 
 	/// Register this connection with the IO event loop.
 	pub fn register_socket<Host: Handler>(&self, reg: Token, event_loop: &mut EventLoop<Host>) -> io::Result<()> {
@@ -169,6 +169,15 @@ impl Connection {
 	}
 }
 
+/// Connection write status.
+#[derive(PartialEq, Eq)]
+pub enum WriteStatus {
+	/// Some data is still pending for current packet
+	Ongoing,
+	/// All data sent.
+	Complete
+}
+
 /// RLPx packet
 pub struct Packet {
 	pub protocol: u16,
@@ -424,4 +433,230 @@ pub fn test_encryption() {
 	assert_eq!(got, after2);
 }
 
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use std::sync::*;
+	use super::super::stats::*;
+	use std::io::{Read, Write, Error, Cursor, ErrorKind};
+	use std::cmp;
+	use mio::{EventSet};
+	use std::collections::VecDeque;
+	use bytes::*;
 
+	struct TestSocket {
+		read_buffer: Vec<u8>,
+		write_buffer: Vec<u8>,
+		cursor: usize,
+		buf_size: usize,
+	}
+
+	impl TestSocket {
+		fn new() -> TestSocket {
+			TestSocket {
+				read_buffer: vec![],
+				write_buffer: vec![],
+				cursor: 0,
+				buf_size: 0,
+			}
+		}
+
+		fn new_buf(buf_size: usize) -> TestSocket {
+			TestSocket {
+				read_buffer: vec![],
+				write_buffer: vec![],
+				cursor: 0,
+				buf_size: buf_size,
+			}
+		}
+	}
+
+	impl Read for TestSocket {
+		fn read(&mut self, buf: &mut [u8]) -> Result<usize, Error> {
+			let end_position = cmp::min(self.read_buffer.len(), self.cursor+buf.len());
+			let len = cmp::max(end_position - self.cursor, 0);
+			match len {
+				0 => Ok(0),
+				_ => {
+					for i in self.cursor..end_position {
+						buf[i-self.cursor] = self.read_buffer[i];
+					}
+					self.cursor = self.cursor + buf.len();
+					Ok(len)
+				}
+			}
+		}
+	}
+
+	impl Write for TestSocket {
+		fn write(&mut self, buf: &[u8]) -> Result<usize, Error> {
+			if self.buf_size == 0 || buf.len() < self.buf_size {
+				self.write_buffer.extend(buf.iter().cloned());
+				Ok(buf.len())
+			}
+			else {
+				self.write_buffer.extend(buf.iter().take(self.buf_size).cloned());
+				Ok(self.buf_size)
+			}
+		}
+
+		fn flush(&mut self) -> Result<(), Error> {
+			unimplemented!();
+		}
+	}
+
+	impl GenericSocket for TestSocket {}
+
+	struct TestBrokenSocket {
+		error: String
+	}
+
+	impl Read for TestBrokenSocket {
+		fn read(&mut self, _: &mut [u8]) -> Result<usize, Error> {
+			Err(Error::new(ErrorKind::Other, self.error.clone()))
+		}
+	}
+
+	impl Write for TestBrokenSocket {
+		fn write(&mut self, _: &[u8]) -> Result<usize, Error> {
+			Err(Error::new(ErrorKind::Other, self.error.clone()))
+		}
+
+		fn flush(&mut self) -> Result<(), Error> {
+			unimplemented!();
+		}
+	}
+
+	impl GenericSocket for TestBrokenSocket {}
+
+	type TestConnection = GenericConnection<TestSocket>;
+
+	impl TestConnection {
+		pub fn new() -> TestConnection {
+			TestConnection {
+				token: 999998888usize,
+				socket: TestSocket::new(),
+				send_queue: VecDeque::new(),
+				rec_buf: Bytes::new(),
+				rec_size: 0,
+				interest: EventSet::hup() | EventSet::readable(),
+				stats: Arc::<NetworkStats>::new(NetworkStats::new()),
+			}
+		}
+	}
+
+	type TestBrokenConnection = GenericConnection<TestBrokenSocket>;
+
+	impl TestBrokenConnection {
+		pub fn new() -> TestBrokenConnection {
+			TestBrokenConnection {
+				token: 999998888usize,
+				socket: TestBrokenSocket { error: "test broken socket".to_owned() },
+				send_queue: VecDeque::new(),
+				rec_buf: Bytes::new(),
+				rec_size: 0,
+				interest: EventSet::hup() | EventSet::readable(),
+				stats: Arc::<NetworkStats>::new(NetworkStats::new()),
+			}
+		}
+	}
+
+	#[test]
+	fn connection_expect() {
+		let mut connection = TestConnection::new();
+		connection.expect(1024);
+		assert_eq!(1024, connection.rec_size);
+	}
+
+	#[test]
+	fn connection_write_empty() {
+		let mut connection = TestConnection::new();
+		let status = connection.writable();
+		assert!(status.is_ok());
+		assert!(WriteStatus::Complete == status.unwrap());
+	}
+
+	#[test]
+	fn connection_write() {
+		let mut connection = TestConnection::new();
+		let data = Cursor::new(vec![0; 10240]);
+		connection.send_queue.push_back(data);
+
+		let status = connection.writable();
+		assert!(status.is_ok());
+		assert!(WriteStatus::Complete == status.unwrap());
+		assert_eq!(10240, connection.socket.write_buffer.len());
+	}
+
+	#[test]
+	fn connection_write_is_buffered() {
+		let mut connection = TestConnection::new();
+		connection.socket = TestSocket::new_buf(1024);
+		let data = Cursor::new(vec![0; 10240]);
+		connection.send_queue.push_back(data);
+
+		let status = connection.writable();
+
+		assert!(status.is_ok());
+		assert!(WriteStatus::Ongoing == status.unwrap());
+		assert_eq!(1024, connection.socket.write_buffer.len());
+	}
+
+	#[test]
+	fn connection_write_to_broken() {
+		let mut connection = TestBrokenConnection::new();
+		let data = Cursor::new(vec![0; 10240]);
+		connection.send_queue.push_back(data);
+
+		let status = connection.writable();
+
+		assert!(!status.is_ok());
+		assert_eq!(1, connection.send_queue.len());
+	}
+
+	#[test]
+	fn connection_read() {
+		let mut connection = TestConnection::new();
+		connection.rec_size = 2048;
+		connection.rec_buf = vec![10; 1024];
+		connection.socket.read_buffer = vec![99; 2048];
+
+		let status = connection.readable();
+
+		assert!(status.is_ok());
+		assert_eq!(1024, connection.socket.cursor);
+	}
+
+	#[test]
+	fn connection_read_from_broken() {
+		let mut connection = TestBrokenConnection::new();
+		connection.rec_size = 2048;
+
+		let status = connection.readable();
+		assert!(!status.is_ok());
+		assert_eq!(0, connection.rec_buf.len());
+	}
+
+	#[test]
+	fn connection_read_nothing() {
+		let mut connection = TestConnection::new();
+		connection.rec_size = 2048;
+
+		let status = connection.readable();
+
+		assert!(status.is_ok());
+		assert_eq!(0, connection.rec_buf.len());
+	}
+
+	#[test]
+	fn connection_read_full() {
+		let mut connection = TestConnection::new();
+		connection.rec_size = 1024;
+		connection.rec_buf = vec![76;1024];
+
+		let status = connection.readable();
+
+		assert!(status.is_ok());
+		assert_eq!(0, connection.socket.cursor);
+	}
+}

util/src/network/node.rs

@@ -83,3 +83,36 @@ impl Hash for Node {
 	}
 }
 
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use std::str::FromStr;
+	use std::net::*;
+	use hash::*;
+
+	#[test]
+	fn endpoint_parse() {
+		let endpoint = NodeEndpoint::from_str("123.99.55.44:7770");
+		assert!(endpoint.is_ok());
+		let v4 = match endpoint.unwrap().address {
+			SocketAddr::V4(v4address) => v4address,
+			_ => panic!("should ve v4 address")
+		};
+		assert_eq!(SocketAddrV4::new(Ipv4Addr::new(123, 99, 55, 44), 7770), v4);
+	}
+
+	#[test]
+	fn node_parse() {
+		let node = Node::from_str("enode://a979fb575495b8d6db44f750317d0f4622bf4c2aa3365d6af7c284339968eef29b69ad0dce72a4d8db5ebb4968de0e3bec910127f134779fbcb0cb6d3331163c@22.99.55.44:7770");
+		assert!(node.is_ok());
+		let node = node.unwrap();
+		let v4 = match node.endpoint.address {
+			SocketAddr::V4(v4address) => v4address,
+			_ => panic!("should ve v4 address")
+		};
+		assert_eq!(SocketAddrV4::new(Ipv4Addr::new(22, 99, 55, 44), 7770), v4);
+		assert_eq!(
+			H512::from_str("a979fb575495b8d6db44f750317d0f4622bf4c2aa3365d6af7c284339968eef29b69ad0dce72a4d8db5ebb4968de0e3bec910127f134779fbcb0cb6d3331163c").unwrap(),
+			node.id);
+	}
+}

util/src/network/stats.rs

@@ -48,4 +48,13 @@ impl NetworkStats {
 	pub fn sessions(&self) -> usize {
 		self.sessions.load(Ordering::Relaxed)
 	}
+
+	#[cfg(test)]
+	pub fn new() -> NetworkStats {
+		NetworkStats {
+			recv: AtomicUsize::new(0),
+			send: AtomicUsize::new(0),
+			sessions: AtomicUsize::new(0),
+		}
+	}
 }

util/src/network/tests.rs

@@ -68,13 +68,13 @@ impl NetworkProtocolHandler<TestProtocolMessage> for TestProtocol {
 
 
 #[test]
-fn test_net_service() {
+fn net_service() {
 	let mut service = NetworkService::<TestProtocolMessage>::start(NetworkConfiguration::new()).expect("Error creating network service");
 	service.register_protocol(Arc::new(TestProtocol::default()), "myproto", &[1u8]).unwrap();
 }
 
 #[test]
-fn test_net_connect() {
+fn net_connect() {
 	let key1 = KeyPair::create().unwrap();
 	let mut config1 = NetworkConfiguration::new_with_port(30344);
 	config1.use_secret = Some(key1.secret().clone());
@@ -93,7 +93,7 @@ fn test_net_connect() {
 }
 
 #[test]
-fn test_net_timeout() {
+fn net_timeout() {
 	let config = NetworkConfiguration::new_with_port(30346);
 	let mut service = NetworkService::<TestProtocolMessage>::start(config).unwrap();
 	let handler = TestProtocol::register(&mut service);