Use substrate codec for network messages (#333)

* Use substrate codec for network messages

* Version bump

* Removed redundant format

substrate/polkadot/network/src/collator_pool.rs

@@ -18,6 +18,7 @@
 
 use polkadot_primitives::{AccountId, Hash};
 use polkadot_primitives::parachain::{Id as ParaId, Collation};
+use codec;
 
 use futures::sync::oneshot;
 
@@ -27,12 +28,28 @@ use std::time::{Duration, Instant};
 const COLLATION_LIFETIME: Duration = Duration::from_secs(60 * 5);
 
 /// The role of the collator. Whether they're the primary or backup for this parachain.
-#[derive(PartialEq, Debug, Serialize, Deserialize)]
+#[derive(PartialEq, Debug, Clone, Copy)]
 pub enum Role {
 	/// Primary collators should send collations whenever it's time.
-	Primary,
+	Primary = 0,
 	/// Backup collators should not.
-	Backup,
+	Backup = 1,
+}
+
+impl codec::Encode for Role {
+	fn encode_to<T: codec::Output>(&self, dest: &mut T) {
+		dest.push_byte(*self as u8);
+	}
+}
+
+impl codec::Decode for Role {
+	fn decode<I: codec::Input>(input: &mut I) -> Option<Self> {
+		match input.read_byte()? {
+			x if x == Role::Primary as u8 => Some(Role::Primary),
+			x if x == Role::Backup as u8 => Some(Role::Backup),
+			_ => None,
+		}
+	}
 }
 
 /// A maintenance action for the collator set.

substrate/polkadot/network/src/consensus.rs

@@ -26,6 +26,7 @@ use polkadot_api::{PolkadotApi, LocalPolkadotApi};
 use polkadot_consensus::{Network, SharedTable, Collators};
 use polkadot_primitives::{AccountId, Block, Hash, SessionKey};
 use polkadot_primitives::parachain::{Id as ParaId, Collation};
+use codec::Decode;
 
 use futures::prelude::*;
 use futures::sync::mpsc;
@@ -175,7 +176,7 @@ impl<P: LocalPolkadotApi + Send + Sync + 'static> MessageProcessTask<P> {
 				}
 			}
 			ConsensusMessage::ChainSpecific(msg, _) => {
-				if let Ok(Message::Statement(parent_hash, statement)) = ::serde_json::from_slice(&msg) {
+				if let Some(Message::Statement(parent_hash, statement)) = Decode::decode(&mut msg.as_slice()) {
 					if ::polkadot_consensus::check_statement(&statement.statement, &statement.signature, statement.sender, &parent_hash) {
 						self.table_router.import_statement(statement);
 					}

substrate/polkadot/network/src/lib.rs

@@ -20,11 +20,6 @@
 //! parachain block and extrinsic data fetching, communication between collators and validators,
 //! and more.
 
-extern crate serde;
-#[macro_use]
-extern crate serde_derive;
-extern crate serde_json;
-
 extern crate substrate_bft as bft;
 extern crate substrate_codec as codec;
 extern crate substrate_network;
@@ -47,7 +42,7 @@ mod collator_pool;
 mod router;
 pub mod consensus;
 
-use codec::{Decode, Encode};
+use codec::{Decode, Encode, Input, Output};
 use futures::sync::oneshot;
 use parking_lot::Mutex;
 use polkadot_consensus::{Statement, SignedStatement, GenericStatement};
@@ -188,7 +183,6 @@ impl CurrentConsensus {
 }
 
 /// Polkadot-specific messages.
-#[derive(Serialize, Deserialize)]
 pub enum Message {
 	/// signed statement and localized parent hash.
 	Statement(Hash, SignedStatement),
@@ -205,8 +199,58 @@ pub enum Message {
 	Collation(Hash, Collation),
 }
 
+impl Encode for Message {
+	fn encode_to<T: Output>(&self, dest: &mut T) {
+		match *self {
+			Message::Statement(ref h, ref s) => {
+				dest.push_byte(0);
+				dest.push(h);
+				dest.push(s);
+			}
+			Message::SessionKey(ref h, ref k) => {
+				dest.push_byte(1);
+				dest.push(h);
+				dest.push(k);
+			}
+			Message::RequestBlockData(ref id, ref d) => {
+				dest.push_byte(2);
+				dest.push(id);
+				dest.push(d);
+			}
+			Message::BlockData(ref id, ref d) => {
+				dest.push_byte(3);
+				dest.push(id);
+				dest.push(d);
+			}
+			Message::CollatorRole(ref r) => {
+				dest.push_byte(4);
+				dest.push(r);
+			}
+			Message::Collation(ref h, ref c) => {
+				dest.push_byte(5);
+				dest.push(h);
+				dest.push(c);
+			}
+		}
+	}
+}
+
+impl Decode for Message {
+	fn decode<I: Input>(input: &mut I) -> Option<Self> {
+		match input.read_byte()? {
+			0 => Some(Message::Statement(Decode::decode(input)?, Decode::decode(input)?)),
+			1 => Some(Message::SessionKey(Decode::decode(input)?, Decode::decode(input)?)),
+			2 => Some(Message::RequestBlockData(Decode::decode(input)?, Decode::decode(input)?)),
+			3 => Some(Message::BlockData(Decode::decode(input)?, Decode::decode(input)?)),
+			4 => Some(Message::CollatorRole(Decode::decode(input)?)),
+			5 => Some(Message::Collation(Decode::decode(input)?, Decode::decode(input)?)),
+			_ => None,
+		}
+	}
+}
+
 fn send_polkadot_message(ctx: &mut Context<Block>, to: PeerId, message: Message) {
-	let encoded = ::serde_json::to_vec(&message).expect("serialization of messages infallible; qed");
+	let encoded = message.encode();
 	ctx.send_message(to, generic_message::Message::ChainSpecific(encoded))
 }
 
@@ -244,9 +288,7 @@ impl PolkadotProtocol {
 	/// Send a statement to a validator.
 	fn send_statement(&mut self, ctx: &mut Context<Block>, _val: SessionKey, parent_hash: Hash, statement: SignedStatement) {
 		// TODO: something more targeted than gossip.
-		let raw = ::serde_json::to_vec(&Message::Statement(parent_hash, statement))
-			.expect("message serialization infallible; qed");
-
+		let raw = Message::Statement(parent_hash, statement).encode();
 		self.consensus_gossip.multicast_chain_specific(ctx, raw, parent_hash);
 	}
 
@@ -427,7 +469,7 @@ impl Specialization<Block> for PolkadotProtocol {
 			);
 		}
 
-		let validator = status.roles.iter().any(|r| *r == message::Role::Authority);
+		let validator = status.roles.contains(substrate_network::Roles::AUTHORITY);
 		let send_key = validator || local_status.collating_for.is_some();
 
 		self.peers.insert(peer_id, PeerInfo {
@@ -436,7 +478,7 @@ impl Specialization<Block> for PolkadotProtocol {
 			validator,
 		});
 
-		self.consensus_gossip.new_peer(ctx, peer_id, &status.roles);
+		self.consensus_gossip.new_peer(ctx, peer_id, status.roles);
 		if let (true, &Some(ref consensus)) = (send_key, &self.live_consensus) {
 			send_polkadot_message(
 				ctx,
@@ -497,11 +539,11 @@ impl Specialization<Block> for PolkadotProtocol {
 				self.consensus_gossip.on_bft_message(ctx, peer_id, msg)
 			}
 			generic_message::Message::ChainSpecific(raw) => {
-				match serde_json::from_slice(&raw) {
-					Ok(msg) => self.on_polkadot_message(ctx, peer_id, raw, msg),
-					Err(e) => {
-						trace!(target: "p_net", "Bad message from {}: {}", peer_id, e);
-						ctx.disable_peer(peer_id, "Unknown Polkadot-protocol reason");
+				match Message::decode(&mut raw.as_slice()) {
+					Some(msg) => self.on_polkadot_message(ctx, peer_id, raw, msg),
+					None => {
+						trace!(target: "p_net", "Bad message from {}", peer_id);
+						ctx.disable_peer(peer_id, "Invalid polkadot protocol message format");
 					}
 				}
 			}

substrate/polkadot/network/src/tests.rs

@@ -24,7 +24,7 @@ use polkadot_primitives::{Block, Hash, SessionKey};
 use polkadot_primitives::parachain::{CandidateReceipt, HeadData, BlockData};
 use substrate_primitives::H512;
 use codec::Encode;
-use substrate_network::{PeerId, PeerInfo, ClientHandle, Context, message::Message as SubstrateMessage, message::Role, specialization::Specialization, generic_message::Message as GenericMessage};
+use substrate_network::{PeerId, PeerInfo, ClientHandle, Context, Roles, message::Message as SubstrateMessage, specialization::Specialization, generic_message::Message as GenericMessage};
 
 use std::sync::Arc;
 use futures::Future;
@@ -62,7 +62,7 @@ impl TestContext {
 	fn has_message(&self, to: PeerId, message: Message) -> bool {
 		use substrate_network::generic_message::Message as GenericMessage;
 
-		let encoded = ::serde_json::to_vec(&message).unwrap();
+		let encoded = message.encode();
 		self.messages.iter().any(|&(ref peer, ref msg)| match msg {
 			GenericMessage::ChainSpecific(ref data) => peer == &to && data == &encoded,
 			_ => false,
@@ -70,7 +70,7 @@ impl TestContext {
 	}
 }
 
-fn make_status(status: &Status, roles: Vec<Role>) -> FullStatus {
+fn make_status(status: &Status, roles: Roles) -> FullStatus {
 	FullStatus {
 		version: 1,
 		roles,
@@ -78,9 +78,6 @@ fn make_status(status: &Status, roles: Vec<Role>) -> FullStatus {
 		best_hash: Default::default(),
 		genesis_hash: Default::default(),
 		chain_status: status.encode(),
-		parachain_id: None,
-		validator_id: None,
-		validator_signature: None,
 	}
 }
 
@@ -97,7 +94,7 @@ fn make_consensus(parent_hash: Hash, local_key: SessionKey) -> (CurrentConsensus
 }
 
 fn on_message(protocol: &mut PolkadotProtocol, ctx: &mut TestContext, from: PeerId, message: Message) {
-	let encoded = ::serde_json::to_vec(&message).unwrap();
+	let encoded = message.encode();
 	protocol.on_message(ctx, from, GenericMessage::ChainSpecific(encoded));
 }
 
@@ -115,7 +112,7 @@ fn sends_session_key() {
 
 	{
 		let mut ctx = TestContext::default();
-		protocol.on_connect(&mut ctx, peer_a, make_status(&validator_status, vec![Role::Authority]));
+		protocol.on_connect(&mut ctx, peer_a, make_status(&validator_status, Roles::AUTHORITY));
 		assert!(ctx.messages.is_empty());
 	}
 
@@ -129,7 +126,7 @@ fn sends_session_key() {
 
 	{
 		let mut ctx = TestContext::default();
-		protocol.on_connect(&mut ctx, peer_b, make_status(&collator_status, vec![]));
+		protocol.on_connect(&mut ctx, peer_b, make_status(&collator_status, Roles::NONE));
 		assert!(ctx.has_message(peer_b, Message::SessionKey(parent_hash, local_key)));
 	}
 }
@@ -171,7 +168,7 @@ fn fetches_from_those_with_knowledge() {
 	// connect peer A
 	{
 		let mut ctx = TestContext::default();
-		protocol.on_connect(&mut ctx, peer_a, make_status(&status, vec![Role::Authority]));
+		protocol.on_connect(&mut ctx, peer_a, make_status(&status, Roles::AUTHORITY));
 		assert!(ctx.has_message(peer_a, Message::SessionKey(parent_hash, local_key)));
 	}
 
@@ -187,7 +184,7 @@ fn fetches_from_those_with_knowledge() {
 	// peer B connects and sends session key. request already assigned to A
 	{
 		let mut ctx = TestContext::default();
-		protocol.on_connect(&mut ctx, peer_b, make_status(&status, vec![Role::Authority]));
+		protocol.on_connect(&mut ctx, peer_b, make_status(&status, Roles::AUTHORITY));
 		on_message(&mut protocol, &mut ctx, peer_b, Message::SessionKey(parent_hash, b_key));
 		assert!(!ctx.has_message(peer_b, Message::RequestBlockData(2, candidate_hash)));
 
@@ -220,7 +217,7 @@ fn remove_bad_collator() {
 
 	{
 		let mut ctx = TestContext::default();
-		protocol.on_connect(&mut ctx, peer_id, make_status(&status, vec![]));
+		protocol.on_connect(&mut ctx, peer_id, make_status(&status, Roles::NONE));
 	}
 
 	{