Traffic statistics

core/network/src/legacy_proto/behaviour.rs

@@ -17,15 +17,14 @@
 use crate::{DiscoveryNetBehaviour, config::ProtocolId};
 use crate::legacy_proto::handler::{CustomProtoHandlerProto, CustomProtoHandlerOut, CustomProtoHandlerIn};
 use crate::legacy_proto::upgrade::RegisteredProtocol;
-use crate::protocol::message::Message;
+use bytes::BytesMut;
 use fnv::FnvHashMap;
 use futures::prelude::*;
 use futures03::{compat::Compat, TryFutureExt as _, StreamExt as _, TryStreamExt as _};
 use libp2p::core::{ConnectedPoint, Multiaddr, PeerId};
 use libp2p::swarm::{NetworkBehaviour, NetworkBehaviourAction, PollParameters};
 use log::{debug, error, trace, warn};
 use rand::distributions::{Distribution as _, Uniform};
-use sr_primitives::traits::Block as BlockT;
 use smallvec::SmallVec;
 use std::{borrow::Cow, collections::hash_map::Entry, cmp, error, marker::PhantomData, mem, pin::Pin};
 use std::time::{Duration, Instant};
@@ -61,9 +60,9 @@ use tokio_io::{AsyncRead, AsyncWrite};
 /// Note that this "banning" system is not an actual ban. If a "banned" node tries to connect to
 /// us, we accept the connection. The "banning" system is only about delaying dialing attempts.
 ///
-pub struct LegacyProto<B: BlockT, TSubstream> {
+pub struct LegacyProto< TSubstream> {
 	/// List of protocols to open with peers. Never modified.
-	protocol: RegisteredProtocol<B>,
+	protocol: RegisteredProtocol,
 
 	/// Receiver for instructions about who to connect to or disconnect from.
 	peerset: peerset::Peerset,
@@ -80,7 +79,7 @@ pub struct LegacyProto<B: BlockT, TSubstream> {
 	next_incoming_index: peerset::IncomingIndex,
 
 	/// Events to produce from `poll()`.
-	events: SmallVec<[NetworkBehaviourAction<CustomProtoHandlerIn<B>, LegacyProtoOut<B>>; 4]>,
+	events: SmallVec<[NetworkBehaviourAction<CustomProtoHandlerIn, LegacyProtoOut>; 4]>,
 
 	/// Marker to pin the generics.
 	marker: PhantomData<TSubstream>,
@@ -189,7 +188,7 @@ struct IncomingPeer {
 
 /// Event that can be emitted by the `LegacyProto`.
 #[derive(Debug)]
-pub enum LegacyProtoOut<B: BlockT> {
+pub enum LegacyProtoOut {
 	/// Opened a custom protocol with the remote.
 	CustomProtocolOpen {
 		/// Version of the protocol that has been opened.
@@ -213,7 +212,7 @@ pub enum LegacyProtoOut<B: BlockT> {
 		/// Id of the peer the message came from.
 		peer_id: PeerId,
 		/// Message that has been received.
-		message: Message<B>,
+		message: BytesMut,
 	},
 
 	/// The substream used by the protocol is pretty large. We should print avoid sending more
@@ -222,11 +221,11 @@ pub enum LegacyProtoOut<B: BlockT> {
 		/// Id of the peer which is clogged.
 		peer_id: PeerId,
 		/// Copy of the messages that are within the buffer, for further diagnostic.
-		messages: Vec<Message<B>>,
+		messages: Vec<Vec<u8>>,
 	},
 }
 
-impl<B: BlockT, TSubstream> LegacyProto<B, TSubstream> {
+impl<TSubstream> LegacyProto<TSubstream> {
 	/// Creates a `CustomProtos`.
 	pub fn new(
 		protocol: impl Into<ProtocolId>,
@@ -350,8 +349,7 @@ impl<B: BlockT, TSubstream> LegacyProto<B, TSubstream> {
 	///
 	/// 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>)
-	where B: BlockT {
+	pub fn send_packet(&mut self, target: &PeerId, message: Vec<u8>) {
 		if !self.is_open(target) {
 			return;
 		}
@@ -607,7 +605,7 @@ impl<B: BlockT, TSubstream> LegacyProto<B, TSubstream> {
 	}
 }
 
-impl<B: BlockT, TSubstream> DiscoveryNetBehaviour for LegacyProto<B, TSubstream> {
+impl<TSubstream> DiscoveryNetBehaviour for LegacyProto<TSubstream> {
 	fn add_discovered_nodes(&mut self, peer_ids: impl Iterator<Item = PeerId>) {
 		self.peerset.discovered(peer_ids.into_iter().map(|peer_id| {
 			debug!(target: "sub-libp2p", "PSM <= Discovered({:?})", peer_id);
@@ -616,13 +614,12 @@ impl<B: BlockT, TSubstream> DiscoveryNetBehaviour for LegacyProto<B, TSubstream>
 	}
 }
 
-impl<B, TSubstream> NetworkBehaviour for LegacyProto<B, TSubstream>
+impl<TSubstream> NetworkBehaviour for LegacyProto<TSubstream>
 where
 	TSubstream: AsyncRead + AsyncWrite,
-	B: BlockT,
 {
-	type ProtocolsHandler = CustomProtoHandlerProto<B, TSubstream>;
-	type OutEvent = LegacyProtoOut<B>;
+	type ProtocolsHandler = CustomProtoHandlerProto<TSubstream>;
+	type OutEvent = LegacyProtoOut;
 
 	fn new_handler(&mut self) -> Self::ProtocolsHandler {
 		CustomProtoHandlerProto::new(self.protocol.clone())
@@ -825,7 +822,7 @@ where
 	fn inject_node_event(
 		&mut self,
 		source: PeerId,
-		event: CustomProtoHandlerOut<B>,
+		event: CustomProtoHandlerOut,
 	) {
 		match event {
 			CustomProtoHandlerOut::CustomProtocolClosed { reason } => {
@@ -954,7 +951,7 @@ where
 		_params: &mut impl PollParameters,
 	) -> Async<
 		NetworkBehaviourAction<
-			CustomProtoHandlerIn<B>,
+			CustomProtoHandlerIn,
 			Self::OutEvent,
 		>,
 	> {

core/network/src/legacy_proto/handler.rs

@@ -15,7 +15,7 @@
 // along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
 
 use crate::legacy_proto::upgrade::{RegisteredProtocol, RegisteredProtocolEvent, RegisteredProtocolSubstream};
-use crate::protocol::message::Message;
+use bytes::BytesMut;
 use futures::prelude::*;
 use futures03::{compat::Compat, TryFutureExt as _};
 use futures_timer::Delay;
@@ -29,7 +29,6 @@ use libp2p::swarm::{
 	SubstreamProtocol,
 };
 use log::{debug, error};
-use sr_primitives::traits::Block as BlockT;
 use smallvec::{smallvec, SmallVec};
 use std::{borrow::Cow, error, fmt, io, marker::PhantomData, mem, time::Duration};
 use tokio_io::{AsyncRead, AsyncWrite};
@@ -88,36 +87,34 @@ use tokio_io::{AsyncRead, AsyncWrite};
 /// We consider that we are now "closed" if the remote closes all the existing substreams.
 /// Re-opening it can then be performed by closing all active substream and re-opening one.
 ///
-pub struct CustomProtoHandlerProto<B, TSubstream> {
+pub struct CustomProtoHandlerProto<TSubstream> {
 	/// Configuration for the protocol upgrade to negotiate.
-	protocol: RegisteredProtocol<B>,
+	protocol: RegisteredProtocol,
 
 	/// Marker to pin the generic type.
 	marker: PhantomData<TSubstream>,
 }
 
-impl<B, TSubstream> CustomProtoHandlerProto<B, TSubstream>
+impl<TSubstream> CustomProtoHandlerProto<TSubstream>
 where
 	TSubstream: AsyncRead + AsyncWrite,
-	B: BlockT,
 {
 	/// Builds a new `CustomProtoHandlerProto`.
-	pub fn new(protocol: RegisteredProtocol<B>) -> Self {
+	pub fn new(protocol: RegisteredProtocol) -> Self {
 		CustomProtoHandlerProto {
 			protocol,
 			marker: PhantomData,
 		}
 	}
 }
 
-impl<B, TSubstream> IntoProtocolsHandler for CustomProtoHandlerProto<B, TSubstream>
+impl<TSubstream> IntoProtocolsHandler for CustomProtoHandlerProto<TSubstream>
 where
 	TSubstream: AsyncRead + AsyncWrite,
-	B: BlockT,
 {
-	type Handler = CustomProtoHandler<B, TSubstream>;
+	type Handler = CustomProtoHandler<TSubstream>;
 
-	fn inbound_protocol(&self) -> RegisteredProtocol<B> {
+	fn inbound_protocol(&self) -> RegisteredProtocol {
 		self.protocol.clone()
 	}
 
@@ -136,12 +133,12 @@ where
 }
 
 /// The actual handler once the connection has been established.
-pub struct CustomProtoHandler<B: BlockT, TSubstream> {
+pub struct CustomProtoHandler<TSubstream> {
 	/// Configuration for the protocol upgrade to negotiate.
-	protocol: RegisteredProtocol<B>,
+	protocol: RegisteredProtocol,
 
 	/// State of the communications with the remote.
-	state: ProtocolState<B, TSubstream>,
+	state: ProtocolState<TSubstream>,
 
 	/// Identifier of the node we're talking to. Used only for logging purposes and shouldn't have
 	/// any influence on the behaviour.
@@ -155,15 +152,15 @@ pub struct CustomProtoHandler<B: BlockT, TSubstream> {
 	///
 	/// This queue must only ever be modified to insert elements at the back, or remove the first
 	/// element.
-	events_queue: SmallVec<[ProtocolsHandlerEvent<RegisteredProtocol<B>, (), CustomProtoHandlerOut<B>>; 16]>,
+	events_queue: SmallVec<[ProtocolsHandlerEvent<RegisteredProtocol, (), CustomProtoHandlerOut>; 16]>,
 }
 
 /// State of the handler.
-enum ProtocolState<B, TSubstream> {
+enum ProtocolState<TSubstream> {
 	/// Waiting for the behaviour to tell the handler whether it is enabled or disabled.
 	Init {
 		/// List of substreams opened by the remote but that haven't been processed yet.
-		substreams: SmallVec<[RegisteredProtocolSubstream<B, TSubstream>; 6]>,
+		substreams: SmallVec<[RegisteredProtocolSubstream<TSubstream>; 6]>,
 		/// Deadline after which the initialization is abnormally long.
 		init_deadline: Compat<Delay>,
 	},
@@ -179,17 +176,17 @@ enum ProtocolState<B, TSubstream> {
 	/// If we are in this state, we have sent a `CustomProtocolOpen` message to the outside.
 	Normal {
 		/// The substreams where bidirectional communications happen.
-		substreams: SmallVec<[RegisteredProtocolSubstream<B, TSubstream>; 4]>,
+		substreams: SmallVec<[RegisteredProtocolSubstream<TSubstream>; 4]>,
 		/// Contains substreams which are being shut down.
-		shutdown: SmallVec<[RegisteredProtocolSubstream<B, TSubstream>; 4]>,
+		shutdown: SmallVec<[RegisteredProtocolSubstream<TSubstream>; 4]>,
 	},
 
 	/// We are disabled. Contains substreams that are being closed.
 	/// If we are in this state, either we have sent a `CustomProtocolClosed` message to the
 	/// outside or we have never sent any `CustomProtocolOpen` in the first place.
 	Disabled {
 		/// List of substreams to shut down.
-		shutdown: SmallVec<[RegisteredProtocolSubstream<B, TSubstream>; 6]>,
+		shutdown: SmallVec<[RegisteredProtocolSubstream<TSubstream>; 6]>,
 
 		/// If true, we should reactivate the handler after all the substreams in `shutdown` have
 		/// been closed.
@@ -210,7 +207,7 @@ enum ProtocolState<B, TSubstream> {
 
 /// Event that can be received by a `CustomProtoHandler`.
 #[derive(Debug)]
-pub enum CustomProtoHandlerIn<B: BlockT> {
+pub enum CustomProtoHandlerIn {
 	/// The node should start using custom protocols.
 	Enable,
 
@@ -220,13 +217,13 @@ pub enum CustomProtoHandlerIn<B: BlockT> {
 	/// Sends a message through a custom protocol substream.
 	SendCustomMessage {
 		/// The message to send.
-		message: Message<B>,
+		message: Vec<u8>,
 	},
 }
 
 /// Event that can be emitted by a `CustomProtoHandler`.
 #[derive(Debug)]
-pub enum CustomProtoHandlerOut<B: BlockT> {
+pub enum CustomProtoHandlerOut {
 	/// Opened a custom protocol with the remote.
 	CustomProtocolOpen {
 		/// Version of the protocol that has been opened.
@@ -242,14 +239,14 @@ pub enum CustomProtoHandlerOut<B: BlockT> {
 	/// Receives a message on a custom protocol substream.
 	CustomMessage {
 		/// Message that has been received.
-		message: Message<B>,
+		message: BytesMut,
 	},
 
 	/// A substream to the remote is clogged. The send buffer is very large, and we should print
 	/// a diagnostic message and/or avoid sending more data.
 	Clogged {
 		/// Copy of the messages that are within the buffer, for further diagnostic.
-		messages: Vec<Message<B>>,
+		messages: Vec<Vec<u8>>,
 	},
 
 	/// An error has happened on the protocol level with this node.
@@ -261,10 +258,9 @@ pub enum CustomProtoHandlerOut<B: BlockT> {
 	},
 }
 
-impl<B, TSubstream> CustomProtoHandler<B, TSubstream>
+impl<TSubstream> CustomProtoHandler<TSubstream>
 where
 	TSubstream: AsyncRead + AsyncWrite,
-	B: BlockT,
 {
 	/// Enables the handler.
 	fn enable(&mut self) {
@@ -342,7 +338,7 @@ where
 	/// Polls the state for events. Optionally returns an event to produce.
 	#[must_use]
 	fn poll_state(&mut self)
-		-> Option<ProtocolsHandlerEvent<RegisteredProtocol<B>, (), CustomProtoHandlerOut<B>>> {
+		-> Option<ProtocolsHandlerEvent<RegisteredProtocol, (), CustomProtoHandlerOut>> {
 		match mem::replace(&mut self.state, ProtocolState::Poisoned) {
 			ProtocolState::Poisoned => {
 				error!(target: "sub-libp2p", "Handler with {:?} is in poisoned state",
@@ -471,7 +467,7 @@ where
 	/// Called by `inject_fully_negotiated_inbound` and `inject_fully_negotiated_outbound`.
 	fn inject_fully_negotiated(
 		&mut self,
-		mut substream: RegisteredProtocolSubstream<B, TSubstream>
+		mut substream: RegisteredProtocolSubstream<TSubstream>
 	) {
 		self.state = match mem::replace(&mut self.state, ProtocolState::Poisoned) {
 			ProtocolState::Poisoned => {
@@ -516,7 +512,7 @@ where
 	}
 
 	/// Sends a message to the remote.
-	fn send_message(&mut self, message: Message<B>) {
+	fn send_message(&mut self, message: Vec<u8>) {
 		match self.state {
 			ProtocolState::Normal { ref mut substreams, .. } =>
 				substreams[0].send_message(message),
@@ -527,14 +523,14 @@ where
 	}
 }
 
-impl<B, TSubstream> ProtocolsHandler for CustomProtoHandler<B, TSubstream>
-where TSubstream: AsyncRead + AsyncWrite, B: BlockT {
-	type InEvent = CustomProtoHandlerIn<B>;
-	type OutEvent = CustomProtoHandlerOut<B>;
+impl<TSubstream> ProtocolsHandler for CustomProtoHandler<TSubstream>
+where TSubstream: AsyncRead + AsyncWrite {
+	type InEvent = CustomProtoHandlerIn;
+	type OutEvent = CustomProtoHandlerOut;
 	type Substream = TSubstream;
 	type Error = ConnectionKillError;
-	type InboundProtocol = RegisteredProtocol<B>;
-	type OutboundProtocol = RegisteredProtocol<B>;
+	type InboundProtocol = RegisteredProtocol;
+	type OutboundProtocol = RegisteredProtocol;
 	type OutboundOpenInfo = ();
 
 	fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol> {
@@ -556,7 +552,7 @@ where TSubstream: AsyncRead + AsyncWrite, B: BlockT {
 		self.inject_fully_negotiated(proto);
 	}
 
-	fn inject_event(&mut self, message: CustomProtoHandlerIn<B>) {
+	fn inject_event(&mut self, message: CustomProtoHandlerIn) {
 		match message {
 			CustomProtoHandlerIn::Disable => self.disable(),
 			CustomProtoHandlerIn::Enable => self.enable(),
@@ -613,7 +609,7 @@ where TSubstream: AsyncRead + AsyncWrite, B: BlockT {
 	}
 }
 
-impl<B: BlockT, TSubstream> fmt::Debug for CustomProtoHandler<B, TSubstream>
+impl<TSubstream> fmt::Debug for CustomProtoHandler<TSubstream>
 where
 	TSubstream: AsyncRead + AsyncWrite,
 {
@@ -625,9 +621,9 @@ where
 
 /// Given a list of substreams, tries to shut them down. The substreams that have been successfully
 /// shut down are removed from the list.
-fn shutdown_list<B, TSubstream>
-	(list: &mut SmallVec<impl smallvec::Array<Item = RegisteredProtocolSubstream<B, TSubstream>>>)
-where TSubstream: AsyncRead + AsyncWrite, B: BlockT {
+fn shutdown_list<TSubstream>
+	(list: &mut SmallVec<impl smallvec::Array<Item = RegisteredProtocolSubstream<TSubstream>>>)
+where TSubstream: AsyncRead + AsyncWrite {
 	'outer: for n in (0..list.len()).rev() {
 		let mut substream = list.swap_remove(n);
 		loop {