Clean up the networking layer (#881)

substrate/core/network/src/io.rs

@@ -14,7 +14,9 @@
 // You should have received a copy of the GNU General Public License
 // along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
 
-use network_libp2p::{NetworkContext, Severity, NodeIndex, SessionInfo};
+use parking_lot::Mutex;
+use network_libp2p::{Service, Severity, NodeIndex, PeerId, ProtocolId};
+use std::sync::Arc;
 
 /// IO interface for the syncing handler.
 /// Provides peer connection management and an interface to the blockchain client.
@@ -24,49 +26,54 @@ pub trait SyncIo {
 	/// Send a packet to a peer.
 	fn send(&mut self, who: NodeIndex, data: Vec<u8>);
 	/// Returns peer identifier string
-	fn peer_info(&self, who: NodeIndex) -> String {
+	fn peer_debug_info(&self, who: NodeIndex) -> String {
 		who.to_string()
 	}
 	/// Returns information on p2p session
-	fn peer_session_info(&self, who: NodeIndex) -> Option<SessionInfo>;
-	/// Check if the session is expired
-	fn is_expired(&self) -> bool;
+	fn peer_id(&self, who: NodeIndex) -> Option<PeerId>;
 }
 
-/// Wraps `NetworkContext` and the blockchain client
+/// Wraps the network service.
 pub struct NetSyncIo<'s> {
-	network: &'s NetworkContext,
+	network: &'s Arc<Mutex<Service>>,
+	protocol: ProtocolId,
 }
 
 impl<'s> NetSyncIo<'s> {
-	/// Creates a new instance from the `NetworkContext` and the blockchain client reference.
-	pub fn new(network: &'s NetworkContext) -> NetSyncIo<'s> {
+	/// Creates a new instance.
+	pub fn new(network: &'s Arc<Mutex<Service>>, protocol: ProtocolId) -> NetSyncIo<'s> {
 		NetSyncIo {
-			network: network,
+			network,
+			protocol,
 		}
 	}
 }
 
 impl<'s> SyncIo for NetSyncIo<'s> {
 	fn report_peer(&mut self, who: NodeIndex, reason: Severity) {
-		self.network.report_peer(who, reason);
+		info!("Purposefully dropping {} ; reason: {:?}", who, reason);
+		match reason {
+			Severity::Bad(_) => self.network.lock().ban_node(who),
+			Severity::Useless(_) => self.network.lock().drop_node(who),
+			Severity::Timeout => self.network.lock().drop_node(who),
+		}
 	}
 
 	fn send(&mut self, who: NodeIndex, data: Vec<u8>) {
-		self.network.send(who, 0, data)
+		self.network.lock().send_custom_message(who, self.protocol, data)
 	}
 
-	fn peer_session_info(&self, who: NodeIndex) -> Option<SessionInfo> {
-		self.network.session_info(who)
+	fn peer_id(&self, who: NodeIndex) -> Option<PeerId> {
+		let net = self.network.lock();
+		net.peer_id_of_node(who).cloned()
 	}
 
-	fn is_expired(&self) -> bool {
-		self.network.is_expired()
-	}
-
-	fn peer_info(&self, who: NodeIndex) -> String {
-		self.network.peer_client_version(who)
+	fn peer_debug_info(&self, who: NodeIndex) -> String {
+		let net = self.network.lock();
+		if let (Some(peer_id), Some(addr)) = (net.peer_id_of_node(who), net.node_endpoint(who)) {
+			format!("{:?} through {:?}", peer_id, addr)
+		} else {
+			"unknown".to_string()
+		}
 	}
 }
-
-

substrate/core/network/src/lib.rs

@@ -32,6 +32,7 @@ extern crate parity_codec as codec;
 extern crate futures;
 extern crate rustc_hex;
 extern crate rand;
+extern crate tokio;
 #[macro_use] extern crate log;
 #[macro_use] extern crate bitflags;
 #[macro_use] extern crate error_chain;

substrate/core/network/src/protocol.rs

@@ -41,8 +41,6 @@ const REQUEST_TIMEOUT_SEC: u64 = 40;
 
 /// Current protocol version.
 pub (crate) const CURRENT_VERSION: u32 = 1;
-/// Current packet count.
-pub (crate) const CURRENT_PACKET_COUNT: u8 = 1;
 
 // Maximum allowed entries in `BlockResponse`
 const MAX_BLOCK_DATA_RESPONSE: u32 = 128;
@@ -289,14 +287,14 @@ impl<B: BlockT, S: Specialization<B>, H: ExHashT> Protocol<B, S, H> {
 
 	/// Called when a new peer is connected
 	pub fn on_peer_connected(&self, io: &mut SyncIo, who: NodeIndex) {
-		trace!(target: "sync", "Connected {}: {}", who, io.peer_info(who));
+		trace!(target: "sync", "Connected {}: {}", who, io.peer_debug_info(who));
 		self.handshaking_peers.write().insert(who, time::Instant::now());
 		self.send_status(io, who);
 	}
 
 	/// Called by peer when it is disconnecting
 	pub fn on_peer_disconnected(&self, io: &mut SyncIo, peer: NodeIndex) {
-		trace!(target: "sync", "Disconnecting {}: {}", peer, io.peer_info(peer));
+		trace!(target: "sync", "Disconnecting {}: {}", peer, io.peer_debug_info(peer));
 
 		// lock all the the peer lists so that add/remove peer events are in order
 		let mut sync = self.sync.write();
@@ -420,16 +418,12 @@ impl<B: BlockT, S: Specialization<B>, H: ExHashT> Protocol<B, S, H> {
 	/// Called by peer to report status
 	fn on_status_message(&self, io: &mut SyncIo, who: NodeIndex, status: message::Status<B>) {
 		trace!(target: "sync", "New peer {} {:?}", who, status);
-		if io.is_expired() {
-			trace!(target: "sync", "Status packet from expired session {}:{}", who, io.peer_info(who));
-			return;
-		}
 
 		{
 			let mut peers = self.context_data.peers.write();
 			let mut handshaking_peers = self.handshaking_peers.write();
 			if peers.contains_key(&who) {
-				debug!(target: "sync", "Unexpected status packet from {}:{}", who, io.peer_info(who));
+				debug!(target: "sync", "Unexpected status packet from {}:{}", who, io.peer_debug_info(who));
 				return;
 			}
 			if status.genesis_hash != self.genesis_hash {
@@ -464,7 +458,7 @@ impl<B: BlockT, S: Specialization<B>, H: ExHashT> Protocol<B, S, H> {
 			};
 			peers.insert(who.clone(), peer);
 			handshaking_peers.remove(&who);
-			debug!(target: "sync", "Connected {} {}", who, io.peer_info(who));
+			debug!(target: "sync", "Connected {} {}", who, io.peer_debug_info(who));
 		}
 
 		let mut context = ProtocolContext::new(&self.context_data, io);
@@ -514,9 +508,7 @@ impl<B: BlockT, S: Specialization<B>, H: ExHashT> Protocol<B, S, H> {
 				.unzip();
 
 			if !to_send.is_empty() {
-				let node_id = io.peer_session_info(*who)
-					.map(|info| format!("{}@{:?}", info.remote_address, info.id));
-
+				let node_id = io.peer_id(*who).map(|id| id.to_base58());
 				if let Some(id) = node_id {
 					for hash in hashes {
 						propagated_to.entry(hash).or_insert_with(Vec::new).push(id.clone());

substrate/core/network/src/service.rs

@@ -16,14 +16,15 @@
 
 use std::collections::HashMap;
 use std::sync::Arc;
-use std::io;
+use std::{io, thread};
 use std::time::Duration;
-use futures::sync::{oneshot, mpsc};
-use network_libp2p::{NetworkProtocolHandler, NetworkContext, NodeIndex, ProtocolId,
-NetworkConfiguration , NonReservedPeerMode, ErrorKind};
-use network_libp2p::{NetworkService, PeerId};
+use futures::{self, Future, Stream, stream, sync::{oneshot, mpsc}};
+use parking_lot::Mutex;
+use network_libp2p::{ProtocolId, PeerId, NetworkConfiguration, ErrorKind};
+use network_libp2p::{start_service, Service as NetworkService, ServiceEvent as NetworkServiceEvent};
+use network_libp2p::{RegisteredProtocol, parse_str_addr, Protocol as Libp2pProtocol};
 use io::NetSyncIo;
-use protocol::{Protocol, ProtocolContext, Context, ProtocolStatus, PeerInfo as ProtocolPeerInfo};
+use protocol::{self, Protocol, ProtocolContext, Context, ProtocolStatus};
 use config::{ProtocolConfig};
 use error::Error;
 use chain::Client;
@@ -32,18 +33,14 @@ use specialization::Specialization;
 use on_demand::OnDemandService;
 use import_queue::AsyncImportQueue;
 use runtime_primitives::traits::{Block as BlockT};
+use tokio::{runtime::Runtime, timer::Interval};
 
 /// Type that represents fetch completion future.
 pub type FetchFuture = oneshot::Receiver<Vec<u8>>;
 /// Type that represents bft messages stream.
 pub type BftMessageStream<B> = mpsc::UnboundedReceiver<LocalizedBftMessage<B>>;
 
-type TimerToken = usize;
-
-const TICK_TOKEN: TimerToken = 0;
 const TICK_TIMEOUT: Duration = Duration::from_millis(1000);
-
-const PROPAGATE_TOKEN: TimerToken = 1;
 const PROPAGATE_TIMEOUT: Duration = Duration::from_millis(5000);
 
 bitflags! {
@@ -76,8 +73,6 @@ impl ::codec::Decode for Roles {
 pub trait SyncProvider<B: BlockT>: Send + Sync {
 	/// Get sync status
 	fn status(&self) -> ProtocolStatus<B>;
-	/// Get peers information
-	fn peers(&self) -> Vec<PeerInfo<B>>;
 	/// Get this node id if available.
 	fn node_id(&self) -> Option<String>;
 }
@@ -113,28 +108,6 @@ pub trait ExecuteInContext<B: BlockT>: Send + Sync {
 	fn execute_in_context<F: Fn(&mut Context<B>)>(&self, closure: F);
 }
 
-/// Network protocol handler
-struct ProtocolHandler<B: BlockT, S: Specialization<B>, H: ExHashT> {
-	protocol: Protocol<B, S, H>,
-}
-
-/// Peer connection information
-#[derive(Debug)]
-pub struct PeerInfo<B: BlockT> {
-	/// Public node id
-	pub id: PeerId,
-	/// Node client ID
-	pub client_version: String,
-	/// Capabilities
-	pub capabilities: Vec<String>,
-	/// Remote endpoint address
-	pub remote_address: String,
-	/// Local endpoint address
-	pub local_address: String,
-	/// Dot protocol info.
-	pub dot_info: Option<ProtocolPeerInfo<B>>,
-}
-
 /// Service initialization parameters.
 pub struct Params<B: BlockT, S, H: ExHashT> {
 	/// Configuration.
@@ -154,50 +127,43 @@ pub struct Params<B: BlockT, S, H: ExHashT> {
 /// Substrate network service. Handles network IO and manages connectivity.
 pub struct Service<B: BlockT + 'static, S: Specialization<B>, H: ExHashT> {
 	/// Network service
-	network: NetworkService,
-	/// Devp2p protocol handler
-	handler: Arc<ProtocolHandler<B, S, H>>,
-	/// Devp2p protocol ID.
+	network: Arc<Mutex<NetworkService>>,
+	/// Protocol handler
+	handler: Arc<Protocol<B, S, H>>,
+	/// Protocol ID.
 	protocol_id: ProtocolId,
+	/// Sender for messages to the backgound service task, and handle for the background thread.
+	/// Dropping the sender should close the task and the thread.
+	/// This is an `Option` because we need to extract it in the destructor.
+	bg_thread: Option<(oneshot::Sender<()>, thread::JoinHandle<()>)>,
 }
 
 impl<B: BlockT + 'static, S: Specialization<B>, H: ExHashT> Service<B, S, H> {
 	/// Creates and register protocol with the network service
 	pub fn new(params: Params<B, S, H>, protocol_id: ProtocolId) -> Result<Arc<Service<B, S, H>>, Error> {
 		let chain = params.chain.clone();
-		// TODO: non-insatnt finality.
+		// TODO: non-instant finality.
 		let import_queue = Arc::new(AsyncImportQueue::new(true));
-		let handler = Arc::new(ProtocolHandler {
-			protocol: Protocol::new(
-				params.config,
-				params.chain,
-				import_queue.clone(),
-				params.on_demand,
-				params.transaction_pool,
-				params.specialization,
-			)?,
-		});
-		let versions = [(::protocol::CURRENT_VERSION as u8, ::protocol::CURRENT_PACKET_COUNT)];
-		let protocols = vec![(handler.clone() as Arc<_>, protocol_id, &versions[..])];
-		let service = match NetworkService::new(params.network_config.clone(), protocols) {
-			Ok(service) => service,
-			Err(err) => {
-				match err.kind() {
-					ErrorKind::Io(ref e) if e.kind() == io::ErrorKind::AddrInUse =>
-						warn!("Network port is already in use, make sure that another instance of Substrate client is not running or change the port using the --port option."),
-					_ => warn!("Error starting network: {}", err),
-				};
-				return Err(err.into())
-			},
-		};
+		let handler = Arc::new(Protocol::new(
+			params.config,
+			params.chain,
+			import_queue.clone(),
+			params.on_demand,
+			params.transaction_pool,
+			params.specialization,
+		)?);
+		let versions = [(protocol::CURRENT_VERSION as u8)];
+		let registered = RegisteredProtocol::new(protocol_id, &versions[..]);
+		let (thread, network) = start_thread(params.network_config, handler.clone(), registered)?;
 		let sync = Arc::new(Service {
-			network: service,
+			network,
 			protocol_id,
 			handler,
+			bg_thread: Some(thread),
 		});
 
 		import_queue.start(
-			Arc::downgrade(sync.handler.protocol.sync()),
+			Arc::downgrade(sync.handler.sync()),
 			Arc::downgrade(&sync),
 			Arc::downgrade(&chain)
 		)?;
@@ -207,106 +173,57 @@ impl<B: BlockT + 'static, S: Specialization<B>, H: ExHashT> Service<B, S, H> {
 
 	/// Called when a new block is imported by the client.
 	pub fn on_block_imported(&self, hash: B::Hash, header: &B::Header) {
-		self.network.with_context(self.protocol_id, |context| {
-			self.handler.protocol.on_block_imported(&mut NetSyncIo::new(context), hash, header)
-		});
+		self.handler.on_block_imported(&mut NetSyncIo::new(&self.network, self.protocol_id), hash, header)
 	}
 
 	/// Called when new transactons are imported by the client.
 	pub fn trigger_repropagate(&self) {
-		self.network.with_context(self.protocol_id, |context| {
-			self.handler.protocol.propagate_extrinsics(&mut NetSyncIo::new(context));
-		});
+		self.handler.propagate_extrinsics(&mut NetSyncIo::new(&self.network, self.protocol_id));
 	}
 
 	/// Execute a closure with the chain-specific network specialization.
-	/// If the network is unavailable, this will return `None`.
-	pub fn with_spec<F, U>(&self, f: F) -> Option<U>
+	pub fn with_spec<F, U>(&self, f: F) -> U
 		where F: FnOnce(&mut S, &mut Context<B>) -> U
 	{
-		let mut res = None;
-		self.network.with_context(self.protocol_id, |context| {
-			res = Some(self.handler.protocol.with_spec(&mut NetSyncIo::new(context), f))
-		});
-
-		res
+		self.handler.with_spec(&mut NetSyncIo::new(&self.network, self.protocol_id), f)
 	}
 }
 
 impl<B: BlockT + 'static, S: Specialization<B>, H:ExHashT> Drop for Service<B, S, H> {
 	fn drop(&mut self) {
-		self.handler.protocol.stop();
+		self.handler.stop();
+		if let Some((sender, join)) = self.bg_thread.take() {
+			let _ = sender.send(());
+			if let Err(e) = join.join() {
+				error!("Error while waiting on background thread: {:?}", e);
+			}
+		}
 	}
 }
+
 impl<B: BlockT + 'static, S: Specialization<B>, H: ExHashT> ExecuteInContext<B> for Service<B, S, H> {
 	fn execute_in_context<F: Fn(&mut ::protocol::Context<B>)>(&self, closure: F) {
-		self.network.with_context(self.protocol_id, |context| {
-			closure(&mut ProtocolContext::new(self.handler.protocol.context_data(), &mut NetSyncIo::new(context)))
-		});
+		closure(&mut ProtocolContext::new(self.handler.context_data(), &mut NetSyncIo::new(&self.network, self.protocol_id)))
 	}
 }
 
 impl<B: BlockT + 'static, S: Specialization<B>, H: ExHashT> SyncProvider<B> for Service<B, S, H> {
 	/// Get sync status
 	fn status(&self) -> ProtocolStatus<B> {
-		self.handler.protocol.status()
-	}
-
-	/// Get sync peers
-	fn peers(&self) -> Vec<PeerInfo<B>> {
-		self.network.with_context_eval(self.protocol_id, |ctx| {
-			let peer_ids = self.network.connected_peers();
-
-			peer_ids.into_iter().filter_map(|who| {
-				let session_info = match ctx.session_info(who) {
-					None => return None,
-					Some(info) => info,
-				};
-
-				Some(PeerInfo {
-					id: session_info.id,
-					client_version: session_info.client_version,
-					capabilities: session_info.peer_capabilities.into_iter().map(|c| c.to_string()).collect(),
-					remote_address: session_info.remote_address,
-					local_address: session_info.local_address,
-					dot_info: self.handler.protocol.peer_info(who),
-				})
-			}).collect()
-		}).unwrap_or_else(Vec::new)
+		self.handler.status()
 	}
 
 	fn node_id(&self) -> Option<String> {
-		self.network.external_url()
-	}
-}
-
-impl<B: BlockT + 'static, S: Specialization<B>, H: ExHashT> NetworkProtocolHandler for ProtocolHandler<B, S, H> {
-	fn initialize(&self, io: &NetworkContext) {
-		io.register_timer(TICK_TOKEN, TICK_TIMEOUT)
-			.expect("Error registering sync timer");
-
-		io.register_timer(PROPAGATE_TOKEN, PROPAGATE_TIMEOUT)
-			.expect("Error registering transaction propagation timer");
-	}
-
-	fn read(&self, io: &NetworkContext, peer: &NodeIndex, _packet_id: u8, data: &[u8]) {
-		self.protocol.handle_packet(&mut NetSyncIo::new(io), *peer, data);
-	}
-
-	fn connected(&self, io: &NetworkContext, peer: &NodeIndex) {
-		self.protocol.on_peer_connected(&mut NetSyncIo::new(io), *peer);
-	}
-
-	fn disconnected(&self, io: &NetworkContext, peer: &NodeIndex) {
-		self.protocol.on_peer_disconnected(&mut NetSyncIo::new(io), *peer);
-	}
-
-	fn timeout(&self, io: &NetworkContext, timer: TimerToken) {
-		match timer {
-			TICK_TOKEN => self.protocol.tick(&mut NetSyncIo::new(io)),
-			PROPAGATE_TOKEN => self.protocol.propagate_extrinsics(&mut NetSyncIo::new(io)),
-			_ => {}
-		}
+		let network = self.network.lock();
+		let ret = network
+			.listeners()
+			.next()
+			.map(|addr| {
+				let mut addr = addr.clone();
+				addr.append(Libp2pProtocol::P2p(network.peer_id().clone().into()));
+				addr.to_string()
+			});
+		ret
 	}
 }
 
@@ -317,26 +234,170 @@ pub trait ManageNetwork: Send + Sync {
 	/// Set to deny unreserved peers to connect
 	fn deny_unreserved_peers(&self);
 	/// Remove reservation for the peer
-	fn remove_reserved_peer(&self, peer: String) -> Result<(), String>;
+	fn remove_reserved_peer(&self, peer: PeerId);
 	/// Add reserved peer
 	fn add_reserved_peer(&self, peer: String) -> Result<(), String>;
 }
 
-
 impl<B: BlockT + 'static, S: Specialization<B>, H: ExHashT> ManageNetwork for Service<B, S, H> {
 	fn accept_unreserved_peers(&self) {
-		self.network.set_non_reserved_mode(NonReservedPeerMode::Accept);
+		self.network.lock().accept_unreserved_peers();
 	}
 
 	fn deny_unreserved_peers(&self) {
-		self.network.set_non_reserved_mode(NonReservedPeerMode::Deny);
+		// This method can disconnect nodes, in which case we have to properly close them in the
+		// protocol.
+		let disconnected = self.network.lock().deny_unreserved_peers();
+		let mut net_sync = NetSyncIo::new(&self.network, self.protocol_id);
+		for node_index in disconnected {
+			self.handler.on_peer_disconnected(&mut net_sync, node_index)
+		}
 	}
 
-	fn remove_reserved_peer(&self, peer: String) -> Result<(), String> {
-		self.network.remove_reserved_peer(&peer).map_err(|e| format!("{:?}", e))
+	fn remove_reserved_peer(&self, peer: PeerId) {
+		// This method can disconnect a node, in which case we have to properly close it in the
+		// protocol.
+		let disconnected = self.network.lock().remove_reserved_peer(peer);
+		if let Some(node_index) = disconnected {
+			let mut net_sync = NetSyncIo::new(&self.network, self.protocol_id);
+			self.handler.on_peer_disconnected(&mut net_sync, node_index)
+		}
 	}
 
 	fn add_reserved_peer(&self, peer: String) -> Result<(), String> {
-		self.network.add_reserved_peer(&peer).map_err(|e| format!("{:?}", e))
+		let (addr, peer_id) = parse_str_addr(&peer).map_err(|e| format!("{:?}", e))?;
+		self.network.lock().add_reserved_peer(addr, peer_id);
+		Ok(())
 	}
 }
+
+/// Starts the background thread that handles the networking.
+fn start_thread<B: BlockT + 'static, S: Specialization<B>, H: ExHashT>(
+	config: NetworkConfiguration,
+	protocol: Arc<Protocol<B, S, H>>,
+	registered: RegisteredProtocol,
+) -> Result<((oneshot::Sender<()>, thread::JoinHandle<()>), Arc<Mutex<NetworkService>>), Error> {
+	let protocol_id = registered.id();
+
+	// Start the main service.
+	let service = match start_service(config, Some(registered)) {
+		Ok(service) => Arc::new(Mutex::new(service)),
+		Err(err) => {
+			match err.kind() {
+				ErrorKind::Io(ref e) if e.kind() == io::ErrorKind::AddrInUse =>
+					warn!("Network port is already in use, make sure that another instance of Substrate client is not running or change the port using the --port option."),
+				_ => warn!("Error starting network: {}", err),
+			};
+			return Err(err.into())
+		},
+	};
+
+	let (close_tx, close_rx) = oneshot::channel();
+	let service_clone = service.clone();
+	let mut runtime = Runtime::new()?;
+	let thread = thread::Builder::new().name("network".to_string()).spawn(move || {
+		let fut = run_thread(service_clone, protocol, protocol_id)
+			.select(close_rx.then(|_| Ok(())))
+			.map(|(val, _)| val)
+			.map_err(|(err,_ )| err);
+
+		// Note that we use `block_on` and not `block_on_all` because we want to kill the thread
+		// instantly if `close_rx` receives something.
+		match runtime.block_on(fut) {
+			Ok(()) => debug!(target: "sub-libp2p", "Networking thread finished"),
+			Err(err) => error!(target: "sub-libp2p", "Error while running libp2p: {:?}", err),
+		};
+	})?;
+
+	Ok(((close_tx, thread), service))
+}
+
+/// Runs the background thread that handles the networking.
+fn run_thread<B: BlockT + 'static, S: Specialization<B>, H: ExHashT>(
+	network_service: Arc<Mutex<NetworkService>>,
+	protocol: Arc<Protocol<B, S, H>>,
+	protocol_id: ProtocolId,
+) -> impl Future<Item = (), Error = io::Error> {
+	// Interval for performing maintenance on the protocol handler.
+	let tick = Interval::new_interval(TICK_TIMEOUT)
+		.for_each({
+			let protocol = protocol.clone();
+			let network_service = network_service.clone();
+			move |_| {
+				protocol.tick(&mut NetSyncIo::new(&network_service, protocol_id));
+				Ok(())
+			}
+		})
+		.then(|res| {
+			match res {
+				Ok(()) => (),
+				Err(err) => error!("Error in the propagation timer: {:?}", err),
+			};
+			Ok(())
+		});
+
+	// Interval at which we gossip extrinsics over the network.
+	let propagate = Interval::new_interval(PROPAGATE_TIMEOUT)
+		.for_each({
+			let protocol = protocol.clone();
+			let network_service = network_service.clone();
+			move |_| {
+				protocol.propagate_extrinsics(&mut NetSyncIo::new(&network_service, protocol_id));
+				Ok(())
+			}
+		})
+		.then(|res| {
+			match res {
+				Ok(()) => (),
+				Err(err) => error!("Error in the propagation timer: {:?}", err),
+			};
+			Ok(())
+		});
+
+	// The network service produces events about what happens on the network. Let's process them.
+	let network_service2 = network_service.clone();
+	let network = stream::poll_fn(move || network_service2.lock().poll()).for_each(move |event| {
+		let mut net_sync = NetSyncIo::new(&network_service, protocol_id);
+
+		match event {
+			NetworkServiceEvent::NodeClosed { node_index, closed_custom_protocols } => {
+				if !closed_custom_protocols.is_empty() {
+					debug_assert_eq!(closed_custom_protocols, &[protocol_id]);
+					protocol.on_peer_disconnected(&mut net_sync, node_index);
+				}
+			}
+			NetworkServiceEvent::ClosedCustomProtocols { node_index, protocols } => {
+				if !protocols.is_empty() {
+					debug_assert_eq!(protocols, &[protocol_id]);
+					protocol.on_peer_disconnected(&mut net_sync, node_index);
+				}
+			}
+			NetworkServiceEvent::OpenedCustomProtocol { node_index, version, .. } => {
+				debug_assert_eq!(version, protocol::CURRENT_VERSION as u8);
+				protocol.on_peer_connected(&mut net_sync, node_index);
+			}
+			NetworkServiceEvent::ClosedCustomProtocol { node_index, .. } => {
+				protocol.on_peer_disconnected(&mut net_sync, node_index);
+			}
+			NetworkServiceEvent::CustomMessage { node_index, data, .. } => {
+				protocol.handle_packet(&mut net_sync, node_index, &data);
+			}
+		};
+
+		Ok(())
+	});
+
+	// Merge all futures into one.
+	let futures: Vec<Box<Future<Item = (), Error = io::Error> + Send>> = vec![
+		Box::new(tick) as Box<_>,
+		Box::new(propagate) as Box<_>,
+		Box::new(network) as Box<_>
+	];
+
+	futures::select_all(futures)
+		.and_then(move |_| {
+			debug!("Networking ended");
+			Ok(())
+		})
+		.map_err(|(r, _, _)| r)
+}

substrate/core/network/src/test/mod.rs

@@ -31,7 +31,7 @@ use protocol::{Context, Protocol};
 use primitives::{Blake2Hasher};
 use config::ProtocolConfig;
 use service::TransactionPool;
-use network_libp2p::{NodeIndex, SessionInfo, Severity};
+use network_libp2p::{NodeIndex, PeerId, Severity};
 use keyring::Keyring;
 use codec::{Encode, Decode};
 use import_queue::SyncImportQueue;
@@ -79,7 +79,6 @@ pub struct TestIo<'p> {
 	queue: &'p RwLock<VecDeque<TestPacket>>,
 	pub to_disconnect: HashSet<NodeIndex>,
 	packets: Vec<TestPacket>,
-	peers_info: HashMap<NodeIndex, String>,
 	_sender: Option<NodeIndex>,
 }
 
@@ -90,7 +89,6 @@ impl<'p> TestIo<'p> where {
 			_sender: sender,
 			to_disconnect: HashSet::new(),
 			packets: Vec::new(),
-			peers_info: HashMap::new(),
 		}
 	}
 }
@@ -106,10 +104,6 @@ impl<'p> SyncIo for TestIo<'p> {
 		self.to_disconnect.insert(who);
 	}
 
-	fn is_expired(&self) -> bool {
-		false
-	}
-
 	fn send(&mut self, who: NodeIndex, data: Vec<u8>) {
 		self.packets.push(TestPacket {
 			data: data,
@@ -117,13 +111,11 @@ impl<'p> SyncIo for TestIo<'p> {
 		});
 	}
 
-	fn peer_info(&self, who: NodeIndex) -> String {
-		self.peers_info.get(&who)
-			.cloned()
-			.unwrap_or_else(|| who.to_string())
+	fn peer_debug_info(&self, _who: NodeIndex) -> String {
+		"unknown".to_string()
 	}
 
-	fn peer_session_info(&self, _peer_id: NodeIndex) -> Option<SessionInfo> {
+	fn peer_id(&self, _peer_id: NodeIndex) -> Option<PeerId> {
 		None
 	}
 }