Merge Protocol and ProtocolBehaviour (#2962)

* Pass the peerset config to ProtocolBehaviour

* Don't pass the protocol versions

* Move fields from protocol_behaviour.rs to protocol.rs

* Remove LocalNetworkOut

* Move CustomProtos from protocol_behaviour.rs to protocol.rs

* Remove ProtocolBehaviour

* Inline poll()

* Force Behaviour to use Protocol

* Don't even attempt to have working tests

* Remove NetworkOut trait

* Line widths

substrate/core/network/src/behaviour.rs

@@ -17,50 +17,53 @@
 use crate::{
 	debug_info, discovery::DiscoveryBehaviour, discovery::DiscoveryOut, DiscoveryNetBehaviour, event::DhtEvent
 };
+use crate::{ExHashT, specialization::NetworkSpecialization};
+use crate::protocol::{CustomMessageOutcome, Protocol};
 use futures::prelude::*;
 use libp2p::NetworkBehaviour;
-use libp2p::core::{Multiaddr, PeerId, ProtocolsHandler, protocols_handler::IntoProtocolsHandler, PublicKey};
-use libp2p::core::swarm::{ConnectedPoint, NetworkBehaviour, NetworkBehaviourAction};
-use libp2p::core::swarm::{NetworkBehaviourEventProcess, PollParameters};
+use libp2p::core::{Multiaddr, PeerId, PublicKey};
+use libp2p::core::swarm::{NetworkBehaviourAction, NetworkBehaviourEventProcess};
+use libp2p::core::{nodes::Substream, muxing::StreamMuxerBox};
 use libp2p::multihash::Multihash;
 #[cfg(not(target_os = "unknown"))]
 use libp2p::core::swarm::toggle::Toggle;
 #[cfg(not(target_os = "unknown"))]
 use libp2p::mdns::{Mdns, MdnsEvent};
 use log::warn;
+use runtime_primitives::traits::Block as BlockT;
 use std::iter;
 use void;
 
-/// General behaviour of the network.
+/// General behaviour of the network. Combines all protocols together.
 #[derive(NetworkBehaviour)]
-#[behaviour(out_event = "BehaviourOut<TBehaviourEv>", poll_method = "poll")]
-pub struct Behaviour<TBehaviour, TBehaviourEv, TSubstream> {
-	/// Main protocol that handles everything except the discovery and the technicalities.
-	user_protocol: UserBehaviourWrap<TBehaviour>,
+#[behaviour(out_event = "BehaviourOut<B>", poll_method = "poll")]
+pub struct Behaviour<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> {
+	/// All the substrate-specific protocols.
+	substrate: Protocol<B, S, H>,
 	/// Periodically pings and identifies the nodes we are connected to, and store information in a
 	/// cache.
-	debug_info: debug_info::DebugInfoBehaviour<TSubstream>,
+	debug_info: debug_info::DebugInfoBehaviour<Substream<StreamMuxerBox>>,
 	/// Discovers nodes of the network. Defined below.
-	discovery: DiscoveryBehaviour<TSubstream>,
+	discovery: DiscoveryBehaviour<Substream<StreamMuxerBox>>,
 	/// Discovers nodes on the local network.
 	#[cfg(not(target_os = "unknown"))]
-	mdns: Toggle<Mdns<TSubstream>>,
+	mdns: Toggle<Mdns<Substream<StreamMuxerBox>>>,
 
 	/// Queue of events to produce for the outside.
 	#[behaviour(ignore)]
-	events: Vec<BehaviourOut<TBehaviourEv>>,
+	events: Vec<BehaviourOut<B>>,
 }
 
-/// A wrapper for the behavbour event that adds DHT-related event variant.
-pub enum BehaviourOut<TBehaviourEv> {
-	Behaviour(TBehaviourEv),
+/// Event generated by `Behaviour`.
+pub enum BehaviourOut<B: BlockT> {
+	SubstrateAction(CustomMessageOutcome<B>),
 	Dht(DhtEvent),
 }
 
-impl<TBehaviour, TBehaviourEv, TSubstream> Behaviour<TBehaviour, TBehaviourEv, TSubstream> {
+impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Behaviour<B, S, H> {
 	/// Builds a new `Behaviour`.
 	pub fn new(
-		user_protocol: TBehaviour,
+		substrate: Protocol<B, S, H>,
 		user_agent: String,
 		local_public_key: PublicKey,
 		known_addresses: Vec<(PeerId, Multiaddr)>,
@@ -74,7 +77,7 @@ impl<TBehaviour, TBehaviourEv, TSubstream> Behaviour<TBehaviour, TBehaviourEv, T
 		}
 
 		Behaviour {
-			user_protocol: UserBehaviourWrap(user_protocol),
+			substrate,
 			debug_info,
 			discovery: DiscoveryBehaviour::new(local_public_key, known_addresses),
 			#[cfg(not(target_os = "unknown"))]
@@ -113,13 +116,13 @@ impl<TBehaviour, TBehaviourEv, TSubstream> Behaviour<TBehaviour, TBehaviourEv, T
 	}
 
 	/// Returns a shared reference to the user protocol.
-	pub fn user_protocol(&self) -> &TBehaviour {
-		&self.user_protocol.0
+	pub fn user_protocol(&self) -> &Protocol<B, S, H> {
+		&self.substrate
 	}
 
 	/// Returns a mutable reference to the user protocol.
-	pub fn user_protocol_mut(&mut self) -> &mut TBehaviour {
-		&mut self.user_protocol.0
+	pub fn user_protocol_mut(&mut self) -> &mut Protocol<B, S, H> {
+		&mut self.substrate
 	}
 
 	/// Start querying a record from the DHT. Will later produce either a `ValueFound` or a `ValueNotFound` event.
@@ -133,23 +136,22 @@ impl<TBehaviour, TBehaviourEv, TSubstream> Behaviour<TBehaviour, TBehaviourEv, T
 	}
 }
 
-impl<TBehaviour, TBehaviourEv, TSubstream> NetworkBehaviourEventProcess<void::Void> for
-Behaviour<TBehaviour, TBehaviourEv, TSubstream> {
+impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> NetworkBehaviourEventProcess<void::Void> for
+Behaviour<B, S, H> {
 	fn inject_event(&mut self, event: void::Void) {
 		void::unreachable(event)
 	}
 }
 
-impl<TBehaviour, TBehaviourEv, TSubstream> NetworkBehaviourEventProcess<UserEventWrap<TBehaviourEv>> for
-Behaviour<TBehaviour, TBehaviourEv, TSubstream> {
-	fn inject_event(&mut self, event: UserEventWrap<TBehaviourEv>) {
-		self.events.push(BehaviourOut::Behaviour(event.0));
+impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> NetworkBehaviourEventProcess<CustomMessageOutcome<B>> for
+Behaviour<B, S, H> {
+	fn inject_event(&mut self, event: CustomMessageOutcome<B>) {
+		self.events.push(BehaviourOut::SubstrateAction(event));
 	}
 }
 
-impl<TBehaviour, TBehaviourEv, TSubstream> NetworkBehaviourEventProcess<debug_info::DebugInfoEvent>
-	for Behaviour<TBehaviour, TBehaviourEv, TSubstream>
-	where TBehaviour: DiscoveryNetBehaviour {
+impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> NetworkBehaviourEventProcess<debug_info::DebugInfoEvent>
+	for Behaviour<B, S, H> {
 	fn inject_event(&mut self, event: debug_info::DebugInfoEvent) {
 		let debug_info::DebugInfoEvent::Identified { peer_id, mut info } = event;
 		if !info.protocol_version.contains("substrate") {
@@ -165,17 +167,16 @@ impl<TBehaviour, TBehaviourEv, TSubstream> NetworkBehaviourEventProcess<debug_in
 		for addr in &info.listen_addrs {
 			self.discovery.add_self_reported_address(&peer_id, addr.clone());
 		}
-		self.user_protocol.0.add_discovered_nodes(iter::once(peer_id.clone()));
+		self.substrate.add_discovered_nodes(iter::once(peer_id.clone()));
 	}
 }
 
-impl<TBehaviour, TBehaviourEv, TSubstream> NetworkBehaviourEventProcess<DiscoveryOut>
-	for Behaviour<TBehaviour, TBehaviourEv, TSubstream>
-	where TBehaviour: DiscoveryNetBehaviour {
+impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> NetworkBehaviourEventProcess<DiscoveryOut>
+	for Behaviour<B, S, H> {
 	fn inject_event(&mut self, out: DiscoveryOut) {
 		match out {
 			DiscoveryOut::Discovered(peer_id) => {
-				self.user_protocol.0.add_discovered_nodes(iter::once(peer_id));
+				self.substrate.add_discovered_nodes(iter::once(peer_id));
 			}
 			DiscoveryOut::ValueFound(results) => {
 				self.events.push(BehaviourOut::Dht(DhtEvent::ValueFound(results)));
@@ -194,21 +195,20 @@ impl<TBehaviour, TBehaviourEv, TSubstream> NetworkBehaviourEventProcess<Discover
 }
 
 #[cfg(not(target_os = "unknown"))]
-impl<TBehaviour, TBehaviourEv, TSubstream> NetworkBehaviourEventProcess<MdnsEvent> for
-	Behaviour<TBehaviour, TBehaviourEv, TSubstream>
-	where TBehaviour: DiscoveryNetBehaviour {
+impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> NetworkBehaviourEventProcess<MdnsEvent> for
+	Behaviour<B, S, H> {
 	fn inject_event(&mut self, event: MdnsEvent) {
 		match event {
 			MdnsEvent::Discovered(list) => {
-				self.user_protocol.0.add_discovered_nodes(list.into_iter().map(|(peer_id, _)| peer_id));
+				self.substrate.add_discovered_nodes(list.into_iter().map(|(peer_id, _)| peer_id));
 			},
 			MdnsEvent::Expired(_) => {}
 		}
 	}
 }
 
-impl<TBehaviour, TBehaviourEv, TSubstream> Behaviour<TBehaviour, TBehaviourEv, TSubstream> {
-	fn poll<TEv>(&mut self) -> Async<NetworkBehaviourAction<TEv, BehaviourOut<TBehaviourEv>>> {
+impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Behaviour<B, S, H> {
+	fn poll<TEv>(&mut self) -> Async<NetworkBehaviourAction<TEv, BehaviourOut<B>>> {
 		if !self.events.is_empty() {
 			return Async::Ready(NetworkBehaviourAction::GenerateEvent(self.events.remove(0)))
 		}
@@ -216,71 +216,3 @@ impl<TBehaviour, TBehaviourEv, TSubstream> Behaviour<TBehaviour, TBehaviourEv, T
 		Async::NotReady
 	}
 }
-
-/// Because of limitations with the network behaviour custom derive and trait impl duplication, we
-/// have to wrap the user protocol into a struct.
-pub struct UserBehaviourWrap<TInner>(TInner);
-/// Event produced by `UserBehaviourWrap`.
-pub struct UserEventWrap<TInner>(TInner);
-impl<TInner: NetworkBehaviour> NetworkBehaviour for UserBehaviourWrap<TInner> {
-	type ProtocolsHandler = TInner::ProtocolsHandler;
-	type OutEvent = UserEventWrap<TInner::OutEvent>;
-	fn new_handler(&mut self) -> Self::ProtocolsHandler { self.0.new_handler() }
-	fn addresses_of_peer(&mut self, peer_id: &PeerId) -> Vec<Multiaddr> {
-		self.0.addresses_of_peer(peer_id)
-	}
-	fn inject_connected(&mut self, peer_id: PeerId, endpoint: ConnectedPoint) {
-		self.0.inject_connected(peer_id, endpoint)
-	}
-	fn inject_disconnected(&mut self, peer_id: &PeerId, endpoint: ConnectedPoint) {
-		self.0.inject_disconnected(peer_id, endpoint)
-	}
-	fn inject_node_event(
-		&mut self,
-		peer_id: PeerId,
-		event: <<Self::ProtocolsHandler as IntoProtocolsHandler>::Handler as ProtocolsHandler>::OutEvent
-	) {
-		self.0.inject_node_event(peer_id, event)
-	}
-	fn poll(
-		&mut self,
-		params: &mut impl PollParameters
-	) -> Async<
-		NetworkBehaviourAction<
-			<<Self::ProtocolsHandler as IntoProtocolsHandler>::Handler as ProtocolsHandler>::InEvent,
-			Self::OutEvent
-		>
-	> {
-		match self.0.poll(params) {
-			Async::NotReady => Async::NotReady,
-			Async::Ready(NetworkBehaviourAction::GenerateEvent(ev)) =>
-				Async::Ready(NetworkBehaviourAction::GenerateEvent(UserEventWrap(ev))),
-			Async::Ready(NetworkBehaviourAction::DialAddress { address }) =>
-				Async::Ready(NetworkBehaviourAction::DialAddress { address }),
-			Async::Ready(NetworkBehaviourAction::DialPeer { peer_id }) =>
-				Async::Ready(NetworkBehaviourAction::DialPeer { peer_id }),
-			Async::Ready(NetworkBehaviourAction::SendEvent { peer_id, event }) =>
-				Async::Ready(NetworkBehaviourAction::SendEvent { peer_id, event }),
-			Async::Ready(NetworkBehaviourAction::ReportObservedAddr { address }) =>
-				Async::Ready(NetworkBehaviourAction::ReportObservedAddr { address }),
-		}
-	}
-	fn inject_replaced(&mut self, peer_id: PeerId, closed_endpoint: ConnectedPoint, new_endpoint: ConnectedPoint) {
-		self.0.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.0.inject_addr_reach_failure(peer_id, addr, error)
-	}
-	fn inject_dial_failure(&mut self, peer_id: &PeerId) {
-		self.0.inject_dial_failure(peer_id)
-	}
-	fn inject_new_listen_addr(&mut self, addr: &Multiaddr) {
-		self.0.inject_new_listen_addr(addr)
-	}
-	fn inject_expired_listen_addr(&mut self, addr: &Multiaddr) {
-		self.0.inject_expired_listen_addr(addr)
-	}
-	fn inject_new_external_addr(&mut self, addr: &Multiaddr) {
-		self.0.inject_new_external_addr(addr)
-	}
-}

substrate/core/network/src/lib.rs

@@ -169,7 +169,6 @@ mod discovery;
 mod on_demand_layer;
 #[macro_use]
 mod protocol;
-mod protocol_behaviour;
 mod service;
 mod transport;
 

substrate/core/network/src/protocol_behaviour.rs

@@ -1,468 +0,0 @@
-// Copyright 2019 Parity Technologies (UK) Ltd.
-// This file is part of Substrate.
-
-// Substrate is free software: you can redistribute it and/or modify
-// it under the terms of the GNU General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-
-// Substrate is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-// GNU General Public License for more details.
-
-// You should have received a copy of the GNU General Public License
-// along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
-
-//! Implementation of libp2p's `NetworkBehaviour` trait that handles everything Substrate-specific.
-
-use crate::{ExHashT, DiscoveryNetBehaviour, ProtocolId};
-use crate::custom_proto::{CustomProto, CustomProtoOut};
-use crate::chain::{Client, FinalityProofProvider};
-use crate::protocol::{self, event::Event, CustomMessageOutcome, Protocol, ProtocolConfig, sync::SyncState};
-use crate::protocol::{PeerInfo, NetworkOut, message::Message, on_demand::RequestData};
-use crate::protocol::consensus_gossip::MessageRecipient as GossipMessageRecipient;
-use crate::protocol::specialization::NetworkSpecialization;
-use crate::service::TransactionPool;
-
-use client::light::fetcher::FetchChecker;
-use futures::prelude::*;
-use consensus::import_queue::SharedFinalityProofRequestBuilder;
-use log::debug;
-use libp2p::{PeerId, Multiaddr};
-use libp2p::core::swarm::{ConnectedPoint, NetworkBehaviour, NetworkBehaviourAction, PollParameters};
-use libp2p::core::{nodes::Substream, muxing::StreamMuxerBox};
-use libp2p::core::protocols_handler::{ProtocolsHandler, IntoProtocolsHandler};
-use runtime_primitives::{traits::{Block as BlockT, NumberFor}, ConsensusEngineId};
-use std::sync::Arc;
-
-/// Implementation of `NetworkBehaviour` that handles everything related to Substrate and Polkadot.
-pub struct ProtocolBehaviour<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> {
-	/// Handles opening the unique substream and sending and receiving raw messages.
-	behaviour: CustomProto<Message<B>, Substream<StreamMuxerBox>>,
-	/// Handles the logic behind the raw messages that we receive.
-	protocol: Protocol<B, S, H>,
-	/// Used to report reputation changes.
-	peerset_handle: peerset::PeersetHandle,
-	transaction_pool: Arc<dyn TransactionPool<H, B>>,
-	/// When asked for a proof of finality, we use this struct to build one.
-	finality_proof_provider: Option<Arc<dyn FinalityProofProvider<B>>>,
-}
-
-impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> ProtocolBehaviour<B, S, H> {
-	/// Builds a new `ProtocolBehaviour`.
-	pub fn new(
-		config: ProtocolConfig,
-		chain: Arc<dyn Client<B>>,
-		checker: Arc<dyn FetchChecker<B>>,
-		specialization: S,
-		transaction_pool: Arc<dyn TransactionPool<H, B>>,
-		finality_proof_provider: Option<Arc<dyn FinalityProofProvider<B>>>,
-		protocol_id: ProtocolId,
-		versions: &[u8],
-		peerset: peerset::Peerset,
-		peerset_handle: peerset::PeersetHandle,
-	) -> crate::error::Result<Self> {
-		let protocol = Protocol::new(config, chain, checker, specialization)?;
-		let behaviour = CustomProto::new(protocol_id, versions, peerset);
-
-		Ok(ProtocolBehaviour {
-			protocol,
-			behaviour,
-			peerset_handle,
-			transaction_pool,
-			finality_proof_provider,
-		})
-	}
-
-	/// Returns the list of all the peers we have an open channel to.
-	pub fn open_peers(&self) -> impl Iterator<Item = &PeerId> {
-		self.behaviour.open_peers()
-	}
-
-	/// Returns true if we have a channel open with this node.
-	pub fn is_open(&self, peer_id: &PeerId) -> bool {
-		self.behaviour.is_open(peer_id)
-	}
-
-	/// Disconnects the given peer if we are connected to it.
-	pub fn disconnect_peer(&mut self, peer_id: &PeerId) {
-		self.behaviour.disconnect_peer(peer_id)
-	}
-
-	/// Adjusts the reputation of a node.
-	pub fn report_peer(&mut self, who: PeerId, reputation: i32) {
-		self.peerset_handle.report_peer(who, reputation)
-	}
-
-	/// Returns true if we try to open protocols with the given peer.
-	pub fn is_enabled(&self, peer_id: &PeerId) -> bool {
-		self.behaviour.is_enabled(peer_id)
-	}
-
-	/// Sends a message to a peer.
-	///
-	/// Has no effect if the custom protocol is not open with the given peer.
-	///
-	/// Also note that even we have a valid open substream, it may in fact be already closed
-	/// without us knowing, in which case the packet will not be received.
-	pub fn send_packet(&mut self, target: &PeerId, message: Message<B>) {
-		self.behaviour.send_packet(target, message)
-	}
-
-	/// Returns the state of the peerset manager, for debugging purposes.
-	pub fn peerset_debug_info(&mut self) -> serde_json::Value {
-		self.behaviour.peerset_debug_info()
-	}
-
-	/// Returns the number of peers we're connected to.
-	pub fn num_connected_peers(&self) -> usize {
-		self.protocol.num_connected_peers()
-	}
-
-	/// Returns the number of peers we're connected to and that are being queried.
-	pub fn num_active_peers(&self) -> usize {
-		self.protocol.num_active_peers()
-	}
-
-	/// Current global sync state.
-	pub fn sync_state(&self) -> SyncState {
-		self.protocol.sync_state()
-	}
-
-	/// Target sync block number.
-	pub fn best_seen_block(&self) -> Option<NumberFor<B>> {
-		self.protocol.best_seen_block()
-	}
-
-	/// Number of peers participating in syncing.
-	pub fn num_sync_peers(&self) -> u32 {
-		self.protocol.num_sync_peers()
-	}
-
-	/// Starts a new data demand request.
-	///
-	/// The parameter contains a `Sender` where the result, once received, must be sent.
-	pub(crate) fn add_on_demand_request(&mut self, rq: RequestData<B>) {
-		self.protocol.add_on_demand_request(
-			&mut LocalNetworkOut { inner: &mut self.behaviour, peerset_handle: &self.peerset_handle },
-			rq
-		);
-	}
-
-	/// Returns information about all the peers we are connected to after the handshake message.
-	pub fn peers_info(&self) -> impl Iterator<Item = (&PeerId, &PeerInfo<B>)> {
-		self.protocol.peers_info()
-	}
-
-	/// Locks `self` and gives access to the protocol and a context that can be used in order to
-	/// use `consensus_gossip_lock` or `specialization_lock`.
-	///
-	/// **Important**: ONLY USE THIS FUNCTION TO CALL `consensus_gossip_lock` or `specialization_lock`.
-	/// This function is a very bad API.
-	pub fn protocol_context_lock<'a>(
-		&'a mut self,
-	) -> (&'a mut Protocol<B, S, H>, LocalNetworkOut<'a, B>) {
-		let net_out = LocalNetworkOut { inner: &mut self.behaviour, peerset_handle: &self.peerset_handle };
-		(&mut self.protocol, net_out)
-	}
-
-	/// Gossip a consensus message to the network.
-	pub fn gossip_consensus_message(
-		&mut self,
-		topic: B::Hash,
-		engine_id: ConsensusEngineId,
-		message: Vec<u8>,
-		recipient: GossipMessageRecipient,
-	) {
-		self.protocol.gossip_consensus_message(
-			&mut LocalNetworkOut { inner: &mut self.behaviour, peerset_handle: &self.peerset_handle },
-			topic,
-			engine_id,
-			message,
-			recipient
-		);
-	}
-
-	/// Call when we must propagate ready extrinsics to peers.
-	pub fn propagate_extrinsics(&mut self) {
-		self.protocol.propagate_extrinsics(
-			&mut LocalNetworkOut { inner: &mut self.behaviour, peerset_handle: &self.peerset_handle },
-			&*self.transaction_pool
-		)
-	}
-
-	/// Make sure an important block is propagated to peers.
-	///
-	/// In chain-based consensus, we often need to make sure non-best forks are
-	/// at least temporarily synced.
-	pub fn announce_block(&mut self, hash: B::Hash) {
-		self.protocol.announce_block(
-			&mut LocalNetworkOut { inner: &mut self.behaviour, peerset_handle: &self.peerset_handle },
-			hash
-		)
-	}
-
-	/// Call this when a block has been imported in the import queue and we should announce it on
-	/// the network.
-	pub fn on_block_imported(&mut self, hash: B::Hash, header: &B::Header) {
-		self.protocol.on_block_imported(
-			&mut LocalNetworkOut { inner: &mut self.behaviour, peerset_handle: &self.peerset_handle },
-			hash,
-			header
-		)
-	}
-
-	/// Call this when a block has been finalized. The sync layer may have some additional
-	/// requesting to perform.
-	pub fn on_block_finalized(&mut self, hash: B::Hash, header: &B::Header) {
-		self.protocol.on_block_finalized(
-			&mut LocalNetworkOut { inner: &mut self.behaviour, peerset_handle: &self.peerset_handle },
-			hash,
-			header
-		)
-	}
-
-	/// Request a justification for the given block.
-	///
-	/// Uses `protocol` to queue a new justification request and tries to dispatch all pending
-	/// requests.
-	pub fn request_justification(&mut self, hash: &B::Hash, number: NumberFor<B>) {
-		self.protocol.request_justification(
-			&mut LocalNetworkOut { inner: &mut self.behaviour, peerset_handle: &self.peerset_handle },
-			hash,
-			number
-		)
-	}
-
-	/// Clears all pending justification requests.
-	pub fn clear_justification_requests(&mut self) {
-		self.protocol.clear_justification_requests()
-	}
-
-	/// A batch of blocks have been processed, with or without errors.
-	/// Call this when a batch of blocks have been processed by the import queue, with or without
-	/// errors.
-	pub fn blocks_processed(
-		&mut self,
-		processed_blocks: Vec<B::Hash>,
-		has_error: bool,
-	) {
-		self.protocol.blocks_processed(
-			&mut LocalNetworkOut { inner: &mut self.behaviour, peerset_handle: &self.peerset_handle },
-			processed_blocks,
-			has_error,
-		)
-	}
-
-	/// Restart the sync process.
-	pub fn restart(&mut self) {
-		let mut net_out = LocalNetworkOut { inner: &mut self.behaviour, peerset_handle: &self.peerset_handle };
-		self.protocol.restart(&mut net_out);
-	}
-
-	/// Notify about successful import of the given block.
-	pub fn block_imported(&mut self, hash: &B::Hash, number: NumberFor<B>) {
-		self.protocol.block_imported(hash, number)
-	}
-
-	pub fn set_finality_proof_request_builder(&mut self, request_builder: SharedFinalityProofRequestBuilder<B>) {
-		self.protocol.set_finality_proof_request_builder(request_builder)
-	}
-
-	/// Call this when a justification has been processed by the import queue, with or without
-	/// errors.
-	pub fn justification_import_result(&mut self, hash: B::Hash, number: NumberFor<B>, success: bool) {
-		self.protocol.justification_import_result(hash, number, success)
-	}
-
-	/// The networking-level event has happened.
-	pub fn on_event(&mut self, event: Event) {
-		self.protocol.on_event(event);
-	}
-
-	/// Request a finality proof for the given block.
-	///
-	/// Queues a new finality proof request and tries to dispatch all pending requests.
-	pub fn request_finality_proof(
-		&mut self,
-		hash: &B::Hash,
-		number: NumberFor<B>,
-	) {
-		self.protocol.request_finality_proof(
-			&mut LocalNetworkOut { inner: &mut self.behaviour, peerset_handle: &self.peerset_handle },
-			&hash,
-			number,
-		);
-	}
-
-	pub fn finality_proof_import_result(
-		&mut self,
-		request_block: (B::Hash, NumberFor<B>),
-		finalization_result: Result<(B::Hash, NumberFor<B>), ()>,
-	) {
-		self.protocol.finality_proof_import_result(request_block, finalization_result)
-	}
-
-	pub fn tick(&mut self) {
-		self.protocol.tick(&mut LocalNetworkOut { inner: &mut self.behaviour, peerset_handle: &self.peerset_handle });
-	}
-}
-
-impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> NetworkBehaviour for
-ProtocolBehaviour<B, S, H> {
-	type ProtocolsHandler = <CustomProto<Message<B>, Substream<StreamMuxerBox>> as NetworkBehaviour>::ProtocolsHandler;
-	type OutEvent = CustomMessageOutcome<B>;
-
-	fn new_handler(&mut self) -> Self::ProtocolsHandler {
-		self.behaviour.new_handler()
-	}
-
-	fn addresses_of_peer(&mut self, peer_id: &PeerId) -> Vec<Multiaddr> {
-		self.behaviour.addresses_of_peer(peer_id)
-	}
-
-	fn inject_connected(&mut self, peer_id: PeerId, endpoint: ConnectedPoint) {
-		self.behaviour.inject_connected(peer_id, endpoint)
-	}
-
-	fn inject_disconnected(&mut self, peer_id: &PeerId, endpoint: ConnectedPoint) {
-		self.behaviour.inject_disconnected(peer_id, endpoint)
-	}
-
-	fn inject_node_event(
-		&mut self,
-		peer_id: PeerId,
-		event: <<Self::ProtocolsHandler as IntoProtocolsHandler>::Handler as ProtocolsHandler>::OutEvent,
-	) {
-		self.behaviour.inject_node_event(peer_id, event)
-	}
-
-	fn poll(
-		&mut self,
-		params: &mut impl PollParameters,
-	) -> Async<
-		NetworkBehaviourAction<
-			<<Self::ProtocolsHandler as IntoProtocolsHandler>::Handler as ProtocolsHandler>::InEvent,
-			Self::OutEvent
-		>
-	> {
-		let mut net_out = LocalNetworkOut { inner: &mut self.behaviour, peerset_handle: &self.peerset_handle };
-		match self.protocol.poll(&mut net_out, &*self.transaction_pool) {
-			Ok(Async::Ready(v)) => void::unreachable(v),
-			Ok(Async::NotReady) => {}
-			Err(err) => void::unreachable(err),
-		}
-
-		let event = match self.behaviour.poll(params) {
-			Async::NotReady => return Async::NotReady,
-			Async::Ready(NetworkBehaviourAction::GenerateEvent(ev)) => ev,
-			Async::Ready(NetworkBehaviourAction::DialAddress { address }) =>
-				return Async::Ready(NetworkBehaviourAction::DialAddress { address }),
-			Async::Ready(NetworkBehaviourAction::DialPeer { peer_id }) =>
-				return Async::Ready(NetworkBehaviourAction::DialPeer { peer_id }),
-			Async::Ready(NetworkBehaviourAction::SendEvent { peer_id, event }) =>
-				return Async::Ready(NetworkBehaviourAction::SendEvent { peer_id, event }),
-			Async::Ready(NetworkBehaviourAction::ReportObservedAddr { address }) =>
-				return Async::Ready(NetworkBehaviourAction::ReportObservedAddr { address }),
-		};
-
-		let mut network_out = LocalNetworkOut {
-			inner: &mut self.behaviour,
-			peerset_handle: &self.peerset_handle,
-		};
-
-		let outcome = match event {
-			CustomProtoOut::CustomProtocolOpen { peer_id, version, .. } => {
-				debug_assert!(
-					version <= protocol::CURRENT_VERSION as u8
-					&& version >= protocol::MIN_VERSION as u8
-				);
-				self.protocol.on_peer_connected(&mut network_out, peer_id);
-				CustomMessageOutcome::None
-			}
-			CustomProtoOut::CustomProtocolClosed { peer_id, .. } => {
-				self.protocol.on_peer_disconnected(&mut network_out, peer_id);
-				CustomMessageOutcome::None
-			},
-			CustomProtoOut::CustomMessage { peer_id, message } =>
-				self.protocol.on_custom_message(
-					&mut network_out,
-					&*self.transaction_pool,
-					peer_id,
-					message,
-					self.finality_proof_provider.as_ref().map(|p| &**p)
-				),
-			CustomProtoOut::Clogged { peer_id, messages } => {
-				debug!(target: "sync", "{} clogging messages:", messages.len());
-				for msg in messages.into_iter().take(5) {
-					debug!(target: "sync", "{:?}", msg);
-					self.protocol.on_clogged_peer(&mut network_out, peer_id.clone(), Some(msg));
-				}
-				CustomMessageOutcome::None
-			}
-		};
-
-		if let CustomMessageOutcome::None = outcome {
-			Async::NotReady
-		} else {
-			Async::Ready(NetworkBehaviourAction::GenerateEvent(outcome))
-		}
-	}
-
-	fn inject_replaced(&mut self, peer_id: PeerId, closed_endpoint: ConnectedPoint, new_endpoint: ConnectedPoint) {
-		self.behaviour.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.behaviour.inject_addr_reach_failure(peer_id, addr, error)
-	}
-
-	fn inject_dial_failure(&mut self, peer_id: &PeerId) {
-		self.behaviour.inject_dial_failure(peer_id)
-	}
-
-	fn inject_new_listen_addr(&mut self, addr: &Multiaddr) {
-		self.behaviour.inject_new_listen_addr(addr)
-	}
-
-	fn inject_expired_listen_addr(&mut self, addr: &Multiaddr) {
-		self.behaviour.inject_expired_listen_addr(addr)
-	}
-
-	fn inject_new_external_addr(&mut self, addr: &Multiaddr) {
-		self.behaviour.inject_new_external_addr(addr)
-	}
-}
-
-impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> DiscoveryNetBehaviour
-	for ProtocolBehaviour<B, S, H> {
-	fn add_discovered_nodes(&mut self, peer_ids: impl Iterator<Item = PeerId>) {
-		self.behaviour.add_discovered_nodes(peer_ids)
-	}
-}
-
-/// Has to be public for stupid API reasons. This should be made private again ASAP.
-pub struct LocalNetworkOut<'a, B: BlockT> {
-	inner: &'a mut CustomProto<Message<B>, Substream<StreamMuxerBox>>,
-	peerset_handle: &'a peerset::PeersetHandle,
-}
-
-impl<'a, B: BlockT> NetworkOut<B> for LocalNetworkOut<'a, B> {
-	fn report_peer(&mut self, who: PeerId, reputation: i32) {
-		self.peerset_handle.report_peer(who, reputation)
-	}
-
-	fn disconnect_peer(&mut self, who: PeerId) {
-		self.inner.disconnect_peer(&who)
-	}
-
-	fn send_message(&mut self, who: PeerId, message: Message<B>) {
-		self.inner.send_packet(&who, message)
-	}
-}

substrate/core/network/src/service.rs

@@ -22,11 +22,11 @@ use std::time::Duration;
 
 use log::{warn, error, info};
 use libp2p::core::swarm::NetworkBehaviour;
-use libp2p::core::{nodes::Substream, transport::boxed::Boxed, muxing::StreamMuxerBox};
+use libp2p::core::{transport::boxed::Boxed, muxing::StreamMuxerBox};
 use libp2p::{Multiaddr, multihash::Multihash};
 use futures::{prelude::*, sync::oneshot, sync::mpsc};
 use parking_lot::{Mutex, RwLock};
-use crate::protocol_behaviour::ProtocolBehaviour;
+use crate::protocol::Protocol;
 use crate::{behaviour::{Behaviour, BehaviourOut}, parse_str_addr};
 use crate::{NetworkState, NetworkStateNotConnectedPeer, NetworkStatePeer};
 use crate::{transport, config::NodeKeyConfig, config::NonReservedPeerMode};
@@ -149,14 +149,13 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> NetworkWorker
 			}
 		}
 
-		// Build the peerset.
-		let (peerset, peerset_handle) = peerset::Peerset::from_config(peerset::PeersetConfig {
+		let peerset_config = peerset::PeersetConfig {
 			in_peers: params.network_config.in_peers,
 			out_peers: params.network_config.out_peers,
 			bootnodes,
 			reserved_only: params.network_config.non_reserved_mode == NonReservedPeerMode::Deny,
 			reserved_nodes,
-		});
+		};
 
 		// Private and public keys configuration.
 		if let NodeKeyConfig::Secp256k1(_) = params.network_config.node_key {
@@ -171,7 +170,7 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> NetworkWorker
 		let is_offline = Arc::new(AtomicBool::new(true));
 		let is_major_syncing = Arc::new(AtomicBool::new(false));
 		let peers: Arc<RwLock<HashMap<PeerId, ConnectedPeer<B>>>> = Arc::new(Default::default());
-		let protocol = ProtocolBehaviour::new(
+		let (protocol, peerset_handle) = Protocol::new(
 			protocol::ProtocolConfig { roles: params.roles },
 			params.chain,
 			params.on_demand.as_ref().map(|od| od.checker().clone())
@@ -180,9 +179,7 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> NetworkWorker
 			params.transaction_pool,
 			params.finality_proof_provider,
 			params.protocol_id,
-			&((protocol::MIN_VERSION as u8)..=(protocol::CURRENT_VERSION as u8)).collect::<Vec<u8>>(),
-			peerset,
-			peerset_handle.clone(),
+			peerset_config,
 		)?;
 
 		// Build the swarm.
@@ -727,13 +724,13 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> Future for Ne
 				ProtocolMsg::BlockFinalized(hash, header) =>
 					network_service.user_protocol_mut().on_block_finalized(hash, &header),
 				ProtocolMsg::ExecuteWithSpec(task) => {
-					let (protocol, mut net_out) = network_service.user_protocol_mut().protocol_context_lock();
-					let (mut context, spec) = protocol.specialization_lock(&mut net_out);
+					let protocol = network_service.user_protocol_mut();
+					let (mut context, spec) = protocol.specialization_lock();
 					task.call_box(spec, &mut context);
 				},
 				ProtocolMsg::ExecuteWithGossip(task) => {
-					let (protocol, mut net_out) = network_service.user_protocol_mut().protocol_context_lock();
-					let (mut context, gossip) = protocol.consensus_gossip_lock(&mut net_out);
+					let protocol = network_service.user_protocol_mut();
+					let (mut context, gossip) = protocol.consensus_gossip_lock();
 					task.call_box(gossip, &mut context);
 				}
 				ProtocolMsg::GossipConsensusMessage(topic, engine_id, message, recipient) =>
@@ -774,7 +771,7 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> Future for Ne
 
 			let outcome = match poll_value {
 				Ok(Async::NotReady) => break,
-				Ok(Async::Ready(Some(BehaviourOut::Behaviour(outcome)))) => outcome,
+				Ok(Async::Ready(Some(BehaviourOut::SubstrateAction(outcome)))) => outcome,
 				Ok(Async::Ready(Some(BehaviourOut::Dht(ev)))) => {
 					network_service.user_protocol_mut()
 						.on_event(Event::Dht(ev));
@@ -812,5 +809,5 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> Future for Ne
 /// The libp2p swarm, customized for our needs.
 type Swarm<B, S, H> = libp2p::core::Swarm<
 	Boxed<(PeerId, StreamMuxerBox), io::Error>,
-	Behaviour<ProtocolBehaviour<B, S, H>, CustomMessageOutcome<B>, Substream<StreamMuxerBox>>
+	Behaviour<B, S, H>
 >;