libp2p-next (#5278)

* Adapt to rust-libp2p#1440.

* Further adapt to libp2p/master.

* Update to libp2p-0.17

* Finishing touches.

* Remove stray TODO.

* Incorporate review feedback.

* Remove unused import.

substrate/client/network/src/protocol/block_requests.rs

@@ -35,6 +35,7 @@ use libp2p::{
 		ConnectedPoint,
 		Multiaddr,
 		PeerId,
+		connection::ConnectionId,
 		upgrade::{InboundUpgrade, ReadOneError, UpgradeInfo, Negotiated},
 		upgrade::{DeniedUpgrade, read_one, write_one}
 	},
@@ -43,6 +44,7 @@ use libp2p::{
 		NetworkBehaviour,
 		NetworkBehaviourAction,
 		OneShotHandler,
+		OneShotHandlerConfig,
 		PollParameters,
 		SubstreamProtocol
 	}
@@ -257,20 +259,27 @@ where
 			max_request_len: self.config.max_request_len,
 			protocol: self.config.protocol.clone(),
 		};
-		OneShotHandler::new(SubstreamProtocol::new(p), self.config.inactivity_timeout)
+		let mut cfg = OneShotHandlerConfig::default();
+		cfg.inactive_timeout = self.config.inactivity_timeout;
+		OneShotHandler::new(SubstreamProtocol::new(p), cfg)
 	}
 
 	fn addresses_of_peer(&mut self, _: &PeerId) -> Vec<Multiaddr> {
 		Vec::new()
 	}
 
-	fn inject_connected(&mut self, _peer: PeerId, _info: ConnectedPoint) {
+	fn inject_connected(&mut self, _peer: &PeerId) {
 	}
 
-	fn inject_disconnected(&mut self, _peer: &PeerId, _info: ConnectedPoint) {
+	fn inject_disconnected(&mut self, _peer: &PeerId) {
 	}
 
-	fn inject_node_event(&mut self, peer: PeerId, Request(request, mut stream): Request<NegotiatedSubstream>) {
+	fn inject_event(
+		&mut self,
+		peer: PeerId,
+		connection: ConnectionId,
+		Request(request, mut stream): Request<NegotiatedSubstream>
+	) {
 		match self.on_block_request(&peer, &request) {
 			Ok(res) => {
 				log::trace!("enqueueing block response for peer {} with {} blocks", peer, res.blocks.len());

substrate/client/network/src/protocol/generic_proto/handler/group.rs

@@ -143,7 +143,7 @@ impl IntoProtocolsHandler for NotifsHandlerProto {
 }
 
 /// Event that can be received by a `NotifsHandler`.
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub enum NotifsHandlerIn {
 	/// The node should start using custom protocols.
 	Enable,
@@ -181,13 +181,18 @@ pub enum NotifsHandlerIn {
 /// Event that can be emitted by a `NotifsHandler`.
 #[derive(Debug)]
 pub enum NotifsHandlerOut {
-	/// Opened the substreams with the remote.
-	Open,
+	/// The connection is open for custom protocols.
+	Open {
+		/// The endpoint of the connection that is open for custom protocols.
+		endpoint: ConnectedPoint,
+	},
 
-	/// Closed the substreams with the remote.
+	/// The connection is closed for custom protocols.
 	Closed {
-		/// Reason why the substream closed, for diagnostic purposes.
+		/// The reason for closing, for diagnostic purposes.
 		reason: Cow<'static, str>,
+		/// The endpoint of the connection that closed for custom protocols.
+		endpoint: ConnectedPoint,
 	},
 
 	/// Received a non-gossiping message on the legacy substream.
@@ -497,13 +502,13 @@ impl ProtocolsHandler for NotifsHandler {
 						protocol: protocol.map_upgrade(EitherUpgrade::B),
 						info: None,
 					}),
-				ProtocolsHandlerEvent::Custom(LegacyProtoHandlerOut::CustomProtocolOpen { .. }) =>
+				ProtocolsHandlerEvent::Custom(LegacyProtoHandlerOut::CustomProtocolOpen { endpoint, .. }) =>
 					return Poll::Ready(ProtocolsHandlerEvent::Custom(
-						NotifsHandlerOut::Open
+						NotifsHandlerOut::Open { endpoint }
 					)),
-				ProtocolsHandlerEvent::Custom(LegacyProtoHandlerOut::CustomProtocolClosed { reason }) =>
+				ProtocolsHandlerEvent::Custom(LegacyProtoHandlerOut::CustomProtocolClosed { endpoint, reason }) =>
 					return Poll::Ready(ProtocolsHandlerEvent::Custom(
-						NotifsHandlerOut::Closed { reason }
+						NotifsHandlerOut::Closed { endpoint, reason }
 					)),
 				ProtocolsHandlerEvent::Custom(LegacyProtoHandlerOut::CustomMessage { message }) =>
 					return Poll::Ready(ProtocolsHandlerEvent::Custom(

substrate/client/network/src/protocol/generic_proto/handler/legacy.rs

@@ -40,9 +40,8 @@ use std::{pin::Pin, task::{Context, Poll}};
 /// it is turned into a `LegacyProtoHandler`. It then handles all communications that are specific
 /// to Substrate on that single connection.
 ///
-/// Note that there can be multiple instance of this struct simultaneously for same peer. However
-/// if that happens, only one main instance can communicate with the outer layers of the code. In
-/// other words, the outer layers of the code only ever see one handler.
+/// Note that there can be multiple instance of this struct simultaneously for same peer,
+/// if there are multiple established connections to the peer.
 ///
 /// ## State of the handler
 ///
@@ -61,6 +60,7 @@ use std::{pin::Pin, task::{Context, Poll}};
 /// these states. For example, if the handler reports a network misbehaviour, it will close the
 /// substreams but it is the role of the user to send a `Disabled` event if it wants the connection
 /// to close. Otherwise, the handler will try to reopen substreams.
+///
 /// The handler starts in the "Initializing" state and must be transitionned to Enabled or Disabled
 /// as soon as possible.
 ///
@@ -111,7 +111,7 @@ impl IntoProtocolsHandler for LegacyProtoHandlerProto {
 	fn into_handler(self, remote_peer_id: &PeerId, connected_point: &ConnectedPoint) -> Self::Handler {
 		LegacyProtoHandler {
 			protocol: self.protocol,
-			endpoint: connected_point.to_endpoint(),
+			endpoint: connected_point.clone(),
 			remote_peer_id: remote_peer_id.clone(),
 			state: ProtocolState::Init {
 				substreams: SmallVec::new(),
@@ -136,7 +136,7 @@ pub struct LegacyProtoHandler {
 
 	/// Whether we are the connection dialer or listener. Used to determine who, between the local
 	/// node and the remote node, has priority.
-	endpoint: Endpoint,
+	endpoint: ConnectedPoint,
 
 	/// Queue of events to send to the outside.
 	///
@@ -218,12 +218,16 @@ pub enum LegacyProtoHandlerOut {
 	CustomProtocolOpen {
 		/// Version of the protocol that has been opened.
 		version: u8,
+		/// The connected endpoint.
+		endpoint: ConnectedPoint,
 	},
 
 	/// Closed a custom protocol with the remote.
 	CustomProtocolClosed {
 		/// Reason why the substream closed, for diagnostic purposes.
 		reason: Cow<'static, str>,
+		/// The connected endpoint.
+		endpoint: ConnectedPoint,
 	},
 
 	/// Receives a message on a custom protocol substream.
@@ -272,7 +276,7 @@ impl LegacyProtoHandler {
 
 			ProtocolState::Init { substreams: incoming, .. } => {
 				if incoming.is_empty() {
-					if let Endpoint::Dialer = self.endpoint {
+					if let ConnectedPoint::Dialer { .. } = self.endpoint {
 						self.events_queue.push(ProtocolsHandlerEvent::OutboundSubstreamRequest {
 							protocol: SubstreamProtocol::new(self.protocol.clone()),
 							info: (),
@@ -281,10 +285,10 @@ impl LegacyProtoHandler {
 					ProtocolState::Opening {
 						deadline: Delay::new(Duration::from_secs(60))
 					}
-
 				} else {
 					let event = LegacyProtoHandlerOut::CustomProtocolOpen {
-						version: incoming[0].protocol_version()
+						version: incoming[0].protocol_version(),
+						endpoint: self.endpoint.clone()
 					};
 					self.events_queue.push(ProtocolsHandlerEvent::Custom(event));
 					ProtocolState::Normal {
@@ -404,6 +408,7 @@ impl LegacyProtoHandler {
 							if substreams.is_empty() {
 								let event = LegacyProtoHandlerOut::CustomProtocolClosed {
 									reason: "All substreams have been closed by the remote".into(),
+									endpoint: self.endpoint.clone()
 								};
 								self.state = ProtocolState::Disabled {
 									shutdown: shutdown.into_iter().collect(),
@@ -416,6 +421,7 @@ impl LegacyProtoHandler {
 							if substreams.is_empty() {
 								let event = LegacyProtoHandlerOut::CustomProtocolClosed {
 									reason: format!("Error on the last substream: {:?}", err).into(),
+									endpoint: self.endpoint.clone()
 								};
 								self.state = ProtocolState::Disabled {
 									shutdown: shutdown.into_iter().collect(),
@@ -479,7 +485,8 @@ impl LegacyProtoHandler {
 
 			ProtocolState::Opening { .. } => {
 				let event = LegacyProtoHandlerOut::CustomProtocolOpen {
-					version: substream.protocol_version()
+					version: substream.protocol_version(),
+					endpoint: self.endpoint.clone()
 				};
 				self.events_queue.push(ProtocolsHandlerEvent::Custom(event));
 				ProtocolState::Normal {

substrate/client/network/src/protocol/generic_proto/handler/notif_in.rs

@@ -72,7 +72,7 @@ pub struct NotifsInHandler {
 }
 
 /// Event that can be received by a `NotifsInHandler`.
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub enum NotifsInHandlerIn {
 	/// Can be sent back as a response to an `OpenRequest`. Contains the status message to send
 	/// to the remote.

substrate/client/network/src/protocol/generic_proto/handler/notif_out.rs

@@ -33,7 +33,7 @@ use libp2p::swarm::{
 	SubstreamProtocol,
 	NegotiatedSubstream,
 };
-use log::error;
+use log::{debug, warn, error};
 use prometheus_endpoint::Histogram;
 use smallvec::SmallVec;
 use std::{borrow::Cow, fmt, mem, pin::Pin, task::{Context, Poll}, time::Duration};
@@ -280,7 +280,7 @@ impl ProtocolsHandler for NotifsOutHandler {
 						// be recovered. When in doubt, let's drop the existing substream and
 						// open a new one.
 						if sub.close().now_or_never().is_none() {
-							log::warn!(
+							warn!(
 								target: "sub-libp2p",
 								"📞 Improperly closed outbound notifications substream"
 							);
@@ -293,16 +293,22 @@ impl ProtocolsHandler for NotifsOutHandler {
 						});
 						self.state = State::Opening { initial_message };
 					},
-					State::Opening { .. } | State::Refused | State::Open { .. } =>
-						error!("☎️ Tried to enable notifications handler that was already enabled"),
-					State::Poisoned => error!("☎️ Notifications handler in a poisoned state"),
+					st @ State::Opening { .. } | st @ State::Refused | st @ State::Open { .. } => {
+						debug!(target: "sub-libp2p",
+							"Tried to enable notifications handler that was already enabled");
+						self.state = st;
+					}
+					State::Poisoned => error!("Notifications handler in a poisoned state"),
 				}
 			}
 
 			NotifsOutHandlerIn::Disable => {
 				match mem::replace(&mut self.state, State::Poisoned) {
-					State::Disabled | State::DisabledOpen(_) | State::DisabledOpening =>
-						error!("☎️ Tried to disable notifications handler that was already disabled"),
+					st @ State::Disabled | st @ State::DisabledOpen(_) | st @ State::DisabledOpening => {
+						debug!(target: "sub-libp2p",
+							"Tried to disable notifications handler that was already disabled");
+						self.state = st;
+					}
 					State::Opening { .. } => self.state = State::DisabledOpening,
 					State::Refused => self.state = State::Disabled,
 					State::Open { substream, .. } => self.state = State::DisabledOpen(substream),
@@ -313,7 +319,7 @@ impl ProtocolsHandler for NotifsOutHandler {
 			NotifsOutHandlerIn::Send(msg) =>
 				if let State::Open { substream, .. } = &mut self.state {
 					if substream.push_message(msg).is_err() {
-						log::warn!(
+						warn!(
 							target: "sub-libp2p",
 							"📞 Notifications queue with peer {} is full, dropped message (protocol: {:?})",
 							self.peer_id,
@@ -325,7 +331,7 @@ impl ProtocolsHandler for NotifsOutHandler {
 					}
 				} else {
 					// This is an API misuse.
-					log::warn!(
+					warn!(
 						target: "sub-libp2p",
 						"📞 Tried to send a notification on a disabled handler"
 					);

substrate/client/network/src/protocol/generic_proto/tests.rs

@@ -18,7 +18,7 @@
 
 use futures::{prelude::*, ready};
 use codec::{Encode, Decode};
-use libp2p::core::nodes::listeners::ListenerId;
+use libp2p::core::connection::{ConnectionId, ListenerId};
 use libp2p::core::ConnectedPoint;
 use libp2p::swarm::{Swarm, ProtocolsHandler, IntoProtocolsHandler};
 use libp2p::swarm::{PollParameters, NetworkBehaviour, NetworkBehaviourAction};
@@ -148,20 +148,29 @@ impl NetworkBehaviour for CustomProtoWithAddr {
 		list
 	}
 
-	fn inject_connected(&mut self, peer_id: PeerId, endpoint: ConnectedPoint) {
-		self.inner.inject_connected(peer_id, endpoint)
+	fn inject_connected(&mut self, peer_id: &PeerId) {
+		self.inner.inject_connected(peer_id)
 	}
 
-	fn inject_disconnected(&mut self, peer_id: &PeerId, endpoint: ConnectedPoint) {
-		self.inner.inject_disconnected(peer_id, endpoint)
+	fn inject_disconnected(&mut self, peer_id: &PeerId) {
+		self.inner.inject_disconnected(peer_id)
 	}
 
-	fn inject_node_event(
+	fn inject_connection_established(&mut self, peer_id: &PeerId, conn: &ConnectionId, endpoint: &ConnectedPoint) {
+		self.inner.inject_connection_established(peer_id, conn, endpoint)
+	}
+
+	fn inject_connection_closed(&mut self, peer_id: &PeerId, conn: &ConnectionId, endpoint: &ConnectedPoint) {
+		self.inner.inject_connection_closed(peer_id, conn, endpoint)
+	}
+
+	fn inject_event(
 		&mut self,
 		peer_id: PeerId,
+		connection: ConnectionId,
 		event: <<Self::ProtocolsHandler as IntoProtocolsHandler>::Handler as ProtocolsHandler>::OutEvent
 	) {
-		self.inner.inject_node_event(peer_id, event)
+		self.inner.inject_event(peer_id, connection, event)
 	}
 
 	fn poll(
@@ -177,10 +186,6 @@ impl NetworkBehaviour for CustomProtoWithAddr {
 		self.inner.poll(cx, params)
 	}
 
-	fn inject_replaced(&mut self, peer_id: PeerId, closed_endpoint: ConnectedPoint, new_endpoint: ConnectedPoint) {
-		self.inner.inject_replaced(peer_id, closed_endpoint, new_endpoint)
-	}
-
 	fn inject_addr_reach_failure(&mut self, peer_id: Option<&PeerId>, addr: &Multiaddr, error: &dyn std::error::Error) {
 		self.inner.inject_addr_reach_failure(peer_id, addr, error)
 	}
@@ -205,8 +210,8 @@ impl NetworkBehaviour for CustomProtoWithAddr {
 		self.inner.inject_listener_error(id, err);
 	}
 
-	fn inject_listener_closed(&mut self, id: ListenerId) {
-		self.inner.inject_listener_closed(id);
+	fn inject_listener_closed(&mut self, id: ListenerId, reason: Result<(), &io::Error>) {
+		self.inner.inject_listener_closed(id, reason);
 	}
 }
 

substrate/client/network/src/protocol/light_client_handler.rs

@@ -37,6 +37,7 @@ use libp2p::{
 		ConnectedPoint,
 		Multiaddr,
 		PeerId,
+		connection::ConnectionId,
 		upgrade::{InboundUpgrade, ReadOneError, UpgradeInfo, Negotiated},
 		upgrade::{OutboundUpgrade, read_one, write_one}
 	},
@@ -44,9 +45,11 @@ use libp2p::{
 		NegotiatedSubstream,
 		NetworkBehaviour,
 		NetworkBehaviourAction,
+		NotifyHandler,
 		OneShotHandler,
+		OneShotHandlerConfig,
 		PollParameters,
-		SubstreamProtocol
+		SubstreamProtocol,
 	}
 };
 use nohash_hasher::IntMap;
@@ -58,6 +61,7 @@ use sp_core::{
 	storage::{ChildInfo, StorageKey},
 	hexdisplay::HexDisplay,
 };
+use smallvec::SmallVec;
 use sp_blockchain::{Error as ClientError};
 use sp_runtime::{
 	traits::{Block, Header, NumberFor, Zero},
@@ -237,25 +241,39 @@ struct RequestWrapper<B: Block, P> {
 	retries: usize,
 	/// The actual request.
 	request: Request<B>,
-	/// Peer information, e.g. `PeerId`.
-	peer: P
+	/// The peer to send the request to, e.g. `PeerId`.
+	peer: P,
+	/// The connection to use for sending the request.
+	connection: Option<ConnectionId>,
 }
 
 /// Information we have about some peer.
 #[derive(Debug)]
 struct PeerInfo<B: Block> {
-	address: Multiaddr,
+	connections: SmallVec<[(ConnectionId, Multiaddr); crate::MAX_CONNECTIONS_PER_PEER]>,
 	best_block: Option<NumberFor<B>>,
 	status: PeerStatus,
 }
 
+impl<B: Block> Default for PeerInfo<B> {
+	fn default() -> Self {
+		PeerInfo {
+			connections: SmallVec::new(),
+			best_block: None,
+			status: PeerStatus::Idle,
+		}
+	}
+}
+
+type RequestId = u64;
+
 /// A peer is either idle or busy processing a request from us.
 #[derive(Debug, Clone, PartialEq, Eq)]
 enum PeerStatus {
 	/// The peer is available.
 	Idle,
 	/// We wait for the peer to return us a response for the given request ID.
-	BusyWith(u64),
+	BusyWith(RequestId),
 }
 
 /// The light client handler behaviour.
@@ -273,9 +291,9 @@ pub struct LightClientHandler<B: Block> {
 	/// Pending (local) requests.
 	pending_requests: VecDeque<RequestWrapper<B, ()>>,
 	/// Requests on their way to remote peers.
-	outstanding: IntMap<u64, RequestWrapper<B, PeerId>>,
+	outstanding: IntMap<RequestId, RequestWrapper<B, PeerId>>,
 	/// (Local) Request ID counter
-	next_request_id: u64,
+	next_request_id: RequestId,
 	/// Handle to use for reporting misbehaviour of peers.
 	peerset: sc_peerset::PeersetHandle,
 }
@@ -323,35 +341,18 @@ where
 			retries: retries(&req),
 			request: req,
 			peer: (), // we do not know the peer yet
+			connection: None,
 		};
 		self.pending_requests.push_back(rw);
 		Ok(())
 	}
 
-	fn next_request_id(&mut self) -> u64 {
+	fn next_request_id(&mut self) -> RequestId {
 		let id = self.next_request_id;
 		self.next_request_id += 1;
 		id
 	}
 
-	// Iterate over peers known to possess a certain block.
-	fn idle_peers_with_block(&mut self, num: NumberFor<B>) -> impl Iterator<Item = PeerId> + '_ {
-		self.peers.iter()
-			.filter(move |(_, info)| {
-				info.status == PeerStatus::Idle && info.best_block >= Some(num)
-			})
-			.map(|(peer, _)| peer.clone())
-	}
-
-	// Iterate over peers without a known block.
-	fn idle_peers_with_unknown_block(&mut self) -> impl Iterator<Item = PeerId> + '_ {
-		self.peers.iter()
-			.filter(|(_, info)| {
-				info.status == PeerStatus::Idle && info.best_block.is_none()
-			})
-			.map(|(peer, _)| peer.clone())
-	}
-
 	/// Remove the given peer.
 	///
 	/// If we have a request to this peer in flight, we move it back to
@@ -364,12 +365,50 @@ where
 				retries: rw.retries,
 				request: rw.request,
 				peer: (), // need to find another peer
+				connection: None,
 			};
 			self.pending_requests.push_back(rw);
 		}
 		self.peers.remove(peer);
 	}
 
+	/// Prepares a request by selecting a suitable peer and connection to send it to.
+	///
+	/// If there is currently no suitable peer for the request, the given request
+	/// is returned as `Err`.
+	fn prepare_request(&self, req: RequestWrapper<B, ()>)
+		-> Result<(PeerId, RequestWrapper<B, PeerId>), RequestWrapper<B, ()>>
+	{
+		let number = required_block(&req.request);
+
+		let mut peer = None;
+		for (peer_id, peer_info) in self.peers.iter() {
+			if peer_info.status == PeerStatus::Idle {
+				match peer_info.best_block {
+					Some(n) => if n >= number {
+						peer = Some((peer_id, peer_info));
+						break
+					},
+					None => peer = Some((peer_id, peer_info))
+				}
+			}
+		}
+
+		if let Some((peer_id, peer_info)) = peer {
+			let connection = peer_info.connections.iter().next().map(|(id, _)| *id);
+			let rw = RequestWrapper {
+				timestamp: req.timestamp,
+				retries: req.retries,
+				request: req.request,
+				peer: peer_id.clone(),
+				connection,
+			};
+			Ok((peer_id.clone(), rw))
+		} else {
+			Err(req)
+		}
+	}
+
 	/// Process a local request's response from remote.
 	///
 	/// If successful, this will give us the actual, checked data we should be
@@ -723,38 +762,68 @@ where
 			max_request_size: self.config.max_request_size,
 			protocol: self.config.light_protocol.clone(),
 		};
-		OneShotHandler::new(SubstreamProtocol::new(p), self.config.inactivity_timeout)
+		let mut cfg = OneShotHandlerConfig::default();
+		cfg.inactive_timeout = self.config.inactivity_timeout;
+		OneShotHandler::new(SubstreamProtocol::new(p), cfg)
 	}
 
 	fn addresses_of_peer(&mut self, peer: &PeerId) -> Vec<Multiaddr> {
 		self.peers.get(peer)
-			.map(|info| vec![info.address.clone()])
+			.map(|info| info.connections.iter().map(|(_, a)| a.clone()).collect())
 			.unwrap_or_default()
 	}
 
-	fn inject_connected(&mut self, peer: PeerId, info: ConnectedPoint) {
+	fn inject_connected(&mut self, peer: &PeerId) {
+	}
+
+	fn inject_connection_established(&mut self, peer: &PeerId, conn: &ConnectionId, info: &ConnectedPoint) {
+		let peer_address = match info {
+			ConnectedPoint::Listener { send_back_addr, .. } => send_back_addr.clone(),
+			ConnectedPoint::Dialer { address } => address.clone()
+		};
+
+		log::trace!("peer {} connected with address {}", peer, peer_address);
+
+		let entry = self.peers.entry(peer.clone()).or_default();
+		entry.connections.push((*conn, peer_address));
+	}
+
+	fn inject_disconnected(&mut self, peer: &PeerId) {
+		log::trace!("peer {} disconnected", peer);
+		self.remove_peer(peer)
+	}
+
+	fn inject_connection_closed(&mut self, peer: &PeerId, conn: &ConnectionId, info: &ConnectedPoint) {
 		let peer_address = match info {
 			ConnectedPoint::Listener { send_back_addr, .. } => send_back_addr,
 			ConnectedPoint::Dialer { address } => address
 		};
 
-		log::trace!("peer {} connected with address {}", peer, peer_address);
+		log::trace!("connection to peer {} closed: {}", peer, peer_address);
 
-		let info = PeerInfo {
-			address: peer_address,
-			best_block: None,
-			status: PeerStatus::Idle,
-		};
+		if let Some(info) = self.peers.get_mut(peer) {
+			info.connections.retain(|(c, _)| c != conn)
+		}
 
-		self.peers.insert(peer, info);
+		// Add any outstanding requests on the closed connection back to the
+		// pending requests.
+		if let Some(id) = self.outstanding.iter()
+			.find(|(_, rw)| &rw.peer == peer && rw.connection == Some(*conn)) // (*)
+			.map(|(id, _)| *id)
+		{
+			let rw = self.outstanding.remove(&id).expect("by (*)");
+			let rw = RequestWrapper {
+				timestamp: rw.timestamp,
+				retries: rw.retries,
+				request: rw.request,
+				peer: (), // need to find another peer
+				connection: None,
+			};
+			self.pending_requests.push_back(rw);
+		}
 	}
 
-	fn inject_disconnected(&mut self, peer: &PeerId, _: ConnectedPoint) {
-		log::trace!("peer {} disconnected", peer);
-		self.remove_peer(peer)
-	}
-
-	fn inject_node_event(&mut self, peer: PeerId, event: Event<NegotiatedSubstream>) {
+	fn inject_event(&mut self, peer: PeerId, conn: ConnectionId, event: Event<NegotiatedSubstream>) {
 		match event {
 			// An incoming request from remote has been received.
 			Event::Request(request, mut stream) => {
@@ -800,9 +869,10 @@ where
 			// A response to one of our own requests has been received.
 			Event::Response(id, response) => {
 				if let Some(request) = self.outstanding.remove(&id) {
-					// We first just check if the response originates from the expected peer.
+					// We first just check if the response originates from the expected peer
+					// and connection.
 					if request.peer != peer {
-						log::debug!("was expecting response from {} instead of {}", request.peer, peer);
+						log::debug!("Expected response from {} instead of {}.", request.peer, peer);
 						self.outstanding.insert(id, request);
 						self.remove_peer(&peer);
 						self.peerset.report_peer(peer, ReputationChange::new_fatal("response from unexpected peer"));
@@ -834,6 +904,7 @@ where
 									retries: request.retries,
 									request: request.request,
 									peer: (),
+									connection: None,
 								};
 								self.pending_requests.push_back(rw);
 							}
@@ -847,6 +918,7 @@ where
 										retries: request.retries - 1,
 										request: request.request,
 										peer: (),
+										connection: None,
 									};
 									self.pending_requests.push_back(rw)
 								} else {
@@ -886,54 +958,54 @@ where
 				request.timestamp = Instant::now();
 				request.retries -= 1
 			}
-			let number = required_block(&request.request);
-			let available_peer = {
-				let p = self.idle_peers_with_block(number).next();
-				if p.is_none() {
-					self.idle_peers_with_unknown_block().next()
-				} else {
-					p
+
+
+			match self.prepare_request(request) {
+				Err(request) => {
+					self.pending_requests.push_front(request);
+					log::debug!("no peer available to send request to");
+					break
 				}
-			};
-			if let Some(peer) = available_peer {
-				let buf = match serialize_request(&request.request) {
-					Ok(b) => b,
-					Err(e) => {
-						log::debug!("failed to serialize request: {}", e);
-						send_reply(Err(ClientError::RemoteFetchFailed), request.request);
-						continue;
-					}
-				};
+				Ok((peer, request)) => {
+					let request_bytes = match serialize_request(&request.request) {
+						Ok(bytes) => bytes,
+						Err(error) => {
+							log::debug!("failed to serialize request: {}", error);
+							send_reply(Err(ClientError::RemoteFetchFailed), request.request);
+							continue
+						}
+					};
 
-				let id = self.next_request_id();
-				log::trace!("sending request {} to peer {}", id, peer);
-				let protocol = OutboundProtocol {
-					request: buf,
-					request_id: id,
-					expected: match request.request {
-						Request::Body { .. } => ExpectedResponseTy::Block,
-						_ => ExpectedResponseTy::Light,
-					},
-					max_response_size: self.config.max_response_size,
-					protocol: match request.request {
-						Request::Body { .. } => self.config.block_protocol.clone(),
-						_ => self.config.light_protocol.clone(),
-					},
-				};
-				self.peers.get_mut(&peer).map(|info| info.status = PeerStatus::BusyWith(id));
-				let rw = RequestWrapper {
-					timestamp: request.timestamp,
-					retries: request.retries,
-					request: request.request,
-					peer: peer.clone(),
-				};
-				self.outstanding.insert(id, rw);
-				return Poll::Ready(NetworkBehaviourAction::SendEvent { peer_id: peer, event: protocol })
+					let (expected, protocol) = match request.request {
+						Request::Body { .. } =>
+							(ExpectedResponseTy::Block, self.config.block_protocol.clone()),
+						_ =>
+							(ExpectedResponseTy::Light, self.config.light_protocol.clone()),
+					};
 
-			} else {
-				self.pending_requests.push_front(request);
-				log::debug!("no peer available to send request to");
-				break
+					let peer_id = peer.clone();
+					let handler = request.connection.map_or(NotifyHandler::Any, NotifyHandler::One);
+
+					let request_id = self.next_request_id();
+					self.peers.get_mut(&peer).map(|p| p.status = PeerStatus::BusyWith(request_id));
+					self.outstanding.insert(request_id, request);
+
+					let event = OutboundProtocol {
+						request_id,
+						request: request_bytes,
+						expected,
+						max_response_size: self.config.max_response_size,
+						protocol,
+					};
+
+					log::trace!("sending request {} to peer {}", request_id, peer_id);
+
+					return Poll::Ready(NetworkBehaviourAction::NotifyHandler {
+						peer_id,
+						handler,
+						event,
+					})
+				}
 			}
 		}
 
@@ -959,6 +1031,7 @@ where
 					retries: rw.retries - 1,
 					request: rw.request,
 					peer: (),
+					connection: None,
 				};
 				self.pending_requests.push_back(rw)
 			}
@@ -1097,7 +1170,7 @@ pub enum Event<T> {
 	/// Incoming request from remote and substream to use for the response.
 	Request(api::v1::light::Request, T),
 	/// Incoming response from remote.
-	Response(u64, Response),
+	Response(RequestId, Response),
 }
 
 /// Incoming response from remote.
@@ -1157,7 +1230,7 @@ pub struct OutboundProtocol {
 	/// The serialized protobuf request.
 	request: Vec<u8>,
 	/// Local identifier for the request. Used to associate it with a response.
-	request_id: u64,
+	request_id: RequestId,
 	/// Kind of response expected for this request.
 	expected: ExpectedResponseTy,
 	/// The max. response length in bytes.
@@ -1244,6 +1317,7 @@ mod tests {
 		Multiaddr,
 		core::{
 			ConnectedPoint,
+			connection::ConnectionId,
 			identity,
 			muxing::{StreamMuxerBox, SubstreamRef},
 			transport::{Transport, boxed::Boxed, memory::MemoryTransport},
@@ -1457,10 +1531,12 @@ mod tests {
 		let pset = peerset();
 		let mut behaviour = make_behaviour(true, pset.1, make_config());
 
-		behaviour.inject_connected(peer.clone(), empty_dialer());
+		behaviour.inject_connection_established(&peer, &ConnectionId::new(1), &empty_dialer());
+		behaviour.inject_connected(&peer);
 		assert_eq!(1, behaviour.peers.len());
 
-		behaviour.inject_disconnected(&peer, empty_dialer());
+		behaviour.inject_connection_closed(&peer, &ConnectionId::new(1), &empty_dialer());
+		behaviour.inject_disconnected(&peer);
 		assert_eq!(0, behaviour.peers.len())
 	}
 
@@ -1471,8 +1547,10 @@ mod tests {
 		let pset = peerset();
 		let mut behaviour = make_behaviour(true, pset.1, make_config());
 
-		behaviour.inject_connected(peer0.clone(), empty_dialer());
-		behaviour.inject_connected(peer1.clone(), empty_dialer());
+		behaviour.inject_connection_established(&peer0, &ConnectionId::new(1), &empty_dialer());
+		behaviour.inject_connected(&peer0);
+		behaviour.inject_connection_established(&peer1, &ConnectionId::new(2), &empty_dialer());
+		behaviour.inject_connected(&peer1);
 
 		// We now know about two peers.
 		assert_eq!(HashSet::from_iter(&[peer0.clone(), peer1.clone()]), behaviour.peers.keys().collect::<HashSet<_>>());
@@ -1494,7 +1572,7 @@ mod tests {
 		assert_eq!(1, behaviour.pending_requests.len());
 
 		// The behaviour should now attempt to send the request.
-		assert_matches!(poll(&mut behaviour), Poll::Ready(NetworkBehaviourAction::SendEvent { peer_id, .. }) => {
+		assert_matches!(poll(&mut behaviour), Poll::Ready(NetworkBehaviourAction::NotifyHandler { peer_id, .. }) => {
 			assert!(peer_id == peer0 || peer_id == peer1)
 		});
 
@@ -1534,7 +1612,9 @@ mod tests {
 		let mut behaviour = make_behaviour(false, pset.1, make_config());
 		//                                 ^--- Making sure the response data check fails.
 
-		behaviour.inject_connected(peer.clone(), empty_dialer());
+		let conn = ConnectionId::new(1);
+		behaviour.inject_connection_established(&peer, &conn, &empty_dialer());
+		behaviour.inject_connected(&peer);
 		assert_eq!(1, behaviour.peers.len());
 
 		let chan = oneshot::channel();
@@ -1562,7 +1642,7 @@ mod tests {
 			}
 		};
 
-		behaviour.inject_node_event(peer.clone(), Event::Response(request_id, Response::Light(response)));
+		behaviour.inject_event(peer.clone(), conn, Event::Response(request_id, Response::Light(response)));
 		assert!(behaviour.peers.is_empty());
 
 		poll(&mut behaviour); // More progress
@@ -1578,7 +1658,9 @@ mod tests {
 		let pset = peerset();
 		let mut behaviour = make_behaviour(true, pset.1, make_config());
 
-		behaviour.inject_connected(peer.clone(), empty_dialer());
+		let conn = ConnectionId::new(1);
+		behaviour.inject_connection_established(&peer, &conn, &empty_dialer());
+		behaviour.inject_connected(&peer);
 		assert_eq!(1, behaviour.peers.len());
 		assert_eq!(0, behaviour.pending_requests.len());
 		assert_eq!(0, behaviour.outstanding.len());
@@ -1591,7 +1673,7 @@ mod tests {
 			}
 		};
 
-		behaviour.inject_node_event(peer.clone(), Event::Response(2347895932, Response::Light(response)));
+		behaviour.inject_event(peer.clone(), conn, Event::Response(2347895932, Response::Light(response)));
 
 		assert!(behaviour.peers.is_empty());
 		poll(&mut behaviour);
@@ -1605,7 +1687,9 @@ mod tests {
 		let pset = peerset();
 		let mut behaviour = make_behaviour(true, pset.1, make_config());
 
-		behaviour.inject_connected(peer.clone(), empty_dialer());
+		let conn = ConnectionId::new(1);
+		behaviour.inject_connection_established(&peer, &conn, &empty_dialer());
+		behaviour.inject_connected(&peer);
 		assert_eq!(1, behaviour.peers.len());
 
 		let chan = oneshot::channel();
@@ -1633,7 +1717,7 @@ mod tests {
 			}
 		};
 
-		behaviour.inject_node_event(peer.clone(), Event::Response(request_id, Response::Light(response)));
+		behaviour.inject_event(peer.clone(), conn, Event::Response(request_id, Response::Light(response)));
 		assert!(behaviour.peers.is_empty());
 
 		poll(&mut behaviour); // More progress
@@ -1653,10 +1737,18 @@ mod tests {
 		let mut behaviour = make_behaviour(false, pset.1, make_config());
 		//                                 ^--- Making sure the response data check fails.
 
-		behaviour.inject_connected(peer1.clone(), empty_dialer());
-		behaviour.inject_connected(peer2.clone(), empty_dialer());
-		behaviour.inject_connected(peer3.clone(), empty_dialer());
-		behaviour.inject_connected(peer4.clone(), empty_dialer());
+		let conn1 = ConnectionId::new(1);
+		behaviour.inject_connection_established(&peer1, &conn1, &empty_dialer());
+		behaviour.inject_connected(&peer1);
+		let conn2 = ConnectionId::new(2);
+		behaviour.inject_connection_established(&peer2, &conn2, &empty_dialer());
+		behaviour.inject_connected(&peer2);
+		let conn3 = ConnectionId::new(3);
+		behaviour.inject_connection_established(&peer3, &conn3, &empty_dialer());
+		behaviour.inject_connected(&peer3);
+		let conn4 = ConnectionId::new(3);
+		behaviour.inject_connection_established(&peer4, &conn4, &empty_dialer());
+		behaviour.inject_connected(&peer4);
 		assert_eq!(4, behaviour.peers.len());
 
 		let mut chan = oneshot::channel();
@@ -1671,11 +1763,11 @@ mod tests {
 
 		assert_eq!(1, behaviour.pending_requests.len());
 		assert_eq!(0, behaviour.outstanding.len());
-		assert_matches!(poll(&mut behaviour), Poll::Ready(NetworkBehaviourAction::SendEvent { .. }));
+		assert_matches!(poll(&mut behaviour), Poll::Ready(NetworkBehaviourAction::NotifyHandler { .. }));
 		assert_eq!(0, behaviour.pending_requests.len());
 		assert_eq!(1, behaviour.outstanding.len());
 
-		for _ in 0 .. 3 {
+		for i in 1 ..= 3 {
 			// Construct an invalid response
 			let request_id = *behaviour.outstanding.keys().next().unwrap();
 			let responding_peer = behaviour.outstanding.values().next().unwrap().peer.clone();
@@ -1685,8 +1777,9 @@ mod tests {
 					response: Some(api::v1::light::response::Response::RemoteCallResponse(r))
 				}
 			};
-			behaviour.inject_node_event(responding_peer, Event::Response(request_id, Response::Light(response.clone())));
-			assert_matches!(poll(&mut behaviour), Poll::Ready(NetworkBehaviourAction::SendEvent { .. }));
+			let conn = ConnectionId::new(i);
+			behaviour.inject_event(responding_peer, conn, Event::Response(request_id, Response::Light(response.clone())));
+			assert_matches!(poll(&mut behaviour), Poll::Ready(NetworkBehaviourAction::NotifyHandler { .. }));
 			assert_matches!(chan.1.try_recv(), Ok(None))
 		}
 		// Final invalid response
@@ -1698,7 +1791,7 @@ mod tests {
 				response: Some(api::v1::light::response::Response::RemoteCallResponse(r)),
 			}
 		};
-		behaviour.inject_node_event(responding_peer, Event::Response(request_id, Response::Light(response)));
+		behaviour.inject_event(responding_peer, conn4, Event::Response(request_id, Response::Light(response)));
 		assert_matches!(poll(&mut behaviour), Poll::Pending);
 		assert_matches!(chan.1.try_recv(), Ok(Some(Err(ClientError::RemoteFetchFailed))))
 	}
@@ -1708,7 +1801,9 @@ mod tests {
 		let pset = peerset();
 		let mut behaviour = make_behaviour(true, pset.1, make_config());
 
-		behaviour.inject_connected(peer.clone(), empty_dialer());
+		let conn = ConnectionId::new(1);
+		behaviour.inject_connection_established(&peer, &conn, &empty_dialer());
+		behaviour.inject_connected(&peer);
 		assert_eq!(1, behaviour.peers.len());
 
 		let response = match request {
@@ -1757,12 +1852,12 @@ mod tests {
 
 		assert_eq!(1, behaviour.pending_requests.len());
 		assert_eq!(0, behaviour.outstanding.len());
-		assert_matches!(poll(&mut behaviour), Poll::Ready(NetworkBehaviourAction::SendEvent { .. }));
+		assert_matches!(poll(&mut behaviour), Poll::Ready(NetworkBehaviourAction::NotifyHandler { .. }));
 		assert_eq!(0, behaviour.pending_requests.len());
 		assert_eq!(1, behaviour.outstanding.len());
 		assert_eq!(1, *behaviour.outstanding.keys().next().unwrap());
 
-		behaviour.inject_node_event(peer.clone(), Event::Response(1, Response::Light(response)));
+		behaviour.inject_event(peer.clone(), conn, Event::Response(1, Response::Light(response)));
 
 		poll(&mut behaviour);