Upgrade to libp2p 0.51.3 (#13587)

* client/network: upgrade to libp2p 0.51.0

* make discovery.rs compile

* make peer_info.rs compile

* changes to notifications and request-response proto

* make service.rs compile

* towards making request_responses.rs compile

* make request_responses.rs compile

* make request_responses.rs compile

* fix notifications/behaviour.rs tests

* fix warnings

* remove old code

* allow deprecated code (temporary)

* upgrade to libp2p 0.51.1

* add TODO for behaviour tests

* return empty vec if peer_id is absent

https://github.com/paritytech/substrate/pull/13587#discussion_r1141695167

fyi: I don't really know what the old behaviour was.

* update comment to reflect new defaults

Closes #13338

* Revert "update comment to reflect new defaults"

This reverts commit 7a981abd69308e9d522ec94905f181439a1b1dba.

* remove config.rs (from wrong merge)

* upgrade to libp2p 0.51.2

* fix formatting

* use handle_pending_outbound_connection in networt_state RPC

* update deps

* use re-exports when we use other libp2p packages

* Apply suggestions from code review

Co-authored-by: Dmitry Markin <dmitry@markin.tech>

* format code

* handle potential errors in network_state RPC

* only update libp2p crate

* update libp2p-core

* fix docs

* use libp2p-identity instead of libp2p

where it's possible. libp2p-identity is much smaller, hence makes sense
to use it instead of larger libp2p crate.

* Update client/network/src/discovery.rs

Co-authored-by: Aaro Altonen <48052676+altonen@users.noreply.github.com>

* update Cargo.lock

* add comment for per_connection_event_buffer_size

current value is somewhat arbitrary and needs to be tweaked depending on
memory usage and network worker sleep stats.

* fix link format

* update Cargo.lock

* upgrade to libp2p 0.51.3

* deprecate mplex

* Revert "deprecate mplex"

This reverts commit 9e25820e706e464a0e962a8604861fcb2a7641eb.

* Revert "upgrade to libp2p 0.51.3"

This reverts commit 6544dd4138e2f89517bd7c7281fc78a638ec7040.

* use new libp2p version in `statement` crate

* pin version temporarily

* libp2p 0.51.3

* deprecate mplex

* deprecate legacy noise handshake

* fix build error

* update libp2p-identity

* enable libp2p-identity:ed25519 feature in sc-consensus

* enable ed25519 for peerset as well

---------

Co-authored-by: Dmitry Markin <dmitry@markin.tech>
Co-authored-by: Aaro Altonen <48052676+altonen@users.noreply.github.com>
Co-authored-by: parity-processbot <>

substrate/client/network/src/protocol/notifications/behaviour.rs

@@ -18,7 +18,7 @@
 
 use crate::{
 	protocol::notifications::handler::{
-		self, NotificationsSink, NotifsHandlerIn, NotifsHandlerOut, NotifsHandlerProto,
+		self, NotificationsSink, NotifsHandler, NotifsHandlerIn, NotifsHandlerOut,
 	},
 	types::ProtocolName,
 };
@@ -27,13 +27,13 @@ use bytes::BytesMut;
 use fnv::FnvHashMap;
 use futures::prelude::*;
 use libp2p::{
-	core::{connection::ConnectionId, Multiaddr, PeerId},
+	core::{ConnectedPoint, Endpoint, Multiaddr},
 	swarm::{
 		behaviour::{ConnectionClosed, ConnectionEstablished, DialFailure, FromSwarm},
-		handler::ConnectionHandler,
-		DialError, IntoConnectionHandler, NetworkBehaviour, NetworkBehaviourAction, NotifyHandler,
-		PollParameters,
+		ConnectionDenied, ConnectionId, DialError, NetworkBehaviour, NotifyHandler, PollParameters,
+		THandler, THandlerInEvent, THandlerOutEvent, ToSwarm,
 	},
+	PeerId,
 };
 use log::{error, trace, warn};
 use parking_lot::RwLock;
@@ -136,7 +136,7 @@ pub struct Notifications {
 	next_incoming_index: sc_peerset::IncomingIndex,
 
 	/// Events to produce from `poll()`.
-	events: VecDeque<NetworkBehaviourAction<NotificationsOut, NotifsHandlerProto>>,
+	events: VecDeque<ToSwarm<NotificationsOut, NotifsHandlerIn>>,
 }
 
 /// Configuration for a notifications protocol.
@@ -454,14 +454,14 @@ impl Notifications {
 					trace!(target: "sub-libp2p", "External API <= Closed({}, {:?})", peer_id, set_id);
 					let event =
 						NotificationsOut::CustomProtocolClosed { peer_id: *peer_id, set_id };
-					self.events.push_back(NetworkBehaviourAction::GenerateEvent(event));
+					self.events.push_back(ToSwarm::GenerateEvent(event));
 				}
 
 				for (connec_id, connec_state) in
 					connections.iter_mut().filter(|(_, s)| matches!(s, ConnectionState::Open(_)))
 				{
 					trace!(target: "sub-libp2p", "Handler({:?}, {:?}) <= Close({:?})", peer_id, *connec_id, set_id);
-					self.events.push_back(NetworkBehaviourAction::NotifyHandler {
+					self.events.push_back(ToSwarm::NotifyHandler {
 						peer_id: *peer_id,
 						handler: NotifyHandler::One(*connec_id),
 						event: NotifsHandlerIn::Close { protocol_index: set_id.into() },
@@ -473,7 +473,7 @@ impl Notifications {
 					connections.iter_mut().filter(|(_, s)| matches!(s, ConnectionState::Opening))
 				{
 					trace!(target: "sub-libp2p", "Handler({:?}, {:?}) <= Close({:?})", peer_id, *connec_id, set_id);
-					self.events.push_back(NetworkBehaviourAction::NotifyHandler {
+					self.events.push_back(ToSwarm::NotifyHandler {
 						peer_id: *peer_id,
 						handler: NotifyHandler::One(*connec_id),
 						event: NotifsHandlerIn::Close { protocol_index: set_id.into() },
@@ -515,7 +515,7 @@ impl Notifications {
 					.filter(|(_, s)| matches!(s, ConnectionState::OpenDesiredByRemote))
 				{
 					trace!(target: "sub-libp2p", "Handler({:?}, {:?}) <= Close({:?})", peer_id, *connec_id, set_id);
-					self.events.push_back(NetworkBehaviourAction::NotifyHandler {
+					self.events.push_back(ToSwarm::NotifyHandler {
 						peer_id: *peer_id,
 						handler: NotifyHandler::One(*connec_id),
 						event: NotifsHandlerIn::Close { protocol_index: set_id.into() },
@@ -548,7 +548,6 @@ impl Notifications {
 	/// Function that is called when the peerset wants us to connect to a peer.
 	fn peerset_report_connect(&mut self, peer_id: PeerId, set_id: sc_peerset::SetId) {
 		// If `PeerId` is unknown to us, insert an entry, start dialing, and return early.
-		let handler = self.new_handler();
 		let mut occ_entry = match self.peers.entry((peer_id, set_id)) {
 			Entry::Occupied(entry) => entry,
 			Entry::Vacant(entry) => {
@@ -560,10 +559,7 @@ impl Notifications {
 					set_id,
 				);
 				trace!(target: "sub-libp2p", "Libp2p <= Dial {}", entry.key().0);
-				self.events.push_back(NetworkBehaviourAction::Dial {
-					opts: entry.key().0.into(),
-					handler,
-				});
+				self.events.push_back(ToSwarm::Dial { opts: entry.key().0.into() });
 				entry.insert(PeerState::Requested);
 				return
 			},
@@ -595,10 +591,7 @@ impl Notifications {
 					set_id,
 				);
 				trace!(target: "sub-libp2p", "Libp2p <= Dial {:?}", occ_entry.key());
-				self.events.push_back(NetworkBehaviourAction::Dial {
-					opts: occ_entry.key().0.into(),
-					handler,
-				});
+				self.events.push_back(ToSwarm::Dial { opts: occ_entry.key().0.into() });
 				*occ_entry.into_mut() = PeerState::Requested;
 			},
 
@@ -646,7 +639,7 @@ impl Notifications {
 					trace!(target: "sub-libp2p", "PSM => Connect({}, {:?}): Enabling connections.",
 						occ_entry.key().0, set_id);
 					trace!(target: "sub-libp2p", "Handler({:?}, {:?}) <= Open({:?})", peer_id, *connec_id, set_id);
-					self.events.push_back(NetworkBehaviourAction::NotifyHandler {
+					self.events.push_back(ToSwarm::NotifyHandler {
 						peer_id,
 						handler: NotifyHandler::One(*connec_id),
 						event: NotifsHandlerIn::Open { protocol_index: set_id.into() },
@@ -720,7 +713,7 @@ impl Notifications {
 				{
 					trace!(target: "sub-libp2p", "Handler({:?}, {:?}) <= Open({:?})",
 						occ_entry.key(), *connec_id, set_id);
-					self.events.push_back(NetworkBehaviourAction::NotifyHandler {
+					self.events.push_back(ToSwarm::NotifyHandler {
 						peer_id: occ_entry.key().0,
 						handler: NotifyHandler::One(*connec_id),
 						event: NotifsHandlerIn::Open { protocol_index: set_id.into() },
@@ -803,7 +796,7 @@ impl Notifications {
 					trace!(target: "sub-libp2p", "External API <= Closed({}, {:?})", entry.key().0, set_id);
 					let event =
 						NotificationsOut::CustomProtocolClosed { peer_id: entry.key().0, set_id };
-					self.events.push_back(NetworkBehaviourAction::GenerateEvent(event));
+					self.events.push_back(ToSwarm::GenerateEvent(event));
 				}
 
 				for (connec_id, connec_state) in
@@ -811,7 +804,7 @@ impl Notifications {
 				{
 					trace!(target: "sub-libp2p", "Handler({:?}, {:?}) <= Close({:?})",
 						entry.key(), *connec_id, set_id);
-					self.events.push_back(NetworkBehaviourAction::NotifyHandler {
+					self.events.push_back(ToSwarm::NotifyHandler {
 						peer_id: entry.key().0,
 						handler: NotifyHandler::One(*connec_id),
 						event: NotifsHandlerIn::Close { protocol_index: set_id.into() },
@@ -824,7 +817,7 @@ impl Notifications {
 				{
 					trace!(target: "sub-libp2p", "Handler({:?}, {:?}) <= Close({:?})",
 						entry.key(), *connec_id, set_id);
-					self.events.push_back(NetworkBehaviourAction::NotifyHandler {
+					self.events.push_back(ToSwarm::NotifyHandler {
 						peer_id: entry.key().0,
 						handler: NotifyHandler::One(*connec_id),
 						event: NotifsHandlerIn::Close { protocol_index: set_id.into() },
@@ -915,7 +908,7 @@ impl Notifications {
 				{
 					trace!(target: "sub-libp2p", "Handler({:?}, {:?}) <= Open({:?})",
 						incoming.peer_id, *connec_id, incoming.set_id);
-					self.events.push_back(NetworkBehaviourAction::NotifyHandler {
+					self.events.push_back(ToSwarm::NotifyHandler {
 						peer_id: incoming.peer_id,
 						handler: NotifyHandler::One(*connec_id),
 						event: NotifsHandlerIn::Open { protocol_index: incoming.set_id.into() },
@@ -975,7 +968,7 @@ impl Notifications {
 				{
 					trace!(target: "sub-libp2p", "Handler({:?}, {:?}) <= Close({:?})",
 						incoming.peer_id, connec_id, incoming.set_id);
-					self.events.push_back(NetworkBehaviourAction::NotifyHandler {
+					self.events.push_back(ToSwarm::NotifyHandler {
 						peer_id: incoming.peer_id,
 						handler: NotifyHandler::One(*connec_id),
 						event: NotifsHandlerIn::Close { protocol_index: incoming.set_id.into() },
@@ -993,15 +986,57 @@ impl Notifications {
 }
 
 impl NetworkBehaviour for Notifications {
-	type ConnectionHandler = NotifsHandlerProto;
+	type ConnectionHandler = NotifsHandler;
 	type OutEvent = NotificationsOut;
 
-	fn new_handler(&mut self) -> Self::ConnectionHandler {
-		NotifsHandlerProto::new(self.notif_protocols.clone())
+	fn handle_pending_inbound_connection(
+		&mut self,
+		_connection_id: ConnectionId,
+		_local_addr: &Multiaddr,
+		_remote_addr: &Multiaddr,
+	) -> Result<(), ConnectionDenied> {
+		Ok(())
 	}
 
-	fn addresses_of_peer(&mut self, _: &PeerId) -> Vec<Multiaddr> {
-		Vec::new()
+	fn handle_pending_outbound_connection(
+		&mut self,
+		_connection_id: ConnectionId,
+		_maybe_peer: Option<PeerId>,
+		_addresses: &[Multiaddr],
+		_effective_role: Endpoint,
+	) -> Result<Vec<Multiaddr>, ConnectionDenied> {
+		Ok(Vec::new())
+	}
+
+	fn handle_established_inbound_connection(
+		&mut self,
+		_connection_id: ConnectionId,
+		peer: PeerId,
+		local_addr: &Multiaddr,
+		remote_addr: &Multiaddr,
+	) -> Result<THandler<Self>, ConnectionDenied> {
+		Ok(NotifsHandler::new(
+			peer,
+			ConnectedPoint::Listener {
+				local_addr: local_addr.clone(),
+				send_back_addr: remote_addr.clone(),
+			},
+			self.notif_protocols.clone(),
+		))
+	}
+
+	fn handle_established_outbound_connection(
+		&mut self,
+		_connection_id: ConnectionId,
+		peer: PeerId,
+		addr: &Multiaddr,
+		role_override: Endpoint,
+	) -> Result<THandler<Self>, ConnectionDenied> {
+		Ok(NotifsHandler::new(
+			peer,
+			ConnectedPoint::Dialer { address: addr.clone(), role_override },
+			self.notif_protocols.clone(),
+		))
 	}
 
 	fn on_swarm_event(&mut self, event: FromSwarm<Self::ConnectionHandler>) {
@@ -1022,7 +1057,7 @@ impl NetworkBehaviour for Notifications {
 								peer_id, set_id, endpoint
 							);
 							trace!(target: "sub-libp2p", "Handler({:?}, {:?}) <= Open({:?})", peer_id, connection_id, set_id);
-							self.events.push_back(NetworkBehaviourAction::NotifyHandler {
+							self.events.push_back(ToSwarm::NotifyHandler {
 								peer_id,
 								handler: NotifyHandler::One(connection_id),
 								event: NotifsHandlerIn::Open { protocol_index: set_id.into() },
@@ -1276,9 +1311,7 @@ impl NetworkBehaviour for Notifications {
 												set_id,
 												notifications_sink: replacement_sink,
 											};
-											self.events.push_back(
-												NetworkBehaviourAction::GenerateEvent(event),
-											);
+											self.events.push_back(ToSwarm::GenerateEvent(event));
 										}
 									} else {
 										trace!(
@@ -1289,9 +1322,7 @@ impl NetworkBehaviour for Notifications {
 											peer_id,
 											set_id,
 										};
-										self.events.push_back(
-											NetworkBehaviourAction::GenerateEvent(event),
-										);
+										self.events.push_back(ToSwarm::GenerateEvent(event));
 									}
 								}
 							} else {
@@ -1434,8 +1465,7 @@ impl NetworkBehaviour for Notifications {
 		&mut self,
 		peer_id: PeerId,
 		connection_id: ConnectionId,
-		event: <<Self::ConnectionHandler as IntoConnectionHandler>::Handler as
-		ConnectionHandler>::OutEvent,
+		event: THandlerOutEvent<Self>,
 	) {
 		match event {
 			NotifsHandlerOut::OpenDesiredByRemote { protocol_index } => {
@@ -1502,7 +1532,7 @@ impl NetworkBehaviour for Notifications {
 							if let ConnectionState::Closed = *connec_state {
 								trace!(target: "sub-libp2p", "Handler({:?}, {:?}) <= Open({:?})",
 									peer_id, connection_id, set_id);
-								self.events.push_back(NetworkBehaviourAction::NotifyHandler {
+								self.events.push_back(ToSwarm::NotifyHandler {
 									peer_id,
 									handler: NotifyHandler::One(connection_id),
 									event: NotifsHandlerIn::Open { protocol_index: set_id.into() },
@@ -1583,7 +1613,7 @@ impl NetworkBehaviour for Notifications {
 							if let ConnectionState::Closed = *connec_state {
 								trace!(target: "sub-libp2p", "Handler({:?}, {:?}) <= Open({:?})",
 									peer_id, connection_id, set_id);
-								self.events.push_back(NetworkBehaviourAction::NotifyHandler {
+								self.events.push_back(ToSwarm::NotifyHandler {
 									peer_id,
 									handler: NotifyHandler::One(connection_id),
 									event: NotifsHandlerIn::Open { protocol_index: set_id.into() },
@@ -1668,7 +1698,7 @@ impl NetworkBehaviour for Notifications {
 						connections[pos].1 = ConnectionState::Closing;
 
 						trace!(target: "sub-libp2p", "Handler({}, {:?}) <= Close({:?})", peer_id, connection_id, set_id);
-						self.events.push_back(NetworkBehaviourAction::NotifyHandler {
+						self.events.push_back(ToSwarm::NotifyHandler {
 							peer_id,
 							handler: NotifyHandler::One(connection_id),
 							event: NotifsHandlerIn::Close { protocol_index: set_id.into() },
@@ -1686,7 +1716,7 @@ impl NetworkBehaviour for Notifications {
 									set_id,
 									notifications_sink: replacement_sink,
 								};
-								self.events.push_back(NetworkBehaviourAction::GenerateEvent(event));
+								self.events.push_back(ToSwarm::GenerateEvent(event));
 							}
 
 							*entry.into_mut() = PeerState::Enabled { connections };
@@ -1706,7 +1736,7 @@ impl NetworkBehaviour for Notifications {
 
 							trace!(target: "sub-libp2p", "External API <= Closed({}, {:?})", peer_id, set_id);
 							let event = NotificationsOut::CustomProtocolClosed { peer_id, set_id };
-							self.events.push_back(NetworkBehaviourAction::GenerateEvent(event));
+							self.events.push_back(ToSwarm::GenerateEvent(event));
 						}
 					},
 
@@ -1790,7 +1820,7 @@ impl NetworkBehaviour for Notifications {
 									received_handshake,
 									notifications_sink: notifications_sink.clone(),
 								};
-								self.events.push_back(NetworkBehaviourAction::GenerateEvent(event));
+								self.events.push_back(ToSwarm::GenerateEvent(event));
 							}
 							*connec_state = ConnectionState::Open(notifications_sink);
 						} else if let Some((_, connec_state)) =
@@ -1937,7 +1967,7 @@ impl NetworkBehaviour for Notifications {
 					);
 					let event = NotificationsOut::Notification { peer_id, set_id, message };
 
-					self.events.push_back(NetworkBehaviourAction::GenerateEvent(event));
+					self.events.push_back(ToSwarm::GenerateEvent(event));
 				} else {
 					trace!(
 						target: "sub-libp2p",
@@ -1956,7 +1986,7 @@ impl NetworkBehaviour for Notifications {
 		&mut self,
 		cx: &mut Context,
 		_params: &mut impl PollParameters,
-	) -> Poll<NetworkBehaviourAction<Self::OutEvent, Self::ConnectionHandler>> {
+	) -> Poll<ToSwarm<Self::OutEvent, THandlerInEvent<Self>>> {
 		if let Some(event) = self.events.pop_front() {
 			return Poll::Ready(event)
 		}
@@ -1988,8 +2018,6 @@ impl NetworkBehaviour for Notifications {
 		while let Poll::Ready(Some((delay_id, peer_id, set_id))) =
 			Pin::new(&mut self.delays).poll_next(cx)
 		{
-			let handler = self.new_handler();
-
 			let peer_state = match self.peers.get_mut(&(peer_id, set_id)) {
 				Some(s) => s,
 				// We intentionally never remove elements from `delays`, and it may
@@ -2005,8 +2033,7 @@ impl NetworkBehaviour for Notifications {
 
 				PeerState::PendingRequest { timer, .. } if *timer == delay_id => {
 					trace!(target: "sub-libp2p", "Libp2p <= Dial {:?} now that ban has expired", peer_id);
-					self.events
-						.push_back(NetworkBehaviourAction::Dial { opts: peer_id.into(), handler });
+					self.events.push_back(ToSwarm::Dial { opts: peer_id.into() });
 					*peer_state = PeerState::Requested;
 				},
 
@@ -2019,7 +2046,7 @@ impl NetworkBehaviour for Notifications {
 					{
 						trace!(target: "sub-libp2p", "Handler({}, {:?}) <= Open({:?}) (ban expired)",
 							peer_id, *connec_id, set_id);
-						self.events.push_back(NetworkBehaviourAction::NotifyHandler {
+						self.events.push_back(ToSwarm::NotifyHandler {
 							peer_id,
 							handler: NotifyHandler::One(*connec_id),
 							event: NotifsHandlerIn::Open { protocol_index: set_id.into() },
@@ -2055,13 +2082,11 @@ impl NetworkBehaviour for Notifications {
 }
 
 #[cfg(test)]
+#[allow(deprecated)]
 mod tests {
 	use super::*;
 	use crate::protocol::notifications::handler::tests::*;
-	use libp2p::{
-		core::ConnectedPoint,
-		swarm::{behaviour::FromSwarm, AddressRecord},
-	};
+	use libp2p::swarm::AddressRecord;
 	use std::{collections::HashSet, iter};
 
 	impl PartialEq for ConnectionState {
@@ -2223,7 +2248,7 @@ mod tests {
 	fn remote_opens_connection_and_substream() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -2273,7 +2298,7 @@ mod tests {
 	async fn disconnect_remote_substream_before_handled_by_peerset() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -2310,7 +2335,7 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let set_id = sc_peerset::SetId::from(0);
 		let peer = PeerId::random();
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -2344,8 +2369,8 @@ mod tests {
 			libp2p::swarm::behaviour::ConnectionClosed {
 				peer_id: peer,
 				connection_id: conn,
-				endpoint: &connected,
-				handler: NotifsHandlerProto::new(vec![]).into_handler(&peer, &connected),
+				endpoint: &connected.clone(),
+				handler: NotifsHandler::new(peer, connected, vec![]),
 				remaining_established: 0usize,
 			},
 		));
@@ -2374,7 +2399,7 @@ mod tests {
 	fn peerset_connect_incoming() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let set_id = sc_peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
@@ -2409,7 +2434,7 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let set_id = sc_peerset::SetId::from(0);
 		let peer = PeerId::random();
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -2455,7 +2480,7 @@ mod tests {
 	fn peerset_disconnect_enabled() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let set_id = sc_peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
@@ -2505,7 +2530,7 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let set_id = sc_peerset::SetId::from(0);
 		let peer = PeerId::random();
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -2534,8 +2559,8 @@ mod tests {
 			libp2p::swarm::behaviour::ConnectionClosed {
 				peer_id: peer,
 				connection_id: conn,
-				endpoint: &connected,
-				handler: NotifsHandlerProto::new(vec![]).into_handler(&peer, &connected),
+				endpoint: &connected.clone(),
+				handler: NotifsHandler::new(peer, connected, vec![]),
 				remaining_established: 0usize,
 			},
 		));
@@ -2557,7 +2582,7 @@ mod tests {
 	fn peerset_accept_peer_not_alive() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let set_id = sc_peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
@@ -2604,8 +2629,8 @@ mod tests {
 	fn secondary_connection_peer_state_incoming() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let conn = ConnectionId::new(0usize);
-		let conn2 = ConnectionId::new(1usize);
+		let conn = ConnectionId::new_unchecked(0);
+		let conn2 = ConnectionId::new_unchecked(1);
 		let set_id = sc_peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
@@ -2659,7 +2684,7 @@ mod tests {
 	fn close_connection_for_disabled_peer() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let set_id = sc_peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
@@ -2681,8 +2706,8 @@ mod tests {
 			libp2p::swarm::behaviour::ConnectionClosed {
 				peer_id: peer,
 				connection_id: conn,
-				endpoint: &connected,
-				handler: NotifsHandlerProto::new(vec![]).into_handler(&peer, &connected),
+				endpoint: &connected.clone(),
+				handler: NotifsHandler::new(peer, connected, vec![]),
 				remaining_established: 0usize,
 			},
 		));
@@ -2693,7 +2718,7 @@ mod tests {
 	fn close_connection_for_incoming_peer_one_connection() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let set_id = sc_peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
@@ -2722,8 +2747,8 @@ mod tests {
 			libp2p::swarm::behaviour::ConnectionClosed {
 				peer_id: peer,
 				connection_id: conn,
-				endpoint: &connected,
-				handler: NotifsHandlerProto::new(vec![]).into_handler(&peer, &connected),
+				endpoint: &connected.clone(),
+				handler: NotifsHandler::new(peer, connected, vec![]),
 				remaining_established: 0usize,
 			},
 		));
@@ -2738,8 +2763,8 @@ mod tests {
 	fn close_connection_for_incoming_peer_two_connections() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let conn = ConnectionId::new(0usize);
-		let conn1 = ConnectionId::new(1usize);
+		let conn = ConnectionId::new_unchecked(0);
+		let conn1 = ConnectionId::new_unchecked(1);
 		let set_id = sc_peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
@@ -2789,8 +2814,8 @@ mod tests {
 			libp2p::swarm::behaviour::ConnectionClosed {
 				peer_id: peer,
 				connection_id: conn,
-				endpoint: &connected,
-				handler: NotifsHandlerProto::new(vec![]).into_handler(&peer, &connected),
+				endpoint: &connected.clone(),
+				handler: NotifsHandler::new(peer, connected, vec![]),
 				remaining_established: 0usize,
 			},
 		));
@@ -2807,7 +2832,7 @@ mod tests {
 	fn connection_and_substream_open() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let set_id = sc_peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
@@ -2851,7 +2876,7 @@ mod tests {
 
 		assert!(std::matches!(
 			notif.events[notif.events.len() - 1],
-			NetworkBehaviourAction::GenerateEvent(NotificationsOut::CustomProtocolOpen { .. })
+			ToSwarm::GenerateEvent(NotificationsOut::CustomProtocolOpen { .. })
 		));
 	}
 
@@ -2859,8 +2884,8 @@ mod tests {
 	fn connection_closed_sink_replaced() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let conn1 = ConnectionId::new(0usize);
-		let conn2 = ConnectionId::new(1usize);
+		let conn1 = ConnectionId::new_unchecked(0);
+		let conn2 = ConnectionId::new_unchecked(1);
 		let set_id = sc_peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
@@ -2931,8 +2956,8 @@ mod tests {
 			libp2p::swarm::behaviour::ConnectionClosed {
 				peer_id: peer,
 				connection_id: conn1,
-				endpoint: &connected,
-				handler: NotifsHandlerProto::new(vec![]).into_handler(&peer, &connected),
+				endpoint: &connected.clone(),
+				handler: NotifsHandler::new(peer, connected, vec![]),
 				remaining_established: 0usize,
 			},
 		));
@@ -2947,7 +2972,7 @@ mod tests {
 
 		assert!(std::matches!(
 			notif.events[notif.events.len() - 1],
-			NetworkBehaviourAction::GenerateEvent(NotificationsOut::CustomProtocolReplaced { .. })
+			ToSwarm::GenerateEvent(NotificationsOut::CustomProtocolReplaced { .. })
 		));
 	}
 
@@ -2963,8 +2988,8 @@ mod tests {
 
 		notif.on_swarm_event(FromSwarm::DialFailure(libp2p::swarm::behaviour::DialFailure {
 			peer_id: Some(peer),
-			handler: NotifsHandlerProto::new(vec![]),
 			error: &libp2p::swarm::DialError::Banned,
+			connection_id: ConnectionId::new_unchecked(1337),
 		}));
 
 		if let Some(PeerState::Backoff { timer_deadline, .. }) = notif.peers.get(&(peer, set_id)) {
@@ -2978,7 +3003,7 @@ mod tests {
 	async fn write_notification() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let set_id = sc_peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
@@ -3028,7 +3053,7 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let set_id = sc_peerset::SetId::from(0);
 		let peer = PeerId::random();
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -3057,8 +3082,8 @@ mod tests {
 			libp2p::swarm::behaviour::ConnectionClosed {
 				peer_id: peer,
 				connection_id: conn,
-				endpoint: &connected,
-				handler: NotifsHandlerProto::new(vec![]).into_handler(&peer, &connected),
+				endpoint: &connected.clone(),
+				handler: NotifsHandler::new(peer, connected, vec![]),
 				remaining_established: 0usize,
 			},
 		));
@@ -3076,7 +3101,7 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
 		let set_id = sc_peerset::SetId::from(0);
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -3102,7 +3127,7 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let set_id = sc_peerset::SetId::from(0);
 		let peer = PeerId::random();
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -3131,8 +3156,8 @@ mod tests {
 			libp2p::swarm::behaviour::ConnectionClosed {
 				peer_id: peer,
 				connection_id: conn,
-				endpoint: &connected,
-				handler: NotifsHandlerProto::new(vec![]).into_handler(&peer, &connected),
+				endpoint: &connected.clone(),
+				handler: NotifsHandler::new(peer, connected, vec![]),
 				remaining_established: 0usize,
 			},
 		));
@@ -3148,8 +3173,8 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let set_id = sc_peerset::SetId::from(0);
 		let peer = PeerId::random();
-		let conn1 = ConnectionId::new(0usize);
-		let conn2 = ConnectionId::new(0usize);
+		let conn1 = ConnectionId::new_unchecked(0);
+		let conn2 = ConnectionId::new_unchecked(1);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -3194,8 +3219,8 @@ mod tests {
 			libp2p::swarm::behaviour::ConnectionClosed {
 				peer_id: peer,
 				connection_id: conn1,
-				endpoint: &connected,
-				handler: NotifsHandlerProto::new(vec![]).into_handler(&peer, &connected),
+				endpoint: &connected.clone(),
+				handler: NotifsHandler::new(peer, connected.clone(), vec![]),
 				remaining_established: 0usize,
 			},
 		));
@@ -3208,8 +3233,8 @@ mod tests {
 			libp2p::swarm::behaviour::ConnectionClosed {
 				peer_id: peer,
 				connection_id: conn2,
-				endpoint: &connected,
-				handler: NotifsHandlerProto::new(vec![]).into_handler(&peer, &connected),
+				endpoint: &connected.clone(),
+				handler: NotifsHandler::new(peer, connected, vec![]),
 				remaining_established: 0usize,
 			},
 		));
@@ -3221,7 +3246,7 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
 		let set_id = sc_peerset::SetId::from(0);
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -3259,8 +3284,8 @@ mod tests {
 			libp2p::swarm::behaviour::ConnectionClosed {
 				peer_id: peer,
 				connection_id: conn,
-				endpoint: &connected,
-				handler: NotifsHandlerProto::new(vec![]).into_handler(&peer, &connected),
+				endpoint: &connected.clone(),
+				handler: NotifsHandler::new(peer, connected, vec![]),
 				remaining_established: 0usize,
 			},
 		));
@@ -3271,8 +3296,8 @@ mod tests {
 	fn two_connections_inactive_connection_gets_closed_peer_state_is_still_incoming() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let conn1 = ConnectionId::new(0usize);
-		let conn2 = ConnectionId::new(1usize);
+		let conn1 = ConnectionId::new_unchecked(0);
+		let conn2 = ConnectionId::new_unchecked(1);
 		let set_id = sc_peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
@@ -3314,8 +3339,8 @@ mod tests {
 			libp2p::swarm::behaviour::ConnectionClosed {
 				peer_id: peer,
 				connection_id: conn2,
-				endpoint: &connected,
-				handler: NotifsHandlerProto::new(vec![]).into_handler(&peer, &connected),
+				endpoint: &connected.clone(),
+				handler: NotifsHandler::new(peer, connected, vec![]),
 				remaining_established: 0usize,
 			},
 		));
@@ -3326,8 +3351,8 @@ mod tests {
 	fn two_connections_active_connection_gets_closed_peer_state_is_disabled() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let conn1 = ConnectionId::new(0usize);
-		let conn2 = ConnectionId::new(1usize);
+		let conn1 = ConnectionId::new_unchecked(0);
+		let conn2 = ConnectionId::new_unchecked(1);
 		let set_id = sc_peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
@@ -3372,8 +3397,8 @@ mod tests {
 			libp2p::swarm::behaviour::ConnectionClosed {
 				peer_id: peer,
 				connection_id: conn1,
-				endpoint: &connected,
-				handler: NotifsHandlerProto::new(vec![]).into_handler(&peer, &connected),
+				endpoint: &connected.clone(),
+				handler: NotifsHandler::new(peer, connected, vec![]),
 				remaining_established: 0usize,
 			},
 		));
@@ -3384,8 +3409,8 @@ mod tests {
 	fn inject_connection_closed_for_active_connection() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let conn1 = ConnectionId::new(0usize);
-		let conn2 = ConnectionId::new(1usize);
+		let conn1 = ConnectionId::new_unchecked(0);
+		let conn2 = ConnectionId::new_unchecked(1);
 		let set_id = sc_peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
@@ -3441,8 +3466,8 @@ mod tests {
 			libp2p::swarm::behaviour::ConnectionClosed {
 				peer_id: peer,
 				connection_id: conn1,
-				endpoint: &connected,
-				handler: NotifsHandlerProto::new(vec![]).into_handler(&peer, &connected),
+				endpoint: &connected.clone(),
+				handler: NotifsHandler::new(peer, connected, vec![]),
 				remaining_established: 0usize,
 			},
 		));
@@ -3453,7 +3478,7 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let set_id = sc_peerset::SetId::from(0);
 		let peer = PeerId::random();
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -3482,8 +3507,8 @@ mod tests {
 			libp2p::swarm::behaviour::ConnectionClosed {
 				peer_id: peer,
 				connection_id: conn,
-				endpoint: &connected,
-				handler: NotifsHandlerProto::new(vec![]).into_handler(&peer, &connected),
+				endpoint: &connected.clone(),
+				handler: NotifsHandler::new(peer, connected, vec![]),
 				remaining_established: 0usize,
 			},
 		));
@@ -3500,8 +3525,8 @@ mod tests {
 		let now = Instant::now();
 		notif.on_swarm_event(FromSwarm::DialFailure(libp2p::swarm::behaviour::DialFailure {
 			peer_id: Some(peer),
-			handler: NotifsHandlerProto::new(vec![]),
 			error: &libp2p::swarm::DialError::Banned,
+			connection_id: ConnectionId::new_unchecked(0),
 		}));
 
 		if let Some(PeerState::PendingRequest { ref timer_deadline, .. }) =
@@ -3516,8 +3541,8 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
 		let set_id = sc_peerset::SetId::from(0);
-		let conn1 = ConnectionId::new(0usize);
-		let conn2 = ConnectionId::new(1usize);
+		let conn1 = ConnectionId::new_unchecked(0);
+		let conn2 = ConnectionId::new_unchecked(1);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -3570,7 +3595,7 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
 		let set_id = sc_peerset::SetId::from(0);
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -3599,8 +3624,8 @@ mod tests {
 			libp2p::swarm::behaviour::ConnectionClosed {
 				peer_id: peer,
 				connection_id: conn,
-				endpoint: &connected,
-				handler: NotifsHandlerProto::new(vec![]).into_handler(&peer, &connected),
+				endpoint: &connected.clone(),
+				handler: NotifsHandler::new(peer, connected, vec![]),
 				remaining_established: 0usize,
 			},
 		));
@@ -3647,7 +3672,7 @@ mod tests {
 	async fn reschedule_disabled_pending_enable_when_connection_not_closed() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let set_id = sc_peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
@@ -3763,7 +3788,7 @@ mod tests {
 	fn peerset_report_connect_with_enabled_peer() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let set_id = sc_peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
@@ -3815,7 +3840,7 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let set_id = sc_peerset::SetId::from(0);
 		let peer = PeerId::random();
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -3872,7 +3897,7 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let set_id = sc_peerset::SetId::from(0);
 		let peer = PeerId::random();
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -3901,8 +3926,8 @@ mod tests {
 			libp2p::swarm::behaviour::ConnectionClosed {
 				peer_id: peer,
 				connection_id: conn,
-				endpoint: &connected,
-				handler: NotifsHandlerProto::new(vec![]).into_handler(&peer, &connected),
+				endpoint: &connected.clone(),
+				handler: NotifsHandler::new(peer, connected, vec![]),
 				remaining_established: 0usize,
 			},
 		));
@@ -3925,7 +3950,7 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
 		let set_id = sc_peerset::SetId::from(0);
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -3960,7 +3985,7 @@ mod tests {
 	fn peerset_report_accept_incoming_peer() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let set_id = sc_peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
@@ -4001,7 +4026,7 @@ mod tests {
 	fn peerset_report_accept_not_incoming_peer() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let set_id = sc_peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
@@ -4058,9 +4083,9 @@ mod tests {
 		notif.on_swarm_event(FromSwarm::ConnectionClosed(
 			libp2p::swarm::behaviour::ConnectionClosed {
 				peer_id: peer,
-				connection_id: ConnectionId::new(0usize),
-				endpoint: &endpoint,
-				handler: NotifsHandlerProto::new(vec![]).into_handler(&peer, &endpoint),
+				connection_id: ConnectionId::new_unchecked(0),
+				endpoint: &endpoint.clone(),
+				handler: NotifsHandler::new(peer, endpoint, vec![]),
 				remaining_established: 0usize,
 			},
 		));
@@ -4102,7 +4127,7 @@ mod tests {
 	fn reject_non_active_connection() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let set_id = sc_peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
@@ -4140,7 +4165,7 @@ mod tests {
 	fn reject_non_existent_peer_but_alive_connection() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let set_id = sc_peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
@@ -4180,7 +4205,7 @@ mod tests {
 	fn inject_non_existent_connection_closed_for_incoming_peer() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let set_id = sc_peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
@@ -4209,9 +4234,9 @@ mod tests {
 		notif.on_swarm_event(FromSwarm::ConnectionClosed(
 			libp2p::swarm::behaviour::ConnectionClosed {
 				peer_id: peer,
-				connection_id: ConnectionId::new(1337usize),
-				endpoint: &connected,
-				handler: NotifsHandlerProto::new(vec![]).into_handler(&peer, &connected),
+				connection_id: ConnectionId::new_unchecked(1337),
+				endpoint: &connected.clone(),
+				handler: NotifsHandler::new(peer, connected, vec![]),
 				remaining_established: 0usize,
 			},
 		));
@@ -4224,7 +4249,7 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let set_id = sc_peerset::SetId::from(0);
 		let peer = PeerId::random();
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -4244,9 +4269,9 @@ mod tests {
 		notif.on_swarm_event(FromSwarm::ConnectionClosed(
 			libp2p::swarm::behaviour::ConnectionClosed {
 				peer_id: peer,
-				connection_id: ConnectionId::new(1337usize),
-				endpoint: &connected,
-				handler: NotifsHandlerProto::new(vec![]).into_handler(&peer, &connected),
+				connection_id: ConnectionId::new_unchecked(1337),
+				endpoint: &connected.clone(),
+				handler: NotifsHandler::new(peer, connected, vec![]),
 				remaining_established: 0usize,
 			},
 		));
@@ -4259,7 +4284,7 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let set_id = sc_peerset::SetId::from(0);
 		let peer = PeerId::random();
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -4295,9 +4320,9 @@ mod tests {
 		notif.on_swarm_event(FromSwarm::ConnectionClosed(
 			libp2p::swarm::behaviour::ConnectionClosed {
 				peer_id: peer,
-				connection_id: ConnectionId::new(1337usize),
-				endpoint: &connected,
-				handler: NotifsHandlerProto::new(vec![]).into_handler(&peer, &connected),
+				connection_id: ConnectionId::new_unchecked(1337),
+				endpoint: &connected.clone(),
+				handler: NotifsHandler::new(peer, connected, vec![]),
 				remaining_established: 0usize,
 			},
 		));
@@ -4309,7 +4334,7 @@ mod tests {
 	fn inject_connection_closed_for_incoming_peer_state_mismatch() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let set_id = sc_peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
@@ -4340,8 +4365,8 @@ mod tests {
 			libp2p::swarm::behaviour::ConnectionClosed {
 				peer_id: peer,
 				connection_id: conn,
-				endpoint: &connected,
-				handler: NotifsHandlerProto::new(vec![]).into_handler(&peer, &connected),
+				endpoint: &connected.clone(),
+				handler: NotifsHandler::new(peer, connected, vec![]),
 				remaining_established: 0usize,
 			},
 		));
@@ -4353,7 +4378,7 @@ mod tests {
 	fn inject_connection_closed_for_enabled_state_mismatch() {
 		let (mut notif, _peerset) = development_notifs();
 		let peer = PeerId::random();
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let set_id = sc_peerset::SetId::from(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
@@ -4386,9 +4411,9 @@ mod tests {
 		notif.on_swarm_event(FromSwarm::ConnectionClosed(
 			libp2p::swarm::behaviour::ConnectionClosed {
 				peer_id: peer,
-				connection_id: ConnectionId::new(1337usize),
-				endpoint: &connected,
-				handler: NotifsHandlerProto::new(vec![]).into_handler(&peer, &connected),
+				connection_id: ConnectionId::new_unchecked(1337),
+				endpoint: &connected.clone(),
+				handler: NotifsHandler::new(peer, connected, vec![]),
 				remaining_established: 0usize,
 			},
 		));
@@ -4401,7 +4426,7 @@ mod tests {
 		let (mut notif, _peerset) = development_notifs();
 		let set_id = sc_peerset::SetId::from(0);
 		let peer = PeerId::random();
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),
@@ -4430,8 +4455,8 @@ mod tests {
 			libp2p::swarm::behaviour::ConnectionClosed {
 				peer_id: peer,
 				connection_id: conn,
-				endpoint: &connected,
-				handler: NotifsHandlerProto::new(vec![]).into_handler(&peer, &connected),
+				endpoint: &connected.clone(),
+				handler: NotifsHandler::new(peer, connected.clone(), vec![]),
 				remaining_established: 0usize,
 			},
 		));
@@ -4441,8 +4466,8 @@ mod tests {
 			libp2p::swarm::behaviour::ConnectionClosed {
 				peer_id: peer,
 				connection_id: conn,
-				endpoint: &connected,
-				handler: NotifsHandlerProto::new(vec![]).into_handler(&peer, &connected),
+				endpoint: &connected.clone(),
+				handler: NotifsHandler::new(peer, connected, vec![]),
 				remaining_established: 0usize,
 			},
 		));
@@ -4453,7 +4478,7 @@ mod tests {
 	#[cfg(debug_assertions)]
 	fn open_result_ok_non_existent_peer() {
 		let (mut notif, _peerset) = development_notifs();
-		let conn = ConnectionId::new(0usize);
+		let conn = ConnectionId::new_unchecked(0);
 		let connected = ConnectedPoint::Listener {
 			local_addr: Multiaddr::empty(),
 			send_back_addr: Multiaddr::empty(),

substrate/client/network/src/protocol/notifications/handler.rs

@@ -72,11 +72,12 @@ use futures::{
 	prelude::*,
 };
 use libp2p::{
-	core::{ConnectedPoint, PeerId},
+	core::ConnectedPoint,
 	swarm::{
-		handler::ConnectionEvent, ConnectionHandler, ConnectionHandlerEvent, IntoConnectionHandler,
-		KeepAlive, NegotiatedSubstream, SubstreamProtocol,
+		handler::ConnectionEvent, ConnectionHandler, ConnectionHandlerEvent, KeepAlive,
+		NegotiatedSubstream, SubstreamProtocol,
 	},
+	PeerId,
 };
 use log::error;
 use parking_lot::{Mutex, RwLock};
@@ -105,19 +106,6 @@ const OPEN_TIMEOUT: Duration = Duration::from_secs(10);
 /// open substreams.
 const INITIAL_KEEPALIVE_TIME: Duration = Duration::from_secs(5);
 
-/// Implements the `IntoConnectionHandler` trait of libp2p.
-///
-/// Every time a connection with a remote starts, an instance of this struct is created and
-/// sent to a background task dedicated to this connection. Once the connection is established,
-/// it is turned into a [`NotifsHandler`].
-///
-/// See the documentation at the module level for more information.
-pub struct NotifsHandlerProto {
-	/// Name of protocols, prototypes for upgrades for inbound substreams, and the message we
-	/// send or respond with in the handshake.
-	protocols: Vec<ProtocolConfig>,
-}
-
 /// The actual handler once the connection has been established.
 ///
 /// See the documentation at the module level for more information.
@@ -140,6 +128,30 @@ pub struct NotifsHandler {
 	>,
 }
 
+impl NotifsHandler {
+	/// Creates new [`NotifsHandler`].
+	pub fn new(peer_id: PeerId, endpoint: ConnectedPoint, protocols: Vec<ProtocolConfig>) -> Self {
+		Self {
+			protocols: protocols
+				.into_iter()
+				.map(|config| {
+					let in_upgrade = NotificationsIn::new(
+						config.name.clone(),
+						config.fallback_names.clone(),
+						config.max_notification_size,
+					);
+
+					Protocol { config, in_upgrade, state: State::Closed { pending_opening: false } }
+				})
+				.collect(),
+			peer_id,
+			endpoint,
+			when_connection_open: Instant::now(),
+			events_queue: VecDeque::with_capacity(16),
+		}
+	}
+}
+
 /// Configuration for a notifications protocol.
 #[derive(Debug, Clone)]
 pub struct ProtocolConfig {
@@ -223,45 +235,6 @@ enum State {
 	},
 }
 
-impl IntoConnectionHandler for NotifsHandlerProto {
-	type Handler = NotifsHandler;
-
-	fn inbound_protocol(&self) -> UpgradeCollec<NotificationsIn> {
-		self.protocols
-			.iter()
-			.map(|cfg| {
-				NotificationsIn::new(
-					cfg.name.clone(),
-					cfg.fallback_names.clone(),
-					cfg.max_notification_size,
-				)
-			})
-			.collect::<UpgradeCollec<_>>()
-	}
-
-	fn into_handler(self, peer_id: &PeerId, connected_point: &ConnectedPoint) -> Self::Handler {
-		NotifsHandler {
-			protocols: self
-				.protocols
-				.into_iter()
-				.map(|config| {
-					let in_upgrade = NotificationsIn::new(
-						config.name.clone(),
-						config.fallback_names.clone(),
-						config.max_notification_size,
-					);
-
-					Protocol { config, in_upgrade, state: State::Closed { pending_opening: false } }
-				})
-				.collect(),
-			peer_id: *peer_id,
-			endpoint: connected_point.clone(),
-			when_connection_open: Instant::now(),
-			events_queue: VecDeque::with_capacity(16),
-		}
-	}
-}
-
 /// Event that can be received by a `NotifsHandler`.
 #[derive(Debug, Clone)]
 pub enum NotifsHandlerIn {
@@ -461,18 +434,6 @@ pub enum NotifsHandlerError {
 	SyncNotificationsClogged,
 }
 
-impl NotifsHandlerProto {
-	/// Builds a new handler.
-	///
-	/// `list` is a list of notification protocols names, the message to send as part of the
-	/// handshake, and the maximum allowed size of a notification. At the moment, the message
-	/// is always the same whether we open a substream ourselves or respond to handshake from
-	/// the remote.
-	pub fn new(list: impl Into<Vec<ProtocolConfig>>) -> Self {
-		Self { protocols: list.into() }
-	}
-}
-
 impl ConnectionHandler for NotifsHandler {
 	type InEvent = NotifsHandlerIn;
 	type OutEvent = NotifsHandlerOut;
@@ -954,7 +915,6 @@ pub mod tests {
 			.await
 		}
 	}
-
 	struct MockSubstream {
 		pub rx: mpsc::Receiver<Vec<u8>>,
 		pub tx: mpsc::Sender<Vec<u8>>,

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

@@ -22,11 +22,12 @@ use crate::protocol::notifications::{Notifications, NotificationsOut, ProtocolCo
 
 use futures::prelude::*;
 use libp2p::{
-	core::{connection::ConnectionId, transport::MemoryTransport, upgrade},
+	core::{transport::MemoryTransport, upgrade, Endpoint},
 	identity, noise,
 	swarm::{
-		behaviour::FromSwarm, ConnectionHandler, Executor, IntoConnectionHandler, NetworkBehaviour,
-		NetworkBehaviourAction, PollParameters, Swarm, SwarmEvent,
+		behaviour::FromSwarm, ConnectionDenied, ConnectionId, Executor, NetworkBehaviour,
+		PollParameters, Swarm, SwarmBuilder, SwarmEvent, THandler, THandlerInEvent,
+		THandlerOutEvent, ToSwarm,
 	},
 	yamux, Multiaddr, PeerId, Transport,
 };
@@ -57,13 +58,10 @@ fn build_nodes() -> (Swarm<CustomProtoWithAddr>, Swarm<CustomProtoWithAddr>) {
 	for index in 0..2 {
 		let keypair = keypairs[index].clone();
 
-		let noise_keys =
-			noise::Keypair::<noise::X25519Spec>::new().into_authentic(&keypair).unwrap();
-
 		let transport = MemoryTransport::new()
 			.upgrade(upgrade::Version::V1)
-			.authenticate(noise::NoiseConfig::xx(noise_keys).into_authenticated())
-			.multiplex(yamux::YamuxConfig::default())
+			.authenticate(noise::Config::new(&keypair).unwrap())
+			.multiplex(yamux::Config::default())
 			.timeout(Duration::from_secs(20))
 			.boxed();
 
@@ -105,12 +103,13 @@ fn build_nodes() -> (Swarm<CustomProtoWithAddr>, Swarm<CustomProtoWithAddr>) {
 		};
 
 		let runtime = tokio::runtime::Runtime::new().unwrap();
-		let mut swarm = Swarm::with_executor(
+		let mut swarm = SwarmBuilder::with_executor(
 			transport,
 			behaviour,
 			keypairs[index].public().to_peer_id(),
 			TokioExecutor(runtime),
-		);
+		)
+		.build();
 		swarm.listen_on(addrs[index].clone()).unwrap();
 		out.push(swarm);
 	}
@@ -146,18 +145,63 @@ impl NetworkBehaviour for CustomProtoWithAddr {
 	type ConnectionHandler = <Notifications as NetworkBehaviour>::ConnectionHandler;
 	type OutEvent = <Notifications as NetworkBehaviour>::OutEvent;
 
-	fn new_handler(&mut self) -> Self::ConnectionHandler {
-		self.inner.new_handler()
+	fn handle_pending_inbound_connection(
+		&mut self,
+		connection_id: ConnectionId,
+		local_addr: &Multiaddr,
+		remote_addr: &Multiaddr,
+	) -> Result<(), ConnectionDenied> {
+		self.inner
+			.handle_pending_inbound_connection(connection_id, local_addr, remote_addr)
 	}
 
-	fn addresses_of_peer(&mut self, peer_id: &PeerId) -> Vec<Multiaddr> {
-		let mut list = self.inner.addresses_of_peer(peer_id);
-		for (p, a) in self.addrs.iter() {
-			if p == peer_id {
-				list.push(a.clone());
+	fn handle_pending_outbound_connection(
+		&mut self,
+		connection_id: ConnectionId,
+		maybe_peer: Option<PeerId>,
+		addresses: &[Multiaddr],
+		effective_role: Endpoint,
+	) -> Result<Vec<Multiaddr>, ConnectionDenied> {
+		let mut list = self.inner.handle_pending_outbound_connection(
+			connection_id,
+			maybe_peer,
+			addresses,
+			effective_role,
+		)?;
+		if let Some(peer_id) = maybe_peer {
+			for (p, a) in self.addrs.iter() {
+				if *p == peer_id {
+					list.push(a.clone());
+				}
 			}
 		}
-		list
+		Ok(list)
+	}
+
+	fn handle_established_inbound_connection(
+		&mut self,
+		connection_id: ConnectionId,
+		peer: PeerId,
+		local_addr: &Multiaddr,
+		remote_addr: &Multiaddr,
+	) -> Result<THandler<Self>, ConnectionDenied> {
+		self.inner.handle_established_inbound_connection(
+			connection_id,
+			peer,
+			local_addr,
+			remote_addr,
+		)
+	}
+
+	fn handle_established_outbound_connection(
+		&mut self,
+		connection_id: ConnectionId,
+		peer: PeerId,
+		addr: &Multiaddr,
+		role_override: Endpoint,
+	) -> Result<THandler<Self>, ConnectionDenied> {
+		self.inner
+			.handle_established_outbound_connection(connection_id, peer, addr, role_override)
 	}
 
 	fn on_swarm_event(&mut self, event: FromSwarm<Self::ConnectionHandler>) {
@@ -168,8 +212,7 @@ impl NetworkBehaviour for CustomProtoWithAddr {
 		&mut self,
 		peer_id: PeerId,
 		connection_id: ConnectionId,
-		event: <<Self::ConnectionHandler as IntoConnectionHandler>::Handler as
-		ConnectionHandler>::OutEvent,
+		event: THandlerOutEvent<Self>,
 	) {
 		self.inner.on_connection_handler_event(peer_id, connection_id, event);
 	}
@@ -178,7 +221,7 @@ impl NetworkBehaviour for CustomProtoWithAddr {
 		&mut self,
 		cx: &mut Context,
 		params: &mut impl PollParameters,
-	) -> Poll<NetworkBehaviourAction<Self::OutEvent, Self::ConnectionHandler>> {
+	) -> Poll<ToSwarm<Self::OutEvent, THandlerInEvent<Self>>> {
 		self.inner.poll(cx, params)
 	}
 }