Open one substream for each notifications protocol

client/finality-grandpa/src/communication/mod.rs

@@ -65,6 +65,7 @@ mod periodic;
 pub(crate) mod tests;
 
 pub use sp_finality_grandpa::GRANDPA_ENGINE_ID;
+pub const GRANDPA_PROTOCOL_NAME: &[u8] = b"/paritytech/grandpa/1";
 
 // cost scalars for reporting peers.
 mod cost {
@@ -185,7 +186,12 @@ impl<B: BlockT, N: Network<B>> NetworkBridge<B, N> {
 		);
 
 		let validator = Arc::new(validator);
-		let gossip_engine = GossipEngine::new(service.clone(), GRANDPA_ENGINE_ID, validator.clone());
+		let gossip_engine = GossipEngine::new(
+			service.clone(),
+			GRANDPA_ENGINE_ID,
+			GRANDPA_PROTOCOL_NAME,
+			validator.clone()
+		);
 
 		{
 			// register all previous votes with the gossip service so that they're

client/finality-grandpa/src/communication/tests.rs

@@ -25,7 +25,7 @@ use std::sync::Arc;
 use sp_keyring::Ed25519Keyring;
 use parity_scale_codec::Encode;
 use sp_runtime::{ConsensusEngineId, traits::NumberFor};
-use std::{pin::Pin, task::{Context, Poll}};
+use std::{borrow::Cow, pin::Pin, task::{Context, Poll}};
 use crate::environment::SharedVoterSetState;
 use sp_finality_grandpa::{AuthorityList, GRANDPA_ENGINE_ID};
 use super::gossip::{self, GossipValidator};
@@ -61,7 +61,7 @@ impl sc_network_gossip::Network<Block> for TestNetwork {
 		let _ = self.sender.unbounded_send(Event::WriteNotification(who, message));
 	}
 
-	fn register_notifications_protocol(&self, _: ConsensusEngineId) {}
+	fn register_notifications_protocol(&self, _: ConsensusEngineId, _: Cow<'static, [u8]>) {}
 
 	fn announce(&self, block: Hash, _associated_data: Vec<u8>) {
 		let _ = self.sender.unbounded_send(Event::Announce(block));

client/finality-grandpa/src/lib.rs

@@ -881,7 +881,10 @@ pub fn setup_disabled_grandpa<B, E, Block: BlockT, RA, N>(
 	// We register the GRANDPA protocol so that we don't consider it an anomaly
 	// to receive GRANDPA messages on the network. We don't process the
 	// messages.
-	network.register_notifications_protocol(communication::GRANDPA_ENGINE_ID);
+	network.register_notifications_protocol(
+		communication::GRANDPA_ENGINE_ID,
+		From::from(communication::GRANDPA_PROTOCOL_NAME),
+	);
 
 	Ok(())
 }

client/network-gossip/src/bridge.rs

@@ -24,7 +24,7 @@ use futures::{prelude::*, channel::mpsc};
 use libp2p::PeerId;
 use parking_lot::Mutex;
 use sp_runtime::{traits::Block as BlockT, ConsensusEngineId};
-use std::{pin::Pin, sync::Arc, task::{Context, Poll}};
+use std::{borrow::Cow, pin::Pin, sync::Arc, task::{Context, Poll}};
 
 /// Wraps around an implementation of the `Network` crate and provides gossiping capabilities on
 /// top of it.
@@ -48,6 +48,7 @@ impl<B: BlockT> GossipEngine<B> {
 	pub fn new<N: Network<B> + Send + Clone + 'static>(
 		mut network: N,
 		engine_id: ConsensusEngineId,
+		protocol_name: impl Into<Cow<'static, [u8]>>,
 		validator: Arc<dyn Validator<B>>,
 	) -> Self where B: 'static {
 		let mut state_machine = ConsensusGossip::new();
@@ -56,7 +57,7 @@ impl<B: BlockT> GossipEngine<B> {
 		// might miss events.
 		let network_event_stream = network.event_stream();
 
-		network.register_notifications_protocol(engine_id);
+		network.register_notifications_protocol(engine_id, protocol_name.into());
 		state_machine.register_validator(&mut network, engine_id, validator);
 
 		let inner = Arc::new(Mutex::new(GossipEngineInner {

client/network-gossip/src/lib.rs

@@ -61,7 +61,7 @@ pub use self::validator::{DiscardAll, MessageIntent, Validator, ValidatorContext
 use futures::prelude::*;
 use sc_network::{specialization::NetworkSpecialization, Event, ExHashT, NetworkService, PeerId, ReputationChange};
 use sp_runtime::{traits::Block as BlockT, ConsensusEngineId};
-use std::{pin::Pin, sync::Arc};
+use std::{borrow::Cow, pin::Pin, sync::Arc};
 
 mod bridge;
 mod state_machine;
@@ -86,7 +86,8 @@ pub trait Network<B: BlockT> {
 	/// See the documentation of [`NetworkService:register_notifications_protocol`] for more information.
 	fn register_notifications_protocol(
 		&self,
-		engine_id: ConsensusEngineId
+		engine_id: ConsensusEngineId,
+		protocol_name: Cow<'static, [u8]>,
 	);
 
 	/// Notify everyone we're connected to that we have the given block.
@@ -116,8 +117,9 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Network<B> for Arc<Netw
 	fn register_notifications_protocol(
 		&self,
 		engine_id: ConsensusEngineId,
+		protocol_name: Cow<'static, [u8]>,
 	) {
-		NetworkService::register_notifications_protocol(self, engine_id)
+		NetworkService::register_notifications_protocol(self, engine_id, protocol_name)
 	}
 
 	fn announce(&self, block: B::Hash, associated_data: Vec<u8>) {

client/network/Cargo.toml

@@ -36,6 +36,7 @@ sc-block-builder = { version = "0.8", path = "../block-builder" }
 sc-client = { version = "0.8", path = "../" }
 sc-client-api = { version = "2.0.0", path = "../api" }
 sc-peerset = { version = "2.0.0", path = "../peerset" }
+pin-project = "0.4.6"
 serde = { version = "1.0.101", features = ["derive"] }
 serde_json = "1.0.41"
 slog = { version = "2.5.2", features = ["nested-values"] }
@@ -51,7 +52,7 @@ sp-runtime = { version = "2.0.0", path = "../../primitives/runtime" }
 substrate-test-client = { version = "2.0.0", optional = true, path = "../../test-utils/client" }
 substrate-test-runtime-client = { version = "2.0.0", optional = true, path = "../../test-utils/runtime/client" }
 thiserror = "1"
-unsigned-varint = { version = "0.3.0", features = ["futures-codec"] }
+unsigned-varint = { version = "0.3.1", features = ["futures", "futures-codec"] }
 void = "1.0.2"
 zeroize = "1.0.0"