Add a KademliaHandler

core/src/peer_id.rs

@@ -147,6 +147,13 @@ impl PartialEq<PeerId> for multihash::Multihash {
     }
 }
 
+impl AsRef<multihash::Multihash> for PeerId {
+    #[inline]
+    fn as_ref(&self) -> &multihash::Multihash {
+        &self.multihash
+    }
+}
+
 impl Into<multihash::Multihash> for PeerId {
     #[inline]
     fn into(self) -> multihash::Multihash {

core/src/swarm.rs

@@ -37,12 +37,12 @@ pub use crate::nodes::raw_swarm::ConnectedPoint;
 /// Contains the state of the network, plus the way it should behave.
 pub struct Swarm<TTransport, TBehaviour, TTopology>
 where TTransport: Transport,
-      TBehaviour: NetworkBehaviour,
+      TBehaviour: NetworkBehaviour<TTopology>,
 {
     raw_swarm: RawSwarm<
         TTransport,
-        <<TBehaviour as NetworkBehaviour>::ProtocolsHandler as ProtocolsHandler>::InEvent,
-        <<TBehaviour as NetworkBehaviour>::ProtocolsHandler as ProtocolsHandler>::OutEvent,
+        <<TBehaviour as NetworkBehaviour<TTopology>>::ProtocolsHandler as ProtocolsHandler>::InEvent,
+        <<TBehaviour as NetworkBehaviour<TTopology>>::ProtocolsHandler as ProtocolsHandler>::OutEvent,
         NodeHandlerWrapper<TBehaviour::ProtocolsHandler>,
     >,
 
@@ -57,7 +57,7 @@ where TTransport: Transport,
 
 impl<TTransport, TBehaviour, TTopology> Deref for Swarm<TTransport, TBehaviour, TTopology>
 where TTransport: Transport,
-      TBehaviour: NetworkBehaviour,
+      TBehaviour: NetworkBehaviour<TTopology>,
 {
     type Target = TBehaviour;
 
@@ -69,7 +69,7 @@ where TTransport: Transport,
 
 impl<TTransport, TBehaviour, TTopology> DerefMut for Swarm<TTransport, TBehaviour, TTopology>
 where TTransport: Transport,
-      TBehaviour: NetworkBehaviour,
+      TBehaviour: NetworkBehaviour<TTopology>,
 {
     #[inline]
     fn deref_mut(&mut self) -> &mut Self::Target {
@@ -78,7 +78,7 @@ where TTransport: Transport,
 }
 
 impl<TTransport, TBehaviour, TMuxer, TTopology> Swarm<TTransport, TBehaviour, TTopology>
-where TBehaviour: NetworkBehaviour,
+where TBehaviour: NetworkBehaviour<TTopology>,
       TMuxer: StreamMuxer + Send + Sync + 'static,
       <TMuxer as StreamMuxer>::OutboundSubstream: Send + 'static,
       <TMuxer as StreamMuxer>::Substream: Send + 'static,
@@ -171,7 +171,7 @@ where TBehaviour: NetworkBehaviour,
 }
 
 impl<TTransport, TBehaviour, TMuxer, TTopology> Stream for Swarm<TTransport, TBehaviour, TTopology>
-where TBehaviour: NetworkBehaviour,
+where TBehaviour: NetworkBehaviour<TTopology>,
       TMuxer: StreamMuxer + Send + Sync + 'static,
       <TMuxer as StreamMuxer>::OutboundSubstream: Send + 'static,
       <TMuxer as StreamMuxer>::Substream: Send + 'static,
@@ -230,7 +230,7 @@ where TBehaviour: NetworkBehaviour,
                 Async::Ready(RawSwarmEvent::UnknownPeerDialError { .. }) => {},
             }
 
-            match self.behaviour.poll() {
+            match self.behaviour.poll(&mut self.topology) {
                 Async::NotReady if raw_swarm_not_ready => return Ok(Async::NotReady),
                 Async::NotReady => (),
                 Async::Ready(NetworkBehaviourAction::GenerateEvent(event)) => {
@@ -256,7 +256,7 @@ where TBehaviour: NetworkBehaviour,
 ///
 /// This trait has been designed to be composable. Multiple implementations can be combined into
 /// one that handles all the behaviours at once.
-pub trait NetworkBehaviour {
+pub trait NetworkBehaviour<TTopology> {
     /// Handler for all the protocols the network supports.
     type ProtocolsHandler: ProtocolsHandler;
     /// Event generated by the swarm.
@@ -286,7 +286,7 @@ pub trait NetworkBehaviour {
     /// Polls for things that swarm should do.
     ///
     /// This API mimics the API of the `Stream` trait.
-    fn poll(&mut self) -> Async<NetworkBehaviourAction<<Self::ProtocolsHandler as ProtocolsHandler>::InEvent, Self::OutEvent>>;
+    fn poll(&mut self, topology: &mut TTopology) -> Async<NetworkBehaviourAction<<Self::ProtocolsHandler as ProtocolsHandler>::InEvent, Self::OutEvent>>;
 }
 
 /// Action to perform.

core/src/topology/mod.rs

@@ -23,8 +23,6 @@ use {Multiaddr, PeerId};
 
 /// Storage for the network topology.
 pub trait Topology {
-    /// Adds a discovered address to the topology.
-    fn add_discovered_address(&mut self, peer: &PeerId, addr: Multiaddr);
     /// Returns the addresses to try use to reach the given peer.
     fn addresses_of_peer(&mut self, peer: &PeerId) -> Vec<Multiaddr>;
 }
@@ -42,6 +40,30 @@ impl MemoryTopology {
             list: Default::default()
         }
     }
+
+    /// Returns true if the topology is empty.
+    #[inline]
+    pub fn is_empty(&self) -> bool {
+        self.list.is_empty()
+    }
+
+    /// Adds an address to the topology.
+    #[inline]
+    pub fn add_address(&mut self, peer: PeerId, addr: Multiaddr) {
+        self.list.entry(peer).or_insert_with(|| Vec::new()).push(addr);
+    }
+
+    /// Returns a list of all the known peers in the topology.
+    #[inline]
+    pub fn peers(&self) -> impl Iterator<Item = &PeerId> {
+        self.list.keys()
+    }
+
+    /// Returns an iterator to all the entries in the topology.
+    #[inline]
+    pub fn iter(&self) -> impl Iterator<Item = (&PeerId, &Multiaddr)> {
+        self.list.iter().flat_map(|(p, l)| l.iter().map(move |ma| (p, ma)))
+    }
 }
 
 impl Default for MemoryTopology {
@@ -52,10 +74,6 @@ impl Default for MemoryTopology {
 }
 
 impl Topology for MemoryTopology {
-    fn add_discovered_address(&mut self, peer: &PeerId, addr: Multiaddr) {
-        self.list.entry(peer.clone()).or_insert_with(|| Vec::new()).push(addr);
-    }
-
     fn addresses_of_peer(&mut self, peer: &PeerId) -> Vec<Multiaddr> {
         self.list.get(peer).map(|v| v.clone()).unwrap_or(Vec::new())
     }

examples/ipfs-kad.rs

@@ -0,0 +1,100 @@
+// Copyright 2018 Parity Technologies (UK) Ltd.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a
+// copy of this software and associated documentation files (the "Software"),
+// to deal in the Software without restriction, including without limitation
+// the rights to use, copy, modify, merge, publish, distribute, sublicense,
+// and/or sell copies of the Software, and to permit persons to whom the
+// Software is furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+// DEALINGS IN THE SOFTWARE.
+
+//! Demonstrates how to perform Kademlia queries on the IPFS network.
+//!
+//! You can pass as parameter a base58 peer ID to search for. If you don't pass any parameter, a
+//! peer ID will be generated randomly.
+
+extern crate futures;
+extern crate libp2p;
+extern crate rand;
+extern crate tokio;
+
+use futures::prelude::*;
+use libp2p::{
+    Transport,
+    core::PublicKey,
+    core::upgrade::{self, OutboundUpgradeExt},
+    secio,
+    mplex,
+};
+
+fn main() {
+    // Create a random key for ourselves.
+    let local_key = secio::SecioKeyPair::ed25519_generated().unwrap();
+    let local_peer_id = local_key.to_peer_id();
+
+    // Set up a an encrypted DNS-enabled TCP Transport over the Mplex protocol
+    let transport = libp2p::CommonTransport::new()
+        .with_upgrade(secio::SecioConfig::new(local_key))
+        .and_then(move |out, _| {
+            let peer_id = out.remote_key.into_peer_id();
+            let upgrade = mplex::MplexConfig::new().map_outbound(move |muxer| (peer_id, muxer) );
+            upgrade::apply_outbound(out.stream, upgrade).map_err(|e| e.into_io_error())
+        });
+
+    // Create the topology of the network with the IPFS bootstrap nodes.
+    let mut topology = libp2p::core::topology::MemoryTopology::empty();
+    topology.add_address("QmaCpDMGvV2BGHeYERUEnRQAwe3N8SzbUtfsmvsqQLuvuJ".parse().unwrap(), "/ip4/104.131.131.82/tcp/4001".parse().unwrap());
+    topology.add_address("QmSoLPppuBtQSGwKDZT2M73ULpjvfd3aZ6ha4oFGL1KrGM".parse().unwrap(), "/ip4/104.236.179.241/tcp/4001".parse().unwrap());
+    topology.add_address("QmSoLV4Bbm51jM9C4gDYZQ9Cy3U6aXMJDAbzgu2fzaDs64".parse().unwrap(), "/ip4/104.236.76.40/tcp/4001".parse().unwrap());
+    topology.add_address("QmSoLSafTMBsPKadTEgaXctDQVcqN88CNLHXMkTNwMKPnu".parse().unwrap(), "/ip4/128.199.219.111/tcp/4001".parse().unwrap());
+    topology.add_address("QmSoLer265NRgSp2LA3dPaeykiS1J6DifTC88f5uVQKNAd".parse().unwrap(), "/ip4/178.62.158.247/tcp/4001".parse().unwrap());
+    topology.add_address("QmSoLSafTMBsPKadTEgaXctDQVcqN88CNLHXMkTNwMKPnu".parse().unwrap(), "/ip6/2400:6180:0:d0::151:6001/tcp/4001".parse().unwrap());
+    topology.add_address("QmSoLPppuBtQSGwKDZT2M73ULpjvfd3aZ6ha4oFGL1KrGM".parse().unwrap(), "/ip6/2604:a880:1:20::203:d001/tcp/4001".parse().unwrap());
+    topology.add_address("QmSoLV4Bbm51jM9C4gDYZQ9Cy3U6aXMJDAbzgu2fzaDs64".parse().unwrap(), "/ip6/2604:a880:800:10::4a:5001/tcp/4001".parse().unwrap());
+    topology.add_address("QmSoLer265NRgSp2LA3dPaeykiS1J6DifTC88f5uVQKNAd".parse().unwrap(), "/ip6/2a03:b0c0:0:1010::23:1001/tcp/4001".parse().unwrap());
+
+    // Create a swarm to manage peers and events.
+    let mut swarm = {
+        // Create a Kademlia behaviour.
+        // Note that normally the Kademlia process starts by performing lots of request in order
+        // to insert our local node in the DHT. However here we use `without_init` because this
+        // example is very ephemeral and we don't want to pollute the DHT. In a real world
+        // application, you want to use `new` instead.
+        let mut behaviour = libp2p::kad::Kademlia::without_init(local_peer_id);
+        libp2p::core::Swarm::new(transport, behaviour, topology)
+    };
+
+    // Order Kademlia to search for a peer.
+    let to_search = if let Some(peer_id) = std::env::args().nth(1) {
+        peer_id.parse().expect("Failed to parse peer ID to find")
+    } else {
+        PublicKey::Secp256k1((0..32).map(|_| -> u8 { rand::random() }).collect()).into_peer_id()
+    };
+    println!("Searching for {:?}", to_search);
+    swarm.find_node(to_search);
+
+    // Kick it off!
+    tokio::run(futures::future::poll_fn(move || -> Result<_, ()> {
+        loop {
+            match swarm.poll().expect("Error while polling swarm") {
+                Async::Ready(Some(event)) => {
+                    println!("Result: {:#?}", event);
+                    return Ok(Async::Ready(()));
+                },
+                Async::Ready(None) | Async::NotReady => break,
+            }
+        }
+
+        Ok(Async::NotReady)
+    }));
+}

misc/core-derive/src/lib.rs

@@ -68,12 +68,23 @@ fn build_struct(ast: &DeriveInput, data_struct: &DataStruct) -> TokenStream {
         quote!{#n}
     };
 
+    // Name of the type parameter that represents the topology.
+    let topology_generic = {
+        let mut n = "TTopology".to_string();
+        // Avoid collisions.
+        while ast.generics.type_params().any(|tp| tp.ident.to_string() == n) {
+            n.push('1');
+        }
+        let n = Ident::new(&n, name.span());
+        quote!{#n}
+    };
+
     // Build the generics.
     let impl_generics = {
         let tp = ast.generics.type_params();
         let lf = ast.generics.lifetimes();
         let cst = ast.generics.const_params();
-        quote!{<#(#lf,)* #(#tp,)* #(#cst,)* #substream_generic>}
+        quote!{<#(#lf,)* #(#tp,)* #(#cst,)* #topology_generic, #substream_generic>}
     };
 
     // Build the `where ...` clause of the trait implementation.
@@ -83,11 +94,11 @@ fn build_struct(ast: &DeriveInput, data_struct: &DataStruct) -> TokenStream {
             .flat_map(|field| {
                 let ty = &field.ty;
                 vec![
-                    quote!{#ty: #trait_to_impl},
-                    quote!{<#ty as #trait_to_impl>::ProtocolsHandler: #protocols_handler<Substream = #substream_generic>},
+                    quote!{#ty: #trait_to_impl<#topology_generic>},
+                    quote!{<#ty as #trait_to_impl<#topology_generic>>::ProtocolsHandler: #protocols_handler<Substream = #substream_generic>},
                     // Note: this bound is required because of https://github.com/rust-lang/rust/issues/55697
-                    quote!{<<#ty as #trait_to_impl>::ProtocolsHandler as #protocols_handler>::InboundProtocol: ::libp2p::core::InboundUpgrade<#substream_generic>},
-                    quote!{<<#ty as #trait_to_impl>::ProtocolsHandler as #protocols_handler>::OutboundProtocol: ::libp2p::core::OutboundUpgrade<#substream_generic>},
+                    quote!{<<#ty as #trait_to_impl<#topology_generic>>::ProtocolsHandler as #protocols_handler>::InboundProtocol: ::libp2p::core::InboundUpgrade<#substream_generic>},
+                    quote!{<<#ty as #trait_to_impl<#topology_generic>>::ProtocolsHandler as #protocols_handler>::OutboundProtocol: ::libp2p::core::OutboundUpgrade<#substream_generic>},
                 ]
             })
             .collect::<Vec<_>>();
@@ -196,7 +207,7 @@ fn build_struct(ast: &DeriveInput, data_struct: &DataStruct) -> TokenStream {
                 continue;
             }
             let ty = &field.ty;
-            let field_info = quote!{ <#ty as #trait_to_impl>::ProtocolsHandler };
+            let field_info = quote!{ <#ty as #trait_to_impl<#topology_generic>>::ProtocolsHandler };
             match ph_ty {
                 Some(ev) => ph_ty = Some(quote!{ #proto_select_ident<#ev, #field_info> }),
                 ref mut ev @ None => *ev = Some(field_info),
@@ -295,7 +306,7 @@ fn build_struct(ast: &DeriveInput, data_struct: &DataStruct) -> TokenStream {
 
         Some(quote!{
             loop {
-                match #field_name.poll() {
+                match #field_name.poll(topology) {
                     Async::Ready(#network_behaviour_action::GenerateEvent(event)) => {
                         #handling
                     }
@@ -319,7 +330,7 @@ fn build_struct(ast: &DeriveInput, data_struct: &DataStruct) -> TokenStream {
 
     // Now the magic happens.
     let final_quote = quote!{
-        impl #impl_generics #trait_to_impl for #name #ty_generics
+        impl #impl_generics #trait_to_impl<#topology_generic> for #name #ty_generics
         #where_clause
         {
             type ProtocolsHandler = #protocols_handler_ty;
@@ -352,7 +363,7 @@ fn build_struct(ast: &DeriveInput, data_struct: &DataStruct) -> TokenStream {
                 }
             }
 
-            fn poll(&mut self) -> ::libp2p::futures::Async<#network_behaviour_action<<Self::ProtocolsHandler as #protocols_handler>::InEvent, Self::OutEvent>> {
+            fn poll(&mut self, topology: &mut #topology_generic) -> ::libp2p::futures::Async<#network_behaviour_action<<Self::ProtocolsHandler as #protocols_handler>::InEvent, Self::OutEvent>> {
                 use libp2p::futures::prelude::*;
                 #(#poll_stmts)*
                 let f: ::libp2p::futures::Async<#network_behaviour_action<<Self::ProtocolsHandler as #protocols_handler>::InEvent, Self::OutEvent>> = #poll_method;

misc/core-derive/tests/test.rs

@@ -23,7 +23,7 @@ extern crate libp2p;
 
 /// Small utility to check that a type implements `NetworkBehaviour`.
 #[allow(dead_code)]
-fn require_net_behaviour<T: libp2p::core::swarm::NetworkBehaviour>() {}
+fn require_net_behaviour<T: libp2p::core::swarm::NetworkBehaviour<libp2p::core::topology::MemoryTopology>>() {}
 
 // TODO: doesn't compile
 /*#[test]
@@ -73,7 +73,8 @@ fn three_fields() {
     }
 }
 
-#[test]
+// TODO: fix this example ; a Rust bug prevent us from doing so
+/*#[test]
 fn event_handler() {
     #[allow(dead_code)]
     #[derive(NetworkBehaviour)]
@@ -93,7 +94,7 @@ fn event_handler() {
     fn foo<TSubstream: libp2p::tokio_io::AsyncRead + libp2p::tokio_io::AsyncWrite>() {
         require_net_behaviour::<Foo<TSubstream>>();
     }
-}
+}*/
 
 #[test]
 fn custom_polling() {

protocols/floodsub/src/layer.rs

@@ -172,7 +172,7 @@ impl<TSubstream> FloodsubBehaviour<TSubstream> {
     }
 }
 
-impl<TSubstream> NetworkBehaviour for FloodsubBehaviour<TSubstream>
+impl<TSubstream, TTopology> NetworkBehaviour<TTopology> for FloodsubBehaviour<TSubstream>
 where
     TSubstream: AsyncRead + AsyncWrite,
 {
@@ -276,6 +276,7 @@ where
 
     fn poll(
         &mut self,
+        _: &mut TTopology,
     ) -> Async<
         NetworkBehaviourAction<
             <Self::ProtocolsHandler as ProtocolsHandler>::InEvent,

protocols/identify/src/listen_layer.rs

@@ -61,7 +61,7 @@ impl<TSubstream> IdentifyListen<TSubstream> {
     }
 }
 
-impl<TSubstream> NetworkBehaviour for IdentifyListen<TSubstream>
+impl<TSubstream, TTopology> NetworkBehaviour<TTopology> for IdentifyListen<TSubstream>
 where
     TSubstream: AsyncRead + AsyncWrite,
 {
@@ -99,6 +99,7 @@ where
 
     fn poll(
         &mut self,
+        _: &mut TTopology,
     ) -> Async<
         NetworkBehaviourAction<
             <Self::ProtocolsHandler as ProtocolsHandler>::InEvent,