Fix examples and update core-derive.

Cargo.toml

@@ -45,9 +45,8 @@ libp2p-tcp = { version = "0.13.0", path = "transports/tcp" }
 libp2p-websocket = { version = "0.13.0", path = "transports/websocket", optional = true }
 
 [dev-dependencies]
+async-std = "1.0"
 env_logger = "0.7.1"
-tokio = "0.1"
-tokio-stdin-stdout = "0.1"
 
 [workspace]
 members = [

examples/chat.rs

@@ -49,20 +49,21 @@
 //!
 //! The two nodes then connect.
 
-use futures::prelude::*;
+use async_std::{io, task};
+use futures::{future, prelude::*};
 use libp2p::{
+    Multiaddr,
     PeerId,
     Swarm,
     NetworkBehaviour,
     identity,
-    tokio_codec::{FramedRead, LinesCodec},
-    tokio_io::{AsyncRead, AsyncWrite},
     floodsub::{self, Floodsub, FloodsubEvent},
     mdns::{Mdns, MdnsEvent},
     swarm::NetworkBehaviourEventProcess
 };
+use std::{error::Error, task::{Context, Poll}};
 
-fn main() {
+fn main() -> Result<(), Box<dyn Error>> {
     env_logger::init();
 
     // Create a random PeerId
@@ -71,7 +72,7 @@ fn main() {
     println!("Local peer id: {:?}", local_peer_id);
 
     // Set up a an encrypted DNS-enabled TCP Transport over the Mplex and Yamux protocols
-    let transport = libp2p::build_development_transport(local_key);
+    let transport = libp2p::build_development_transport(local_key)?;
 
     // Create a Floodsub topic
     let floodsub_topic = floodsub::TopicBuilder::new("chat").build();
@@ -87,18 +88,16 @@ fn main() {
     impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<MdnsEvent> for MyBehaviour<TSubstream> {
         fn inject_event(&mut self, event: MdnsEvent) {
             match event {
-                MdnsEvent::Discovered(list) => {
+                MdnsEvent::Discovered(list) =>
                     for (peer, _) in list {
                         self.floodsub.add_node_to_partial_view(peer);
                     }
-                },
-                MdnsEvent::Expired(list) => {
+                MdnsEvent::Expired(list) =>
                     for (peer, _) in list {
                         if !self.mdns.has_node(&peer) {
                             self.floodsub.remove_node_from_partial_view(&peer);
                         }
                     }
-                }
             }
         }
     }
@@ -114,9 +113,10 @@ fn main() {
 
     // Create a Swarm to manage peers and events
     let mut swarm = {
+        let mdns = task::block_on(Mdns::new())?;
         let mut behaviour = MyBehaviour {
             floodsub: Floodsub::new(local_peer_id.clone()),
-            mdns: Mdns::new().expect("Failed to create mDNS service"),
+            mdns
         };
 
         behaviour.floodsub.subscribe(floodsub_topic.clone());
@@ -125,42 +125,32 @@ fn main() {
 
     // Reach out to another node if specified
     if let Some(to_dial) = std::env::args().nth(1) {
-        let dialing = to_dial.clone();
-        match to_dial.parse() {
-            Ok(to_dial) => {
-                match libp2p::Swarm::dial_addr(&mut swarm, to_dial) {
-                    Ok(_) => println!("Dialed {:?}", dialing),
-                    Err(e) => println!("Dial {:?} failed: {:?}", dialing, e)
-                }
-            },
-            Err(err) => println!("Failed to parse address to dial: {:?}", err),
-        }
+        let addr: Multiaddr = to_dial.parse()?;
+        Swarm::dial_addr(&mut swarm, addr)?;
+        println!("Dialed {:?}", to_dial)
     }
 
     // Read full lines from stdin
-    let stdin = tokio_stdin_stdout::stdin(0);
-    let mut framed_stdin = FramedRead::new(stdin, LinesCodec::new());
+    let mut stdin = io::BufReader::new(io::stdin()).lines();
 
     // Listen on all interfaces and whatever port the OS assigns
-    Swarm::listen_on(&mut swarm, "/ip4/0.0.0.0/tcp/0".parse().unwrap()).unwrap();
+    Swarm::listen_on(&mut swarm, "/ip4/0.0.0.0/tcp/0".parse()?)?;
 
     // Kick it off
     let mut listening = false;
-    tokio::run(futures::future::poll_fn(move || -> Result<_, ()> {
+    task::block_on(future::poll_fn(move |cx: &mut Context| {
         loop {
-            match framed_stdin.poll().expect("Error while polling stdin") {
-                Async::Ready(Some(line)) => swarm.floodsub.publish(&floodsub_topic, line.as_bytes()),
-                Async::Ready(None) => panic!("Stdin closed"),
-                Async::NotReady => break,
-            };
+            match stdin.try_poll_next_unpin(cx)? {
+                Poll::Ready(Some(line)) => swarm.floodsub.publish(&floodsub_topic, line.as_bytes()),
+                Poll::Ready(None) => panic!("Stdin closed"),
+                Poll::Pending => break
+            }
         }
-
         loop {
-            match swarm.poll().expect("Error while polling swarm") {
-                Async::Ready(Some(_)) => {
-
-                },
-                Async::Ready(None) | Async::NotReady => {
+            match swarm.poll_next_unpin(cx) {
+                Poll::Ready(Some(event)) => println!("{:?}", event),
+                Poll::Ready(None) => return Poll::Ready(Ok(())),
+                Poll::Pending => {
                     if !listening {
                         if let Some(a) = Swarm::listeners(&swarm).next() {
                             println!("Listening on {:?}", a);
@@ -171,7 +161,6 @@ fn main() {
                 }
             }
         }
-
-        Ok(Async::NotReady)
-    }));
+        Poll::Pending
+    }))
 }

examples/distributed-key-value-store.rs

@@ -29,37 +29,41 @@
 //!
 //! 4. Close with Ctrl-c.
 
+use async_std::{io, task};
 use futures::prelude::*;
 use libp2p::kad::record::store::MemoryStore;
 use libp2p::kad::{record::Key, Kademlia, KademliaEvent, PutRecordOk, Quorum, Record};
 use libp2p::{
-    build_development_transport, identity,
+    NetworkBehaviour,
+    PeerId,
+    Swarm,
+    build_development_transport,
+    identity,
     mdns::{Mdns, MdnsEvent},
-    swarm::NetworkBehaviourEventProcess,
-    tokio_codec::{FramedRead, LinesCodec},
-    tokio_io::{AsyncRead, AsyncWrite},
-    NetworkBehaviour, PeerId, Swarm,
+    swarm::NetworkBehaviourEventProcess
 };
+use std::{error::Error, task::{Context, Poll}};
 
-fn main() {
+fn main() -> Result<(), Box<dyn Error>> {
     env_logger::init();
 
     // Create a random key for ourselves.
     let local_key = identity::Keypair::generate_ed25519();
     let local_peer_id = PeerId::from(local_key.public());
 
     // Set up a an encrypted DNS-enabled TCP Transport over the Mplex protocol.
-    let transport = build_development_transport(local_key);
+    let transport = build_development_transport(local_key)?;
 
     // We create a custom network behaviour that combines Kademlia and mDNS.
     #[derive(NetworkBehaviour)]
     struct MyBehaviour<TSubstream: AsyncRead + AsyncWrite> {
         kademlia: Kademlia<TSubstream, MemoryStore>,
-        mdns: Mdns<TSubstream>,
+        mdns: Mdns<TSubstream>
     }
 
-    impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<MdnsEvent>
-        for MyBehaviour<TSubstream>
+    impl<T> NetworkBehaviourEventProcess<MdnsEvent> for MyBehaviour<T>
+    where
+        T: AsyncRead + AsyncWrite
     {
         // Called when `mdns` produces an event.
         fn inject_event(&mut self, event: MdnsEvent) {
@@ -71,8 +75,9 @@ fn main() {
         }
     }
 
-    impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<KademliaEvent>
-        for MyBehaviour<TSubstream>
+    impl<T> NetworkBehaviourEventProcess<KademliaEvent> for MyBehaviour<T>
+    where
+        T: AsyncRead + AsyncWrite
     {
         // Called when `kademlia` produces an event.
         fn inject_event(&mut self, message: KademliaEvent) {
@@ -108,58 +113,50 @@ fn main() {
         // Create a Kademlia behaviour.
         let store = MemoryStore::new(local_peer_id.clone());
         let kademlia = Kademlia::new(local_peer_id.clone(), store);
-
-        let behaviour = MyBehaviour {
-            kademlia,
-            mdns: Mdns::new().expect("Failed to create mDNS service"),
-        };
-
+        let mdns = task::block_on(Mdns::new())?;
+        let behaviour = MyBehaviour { kademlia, mdns };
         Swarm::new(transport, behaviour, local_peer_id)
     };
 
-    // Read full lines from stdin.
-    let stdin = tokio_stdin_stdout::stdin(0);
-    let mut framed_stdin = FramedRead::new(stdin, LinesCodec::new());
+    // Read full lines from stdin
+    let mut stdin = io::BufReader::new(io::stdin()).lines();
 
     // Listen on all interfaces and whatever port the OS assigns.
-    Swarm::listen_on(&mut swarm, "/ip4/0.0.0.0/tcp/0".parse().unwrap()).unwrap();
+    Swarm::listen_on(&mut swarm, "/ip4/0.0.0.0/tcp/0".parse()?)?;
 
     // Kick it off.
     let mut listening = false;
-    tokio::run(futures::future::poll_fn(move || {
+    task::block_on(future::poll_fn(move |cx: &mut Context| {
         loop {
-            match framed_stdin.poll().expect("Error while polling stdin") {
-                Async::Ready(Some(line)) => {
-                    handle_input_line(&mut swarm.kademlia, line);
-                }
-                Async::Ready(None) => panic!("Stdin closed"),
-                Async::NotReady => break,
-            };
+            match stdin.try_poll_next_unpin(cx)? {
+                Poll::Ready(Some(line)) => handle_input_line(&mut swarm.kademlia, line),
+                Poll::Ready(None) => panic!("Stdin closed"),
+                Poll::Pending => break
+            }
         }
-
         loop {
-            match swarm.poll().expect("Error while polling swarm") {
-                Async::Ready(Some(_)) => {}
-                Async::Ready(None) | Async::NotReady => {
+            match swarm.poll_next_unpin(cx) {
+                Poll::Ready(Some(event)) => println!("{:?}", event),
+                Poll::Ready(None) => return Poll::Ready(Ok(())),
+                Poll::Pending => {
                     if !listening {
                         if let Some(a) = Swarm::listeners(&swarm).next() {
                             println!("Listening on {:?}", a);
                             listening = true;
                         }
                     }
-                    break;
+                    break
                 }
             }
         }
-
-        Ok(Async::NotReady)
-    }));
+        Poll::Pending
+    }))
 }
 
-fn handle_input_line<TSubstream: AsyncRead + AsyncWrite>(
-    kademlia: &mut Kademlia<TSubstream, MemoryStore>,
-    line: String,
-) {
+fn handle_input_line<T>(kademlia: &mut Kademlia<T, MemoryStore>, line: String)
+where
+    T: AsyncRead + AsyncWrite
+{
     let mut args = line.split(" ");
 
     match args.next() {

examples/ipfs-kad.rs

@@ -23,6 +23,7 @@
 //! 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.
 
+use async_std::task;
 use futures::prelude::*;
 use libp2p::{
     Swarm,
@@ -32,18 +33,17 @@ use libp2p::{
 };
 use libp2p::kad::{Kademlia, KademliaConfig, KademliaEvent, GetClosestPeersError};
 use libp2p::kad::record::store::MemoryStore;
-use std::env;
-use std::time::Duration;
+use std::{env, error::Error, time::Duration};
 
-fn main() {
+fn main() -> Result<(), Box<dyn Error>> {
     env_logger::init();
 
     // Create a random key for ourselves.
     let local_key = identity::Keypair::generate_ed25519();
     let local_peer_id = PeerId::from(local_key.public());
 
     // Set up a an encrypted DNS-enabled TCP Transport over the Mplex protocol
-    let transport = build_development_transport(local_key);
+    let transport = build_development_transport(local_key)?;
 
     // Create a swarm to manage peers and events.
     let mut swarm = {
@@ -60,7 +60,7 @@ fn main() {
         behaviour.add_address(&"QmcZf59bWwK5XFi76CZX8cbJ4BhTzzA3gU1ZjYZcYW3dwt".parse().unwrap(), "/dnsaddr/bootstrap.libp2p.io".parse().unwrap());*/
 
         // The only address that currently works.
-        behaviour.add_address(&"QmaCpDMGvV2BGHeYERUEnRQAwe3N8SzbUtfsmvsqQLuvuJ".parse().unwrap(), "/ip4/104.131.131.82/tcp/4001".parse().unwrap());
+        behaviour.add_address(&"QmaCpDMGvV2BGHeYERUEnRQAwe3N8SzbUtfsmvsqQLuvuJ".parse()?, "/ip4/104.131.131.82/tcp/4001".parse()?);
 
         // The following addresses always fail signature verification, possibly due to
         // RSA keys with < 2048 bits.
@@ -80,7 +80,7 @@ fn main() {
 
     // Order Kademlia to search for a peer.
     let to_search: PeerId = if let Some(peer_id) = env::args().nth(1) {
-        peer_id.parse().expect("Failed to parse peer ID to find")
+        peer_id.parse()?
     } else {
         identity::Keypair::generate_ed25519().public().into()
     };
@@ -89,38 +89,29 @@ fn main() {
     swarm.get_closest_peers(to_search);
 
     // Kick it off!
-    tokio::run(futures::future::poll_fn(move || {
-        loop {
-            match swarm.poll().expect("Error while polling swarm") {
-                Async::Ready(Some(KademliaEvent::GetClosestPeersResult(res))) => {
-                    match res {
-                        Ok(ok) => {
-                            if !ok.peers.is_empty() {
-                                println!("Query finished with closest peers: {:#?}", ok.peers);
-                                return Ok(Async::Ready(()));
-                            } else {
-                                // The example is considered failed as there
-                                // should always be at least 1 reachable peer.
-                                panic!("Query finished with no closest peers.");
-                            }
+    task::block_on(async move {
+        while let Some(event) = swarm.try_next().await? {
+            if let KademliaEvent::GetClosestPeersResult(result) = event {
+                match result {
+                    Ok(ok) =>
+                        if !ok.peers.is_empty() {
+                            println!("Query finished with closest peers: {:#?}", ok.peers)
+                        } else {
+                            // The example is considered failed as there
+                            // should always be at least 1 reachable peer.
+                            panic!("Query finished with no closest peers.")
                         }
-                        Err(GetClosestPeersError::Timeout { peers, .. }) => {
-                            if !peers.is_empty() {
-                                println!("Query timed out with closest peers: {:#?}", peers);
-                                return Ok(Async::Ready(()));
-                            } else {
-                                // The example is considered failed as there
-                                // should always be at least 1 reachable peer.
-                                panic!("Query timed out with no closest peers.");
-                            }
+                    Err(GetClosestPeersError::Timeout { peers, .. }) =>
+                        if !peers.is_empty() {
+                            println!("Query timed out with closest peers: {:#?}", peers)
+                        } else {
+                            // The example is considered failed as there
+                            // should always be at least 1 reachable peer.
+                            panic!("Query timed out with no closest peers.");
                         }
-                    }
-                },
-                Async::Ready(Some(_)) => {},
-                Async::Ready(None) | Async::NotReady => break,
+                }
             }
         }
-
-        Ok(Async::NotReady)
-    }));
+        Ok(())
+    })
 }