rewrite network protocol/service to use channels

core/finality-grandpa/src/lib.rs

@@ -230,18 +230,18 @@ pub trait Network: Clone {
 }
 
 ///  Bridge between NetworkService, gossiping consensus messages and Grandpa
-pub struct NetworkBridge<B: BlockT, S: network::specialization::NetworkSpecialization<B>, H: ExHashT> {
-	service: Arc<NetworkService<B, S, H>>
+pub struct NetworkBridge<B: BlockT> {
+	service: Arc<NetworkService<B>>
 }
 
-impl<B: BlockT, S: network::specialization::NetworkSpecialization<B>, H: ExHashT> NetworkBridge<B, S, H> {
+impl<B: BlockT> NetworkBridge<B> {
 	/// Create a new NetworkBridge to the given NetworkService
-	pub fn new(service: Arc<NetworkService<B, S, H>>) -> Self {
+	pub fn new(service: Arc<NetworkService<B>>) -> Self {
 		NetworkBridge { service }
 	}
 }
 
-impl<B: BlockT, S: network::specialization::NetworkSpecialization<B>, H: ExHashT> Clone for NetworkBridge<B, S, H> {
+impl<B: BlockT> Clone for NetworkBridge<B> {
 	fn clone(&self) -> Self {
 		NetworkBridge {
 			service: Arc::clone(&self.service)
@@ -257,10 +257,10 @@ fn commit_topic<B: BlockT>(set_id: u64) -> B::Hash {
 	<<B::Header as HeaderT>::Hashing as HashT>::hash(format!("{}-COMMITS", set_id).as_bytes())
 }
 
-impl<B: BlockT, S: network::specialization::NetworkSpecialization<B>, H: ExHashT> Network for NetworkBridge<B, S, H> {
+impl<B: BlockT> Network for NetworkBridge<B> {
 	type In = mpsc::UnboundedReceiver<ConsensusMessage>;
 	fn messages_for(&self, round: u64, set_id: u64) -> Self::In {
-		self.service.consensus_gossip().write().messages_for(message_topic::<B>(round, set_id))
+		self.service.consensus_gossip_messages_for(message_topic::<B>(round, set_id))
 	}
 
 	fn send_message(&self, round: u64, set_id: u64, message: Vec<u8>) {
@@ -270,11 +270,11 @@ impl<B: BlockT, S: network::specialization::NetworkSpecialization<B>, H: ExHashT
 
 	fn drop_messages(&self, round: u64, set_id: u64) {
 		let topic = message_topic::<B>(round, set_id);
-		self.service.consensus_gossip().write().collect_garbage(|t| t == &topic);
+		self.service.consensus_gossip_collect_garbage_for(topic);
 	}
 
 	fn commit_messages(&self, set_id: u64) -> Self::In {
-		self.service.consensus_gossip().write().messages_for(commit_topic::<B>(set_id))
+		self.service.consensus_gossip_messages_for(commit_topic::<B>(set_id))
 	}
 
 	fn send_commit(&self, set_id: u64, message: Vec<u8>) {

core/finality-grandpa/src/tests.rs

@@ -178,10 +178,8 @@ impl Network for MessageRouting {
 	fn messages_for(&self, round: u64, set_id: u64) -> Self::In {
 		let inner = self.inner.lock();
 		let peer = inner.peer(self.peer_id);
-		let mut gossip = peer.consensus_gossip().write();
-		let messages = peer.with_spec(move |_, _| {
-			gossip.messages_for(make_topic(round, set_id))
-		});
+		//let mut gossip = peer.consensus_gossip().write();
+		let messages = peer.consensus_gossip_messages_for(make_topic(round, set_id));
 
 		let messages = messages.map_err(
 			move |_| panic!("Messages for round {} dropped too early", round)
@@ -200,19 +198,15 @@ impl Network for MessageRouting {
 		let topic = make_topic(round, set_id);
 		let inner = self.inner.lock();
 		let peer = inner.peer(self.peer_id);
-		let mut gossip = peer.consensus_gossip().write();
-		peer.with_spec(move |_, _| {
-			gossip.collect_garbage(|t| t == &topic)
-		});
+		//let mut gossip = peer.consensus_gossip().write();
+        peer.consensus_gossip_collect_garbage_for(topic);
 	}
 
 	fn commit_messages(&self, set_id: u64) -> Self::In {
 		let inner = self.inner.lock();
 		let peer = inner.peer(self.peer_id);
-		let mut gossip = peer.consensus_gossip().write();
-		let messages = peer.with_spec(move |_, _| {
-			gossip.messages_for(make_commit_topic(set_id))
-		});
+		//let mut gossip = peer.consensus_gossip().write();
+        let messages = peer.consensus_gossip_messages_for(make_commit_topic(set_id));
 
 		let messages = messages.map_err(
 			move |_| panic!("Commit messages for set {} dropped too early", set_id)

core/network-libp2p/Cargo.toml

@@ -7,6 +7,7 @@ version = "0.1.0"
 authors = ["Parity Technologies <admin@parity.io>"]
 
 [dependencies]
+crossbeam-channel = "0.3.4"
 bytes = "0.4"
 error-chain = { version = "0.12", default-features = false }
 fnv = "1.0"

core/network-libp2p/src/lib.rs

@@ -18,6 +18,7 @@
 
 #![recursion_limit = "128"]
 
+extern crate crossbeam_channel;
 extern crate parking_lot;
 extern crate fnv;
 extern crate futures;
@@ -55,7 +56,7 @@ pub use custom_proto::RegisteredProtocol;
 pub use error::{Error, ErrorKind, DisconnectReason};
 pub use libp2p::{Multiaddr, multiaddr::{Protocol}, multiaddr, PeerId};
 pub use secret::obtain_private_key;
-pub use service_task::{start_service, Service, ServiceEvent};
+pub use service_task::{start_service, NetworkMsg, NetworkPort, Service, ServiceEvent};
 pub use traits::{NetworkConfiguration, NodeIndex, NodeId, NonReservedPeerMode};
 pub use traits::{ProtocolId, Secret, Severity};