Rewrite the libp2p networking

.cargo/config

@@ -0,0 +1 @@
+#paths = ["../rust-libp2p"]

core/cli/src/lib.rs

@@ -58,7 +58,7 @@ pub mod error;
 pub mod informant;
 mod panic_hook;
 
-use network_libp2p::AddrComponent;
+use network_libp2p::Protocol;
 use runtime_primitives::traits::As;
 use service::{
 	ServiceFactory, FactoryFullConfiguration, RuntimeGenesis,
@@ -305,8 +305,8 @@ where
 				None => 30333,
 			};
 			config.network.listen_addresses = vec![
-				iter::once(AddrComponent::IP4(Ipv4Addr::new(0, 0, 0, 0)))
-					.chain(iter::once(AddrComponent::TCP(port)))
+				iter::once(Protocol::Ip4(Ipv4Addr::new(0, 0, 0, 0)))
+					.chain(iter::once(Protocol::Tcp(port)))
 					.collect()
 			];
 		}

core/network-libp2p/Cargo.toml

@@ -11,7 +11,7 @@ bytes = "0.4"
 error-chain = { version = "0.12", default-features = false }
 fnv = "1.0"
 futures = "0.1"
-libp2p = { git = "https://github.com/libp2p/rust-libp2p", rev = "5980a4538ef6fc8af450893acb01290eaed136de", default-features = false, features = ["libp2p-secio", "libp2p-secio-secp256k1"] }
+libp2p = { git = "https://github.com/tomaka/libp2p-rs", rev = "8111062f0177fd7423626f2db9560273644a4c4d", default-features = false, features = ["libp2p-secio", "libp2p-secio-secp256k1"] }
 ethereum-types = "0.3"
 parking_lot = "0.5"
 libc = "0.2"
@@ -20,7 +20,9 @@ rand = "0.5.0"
 serde = "1.0.70"
 serde_derive = "1.0.70"
 serde_json = "1.0.24"
+smallvec = "0.6.5"
 tokio = "0.1"
+tokio-executor = "0.1"
 tokio-io = "0.1"
 tokio-timer = "0.2"
 unsigned-varint = { version = "0.2.1", features = ["codec"] }

core/network-libp2p/src/custom_proto.rs

@@ -72,7 +72,7 @@ impl<T> RegisteredProtocol<T> {
 	/// passed inside the `RegisteredProtocolOutput`.
 	pub fn new(custom_data: T, protocol: ProtocolId, versions: &[(u8, u8)])
 		-> Self {
-		let mut proto_name = Bytes::from_static(b"/core/");
+		let mut proto_name = Bytes::from_static(b"/substrate/");
 		proto_name.extend_from_slice(&protocol);
 		proto_name.extend_from_slice(b"/");
 
@@ -100,9 +100,8 @@ impl<T> RegisteredProtocol<T> {
 }
 
 // `Maf` is short for `MultiaddressFuture`
-impl<T, C, Maf> ConnectionUpgrade<C, Maf> for RegisteredProtocol<T>
+impl<T, C> ConnectionUpgrade<C> for RegisteredProtocol<T>
 where C: AsyncRead + AsyncWrite + Send + 'static,		// TODO: 'static :-/
-	Maf: Future<Item = Multiaddr, Error = IoError> + Send + 'static,		// TODO: 'static :(
 {
 	type NamesIter = VecIntoIter<(Bytes, Self::UpgradeIdentifier)>;
 	type UpgradeIdentifier = u8;		// Protocol version
@@ -119,16 +118,15 @@ where C: AsyncRead + AsyncWrite + Send + 'static,		// TODO: 'static :-/
 	}
 
 	type Output = RegisteredProtocolOutput<T>;
-	type MultiaddrFuture = Maf;
-	type Future = future::FutureResult<(Self::Output, Self::MultiaddrFuture), IoError>;
+	type Future = future::FutureResult<Self::Output, IoError>;
 
 	#[allow(deprecated)]
 	fn upgrade(
 		self,
 		socket: C,
 		protocol_version: Self::UpgradeIdentifier,
 		endpoint: Endpoint,
-		remote_addr: Maf
+		_: &Multiaddr
 	) -> Self::Future {
 		let packet_count = self.supported_versions
 			.iter()
@@ -224,7 +222,7 @@ where C: AsyncRead + AsyncWrite + Send + 'static,		// TODO: 'static :-/
 			incoming: Box::new(incoming),
 		};
 
-		future::ok((out, remote_addr))
+		future::ok(out)
 	}
 }
 
@@ -233,6 +231,12 @@ where C: AsyncRead + AsyncWrite + Send + 'static,		// TODO: 'static :-/
 pub struct RegisteredProtocols<T>(pub Vec<RegisteredProtocol<T>>);
 
 impl<T> RegisteredProtocols<T> {
+	/// Returns the number of protocols.
+	#[inline]
+	pub fn len(&self) -> usize {
+		self.0.len()
+	}
+
 	/// Finds a protocol in the list by its id.
 	pub fn find_protocol(&self, protocol: ProtocolId)
 		-> Option<&RegisteredProtocol<T>> {
@@ -251,35 +255,32 @@ impl<T> Default for RegisteredProtocols<T> {
 	}
 }
 
-impl<T, C, Maf> ConnectionUpgrade<C, Maf> for RegisteredProtocols<T>
+impl<T, C> ConnectionUpgrade<C> for RegisteredProtocols<T>
 where C: AsyncRead + AsyncWrite + Send + 'static,		// TODO: 'static :-/
-	Maf: Future<Item = Multiaddr, Error = IoError> + Send + 'static,		// TODO: 'static :(
 {
 	type NamesIter = VecIntoIter<(Bytes, Self::UpgradeIdentifier)>;
 	type UpgradeIdentifier = (usize,
-		<RegisteredProtocol<T> as ConnectionUpgrade<C, Maf>>::UpgradeIdentifier);
+		<RegisteredProtocol<T> as ConnectionUpgrade<C>>::UpgradeIdentifier);
 
 	fn protocol_names(&self) -> Self::NamesIter {
 		// We concat the lists of `RegisteredProtocol::protocol_names` for
 		// each protocol.
 		self.0.iter().enumerate().flat_map(|(n, proto)|
-			ConnectionUpgrade::<C, Maf>::protocol_names(proto)
+			ConnectionUpgrade::<C>::protocol_names(proto)
 				.map(move |(name, id)| (name, (n, id)))
 		).collect::<Vec<_>>().into_iter()
 	}
 
-	type Output = <RegisteredProtocol<T> as ConnectionUpgrade<C, Maf>>::Output;
-	type MultiaddrFuture = <RegisteredProtocol<T> as
-		ConnectionUpgrade<C, Maf>>::MultiaddrFuture;
-	type Future = <RegisteredProtocol<T> as ConnectionUpgrade<C, Maf>>::Future;
+	type Output = <RegisteredProtocol<T> as ConnectionUpgrade<C>>::Output;
+	type Future = <RegisteredProtocol<T> as ConnectionUpgrade<C>>::Future;
 
 	#[inline]
 	fn upgrade(
 		self,
 		socket: C,
 		upgrade_identifier: Self::UpgradeIdentifier,
 		endpoint: Endpoint,
-		remote_addr: Maf
+		remote_addr: &Multiaddr
 	) -> Self::Future {
 		let (protocol_index, inner_proto_id) = upgrade_identifier;
 		self.0.into_iter()

core/network-libp2p/src/lib.rs

@@ -25,15 +25,18 @@ extern crate parking_lot;
 extern crate fnv;
 extern crate futures;
 extern crate tokio;
+extern crate tokio_executor;
 extern crate tokio_io;
 extern crate tokio_timer;
 extern crate libc;
+#[macro_use]
 extern crate libp2p;
 extern crate rand;
 extern crate serde;
 #[macro_use]
 extern crate serde_derive;
 extern crate serde_json;
+extern crate smallvec;
 extern crate bytes;
 extern crate unsigned_varint;
 
@@ -46,18 +49,25 @@ extern crate log;
 #[cfg(test)] #[macro_use]
 extern crate assert_matches;
 
+use libp2p::PeerId;
+
 pub use connection_filter::{ConnectionFilter, ConnectionDirection};
 pub use error::{Error, ErrorKind, DisconnectReason};
-pub use libp2p::{Multiaddr, multiaddr::AddrComponent};
+pub use libp2p::{Multiaddr, multiaddr::Protocol};
 pub use traits::*;
 
 pub type TimerToken = usize;
 
+// TODO: remove as it is unused ; however modifying `network` causes a clusterfuck of dependencies
+// resolve errors at the moment
 mod connection_filter;
 mod custom_proto;
 mod error;
-mod network_state;
+mod node_handler;
+mod secret;
 mod service;
+mod service_task;
+mod swarm;
 mod timeouts;
 mod topology;
 mod traits;
@@ -67,8 +77,19 @@ pub use service::NetworkService;
 
 /// Check if node url is valid
 pub fn validate_node_url(url: &str) -> Result<(), Error> {
-	match url.parse::<libp2p::multiaddr::Multiaddr>() {
+	match url.parse::<Multiaddr>() {
 		Ok(_) => Ok(()),
 		Err(_) => Err(ErrorKind::InvalidNodeId.into()),
 	}
 }
+
+/// Parses a string address and returns the component, if valid.
+pub(crate) fn parse_str_addr(addr_str: &str) -> Result<(PeerId, Multiaddr), Error> {
+	let mut addr: Multiaddr = addr_str.parse().map_err(|_| ErrorKind::AddressParse)?;
+	let who = match addr.pop() {
+		Some(Protocol::P2p(key)) =>
+			PeerId::from_multihash(key).map_err(|_| ErrorKind::AddressParse)?,
+		_ => return Err(ErrorKind::AddressParse.into()),
+	};
+	Ok((who, addr))
+}