Merge remote-tracking branch 'upstream/master' into dp/chore/test-cor…

…e-handled_node

* upstream/master:
  Add protocol to report external address view. (libp2p#566)
  Add a TransportExt trait (libp2p#533)
  libp2p#399 remove tokio_current_thread tests (libp2p#577)

Cargo.toml

@@ -28,6 +28,7 @@ libp2p-uds = { path = "./transports/uds" }
 libp2p-websocket = { path = "./transports/websocket" }
 libp2p-yamux = { path = "./muxers/yamux" }
 tokio-codec = "0.1"
+tokio-executor = "0.1"
 tokio-io = "0.1"
 
 [target.'cfg(not(target_os = "emscripten"))'.dependencies]
@@ -43,7 +44,7 @@ bigint = "4.2"
 env_logger = "0.5.4"
 rand = "0.4"
 structopt =  "0.2"
-tokio-current-thread = "0.1"
+tokio = "0.1"
 tokio-io = "0.1"
 tokio-stdin = "0.1"
 
@@ -59,6 +60,7 @@ members = [
     "protocols/identify",
     "protocols/kad",
     "protocols/ping",
+    "protocols/observed",
     "transports/relay",
     "protocols/secio",
     "muxers/mplex",

core/Cargo.toml

@@ -29,7 +29,6 @@ libp2p-mplex = { path = "../muxers/mplex" }
 rand = "0.5"
 tokio = "0.1"
 tokio-codec = "0.1"
-tokio-current-thread = "0.1"
 tokio-timer = "0.2"
 assert_matches = "1.3"
 tokio-mock-task = "0.1"

core/src/lib.rs

@@ -129,11 +129,12 @@
 //! extern crate libp2p_ping;
 //! extern crate libp2p_core;
 //! extern crate libp2p_tcp_transport;
-//! extern crate tokio_current_thread;
+//! extern crate tokio;
 //!
 //! use futures::{Future, Stream};
 //! use libp2p_ping::{Ping, PingOutput};
 //! use libp2p_core::Transport;
+//! use tokio::runtime::current_thread::Runtime;
 //!
 //! # fn main() {
 //! let ping_finished_future = libp2p_tcp_transport::TcpConfig::new()
@@ -154,7 +155,8 @@
 //!     });
 //!
 //! // Runs until the ping arrives.
-//! tokio_current_thread::block_on_all(ping_finished_future).unwrap();
+//! let mut rt = Runtime::new().unwrap();
+//! let _ = rt.block_on(ping_finished_future).unwrap();
 //! # }
 //! ```
 //!
@@ -191,8 +193,6 @@ extern crate tokio;
 #[cfg(test)]
 extern crate tokio_codec;
 #[cfg(test)]
-extern crate tokio_current_thread;
-#[cfg(test)]
 extern crate tokio_timer;
 #[cfg(test)]
 #[macro_use]

core/src/upgrade/traits.rs

@@ -75,10 +75,5 @@ pub trait ConnectionUpgrade<C> {
     ///
     /// Because performing the upgrade may not be instantaneous (eg. it may require a handshake),
     /// this function returns a future instead of the direct output.
-    fn upgrade(
-        self,
-        socket: C,
-        id: Self::UpgradeIdentifier,
-        ty: Endpoint,
-    ) -> Self::Future;
+    fn upgrade(self, socket: C, id: Self::UpgradeIdentifier, ty: Endpoint) -> Self::Future;
 }

misc/multistream-select/Cargo.toml

@@ -14,5 +14,5 @@ tokio-io = "0.1"
 unsigned-varint = { version = "0.2.1", features = ["codec"] }
 
 [dev-dependencies]
-tokio-current-thread = "0.1"
+tokio = "0.1"
 tokio-tcp = "0.1"

misc/multistream-select/src/lib.rs

@@ -45,14 +45,15 @@
 //! extern crate bytes;
 //! extern crate futures;
 //! extern crate multistream_select;
-//! extern crate tokio_current_thread;
+//! extern crate tokio;
 //! extern crate tokio_tcp;
 //!
 //! # fn main() {
 //! use bytes::Bytes;
 //! use multistream_select::dialer_select_proto;
 //! use futures::{Future, Sink, Stream};
 //! use tokio_tcp::TcpStream;
+//! use tokio::runtime::current_thread::Runtime;
 //!
 //! #[derive(Debug, Copy, Clone)]
 //! enum MyProto { Echo, Hello }
@@ -68,7 +69,8 @@
 //!         dialer_select_proto(connec, protos).map(|r| r.0)
 //!     });
 //!
-//! let negotiated_protocol: MyProto = tokio_current_thread::block_on_all(client)
+//! let mut rt = Runtime::new().unwrap();
+//! let negotiated_protocol: MyProto = rt.block_on(client)
 //!     .expect("failed to find a protocol");
 //! println!("negotiated: {:?}", negotiated_protocol);
 //! # }
@@ -80,14 +82,15 @@
 //! extern crate bytes;
 //! extern crate futures;
 //! extern crate multistream_select;
-//! extern crate tokio_current_thread;
+//! extern crate tokio;
 //! extern crate tokio_tcp;
 //!
 //! # fn main() {
 //! use bytes::Bytes;
 //! use multistream_select::listener_select_proto;
 //! use futures::{Future, Sink, Stream};
 //! use tokio_tcp::TcpListener;
+//! use tokio::runtime::current_thread::Runtime;
 //!
 //! #[derive(Debug, Copy, Clone)]
 //! enum MyProto { Echo, Hello }
@@ -108,7 +111,8 @@
 //!         Ok(())
 //!     });
 //!
-//! tokio_current_thread::block_on_all(server).expect("failed to run server");
+//! let mut rt = Runtime::new().unwrap();
+//! let _ = rt.block_on(server).expect("failed to run server");
 //! # }
 //! ```
 

misc/multistream-select/src/protocol/dialer.rs

@@ -183,8 +183,9 @@ impl<T: AsyncWrite> Future for DialerFuture<T> {
 
 #[cfg(test)]
 mod tests {
-    extern crate tokio_current_thread;
+    extern crate tokio;
     extern crate tokio_tcp;
+    use self::tokio::runtime::current_thread::Runtime;
     use self::tokio_tcp::{TcpListener, TcpStream};
     use bytes::Bytes;
     use futures::Future;
@@ -210,7 +211,8 @@ mod tests {
                 dialer.send(DialerToListenerMessage::ProtocolRequest { name: p })
             });
 
-        match tokio_current_thread::block_on_all(server.join(client)) {
+        let mut rt = Runtime::new().unwrap();
+        match rt.block_on(server.join(client)) {
             Err(MultistreamSelectError::WrongProtocolName) => (),
             _ => panic!(),
         }

misc/multistream-select/src/protocol/listener.rs

@@ -225,8 +225,9 @@ impl<T: AsyncRead + AsyncWrite> Future for ListenerFuture<T> {
 
 #[cfg(test)]
 mod tests {
-    extern crate tokio_current_thread;
+    extern crate tokio;
     extern crate tokio_tcp;
+    use self::tokio::runtime::current_thread::Runtime;
     use self::tokio_tcp::{TcpListener, TcpStream};
     use bytes::Bytes;
     use futures::Future;
@@ -252,7 +253,8 @@ mod tests {
             .from_err()
             .and_then(move |stream| Dialer::new(stream));
 
-        match tokio_current_thread::block_on_all(server.join(client)) {
+        let mut rt = Runtime::new().unwrap();
+        match rt.block_on(server.join(client)) {
             Err(MultistreamSelectError::WrongProtocolName) => (),
             _ => panic!(),
         }

misc/multistream-select/src/tests.rs

@@ -22,9 +22,10 @@
 
 #![cfg(test)]
 
-extern crate tokio_current_thread;
+extern crate tokio;
 extern crate tokio_tcp;
 
+use self::tokio::runtime::current_thread::Runtime;
 use self::tokio_tcp::{TcpListener, TcpStream};
 use bytes::Bytes;
 use dialer_select::{dialer_select_proto_parallel, dialer_select_proto_serial};
@@ -69,8 +70,8 @@ fn negotiate_with_self_succeeds() {
             assert_eq!(proto, "/hello/1.0.0");
             Ok(())
         });
-
-    tokio_current_thread::block_on_all(server.join(client)).unwrap();
+    let mut rt = Runtime::new().unwrap();
+    let _ = rt.block_on(server.join(client)).unwrap();
 }
 
 #[test]
@@ -100,9 +101,9 @@ fn select_proto_basic() {
             ].into_iter();
             dialer_select_proto(connec, protos).map(|r| r.0)
         });
-
+    let mut rt = Runtime::new().unwrap();
     let (dialer_chosen, listener_chosen) =
-        tokio_current_thread::block_on_all(client.join(server)).unwrap();
+        rt.block_on(client.join(server)).unwrap();
     assert_eq!(dialer_chosen, 3);
     assert_eq!(listener_chosen, 1);
 }
@@ -134,8 +135,8 @@ fn no_protocol_found() {
             ].into_iter();
             dialer_select_proto(connec, protos).map(|r| r.0)
         });
-
-    match tokio_current_thread::block_on_all(client.join(server)) {
+    let mut rt = Runtime::new().unwrap();
+    match rt.block_on(client.join(server)) {
         Err(ProtocolChoiceError::NoProtocolFound) => (),
         _ => panic!(),
     }
@@ -169,8 +170,9 @@ fn select_proto_parallel() {
             dialer_select_proto_parallel(connec, protos).map(|r| r.0)
         });
 
+    let mut rt = Runtime::new().unwrap();
     let (dialer_chosen, listener_chosen) =
-        tokio_current_thread::block_on_all(client.join(server)).unwrap();
+        rt.block_on(client.join(server)).unwrap();
     assert_eq!(dialer_chosen, 3);
     assert_eq!(listener_chosen, 1);
 }
@@ -200,8 +202,9 @@ fn select_proto_serial() {
             dialer_select_proto_serial(connec, protos).map(|r| r.0)
         });
 
+    let mut rt = Runtime::new().unwrap();
     let (dialer_chosen, listener_chosen) =
-        tokio_current_thread::block_on_all(client.join(server)).unwrap();
+        rt.block_on(client.join(server)).unwrap();
     assert_eq!(dialer_chosen, 3);
     assert_eq!(listener_chosen, 1);
 }

muxers/mplex/Cargo.toml

@@ -17,4 +17,4 @@ unsigned-varint = { version = "0.2.1", features = ["codec"] }
 
 [dev-dependencies]
 libp2p-tcp-transport = { path = "../../transports/tcp" }
-tokio-current-thread = "0.1"
+tokio = "0.1"

muxers/mplex/tests/two_peers.rs

@@ -23,7 +23,7 @@ extern crate futures;
 extern crate libp2p_mplex as multiplex;
 extern crate libp2p_core as swarm;
 extern crate libp2p_tcp_transport as tcp;
-extern crate tokio_current_thread;
+extern crate tokio;
 extern crate tokio_io;
 
 use futures::future::Future;
@@ -33,6 +33,7 @@ use std::thread;
 use swarm::{muxing, Transport};
 use tcp::TcpConfig;
 use tokio_io::codec::length_delimited::Framed;
+use tokio::runtime::current_thread::Runtime;
 
 #[test]
 fn client_to_server_outbound() {
@@ -67,7 +68,8 @@ fn client_to_server_outbound() {
                 Ok(())
             });
 
-        tokio_current_thread::block_on_all(future).unwrap();
+        let mut rt = Runtime::new().unwrap();
+        let _ = rt.block_on(future).unwrap();
     });
 
     let transport = TcpConfig::new().with_upgrade(multiplex::MplexConfig::new());
@@ -80,7 +82,8 @@ fn client_to_server_outbound() {
         .and_then(|server| server.send("hello world".into()))
         .map(|_| ());
 
-    tokio_current_thread::block_on_all(future).unwrap();
+    let mut rt = Runtime::new().unwrap();
+    let _ = rt.block_on(future).unwrap();
     bg_thread.join().unwrap();
 }
 
@@ -117,7 +120,8 @@ fn client_to_server_inbound() {
                 Ok(())
             });
 
-        tokio_current_thread::block_on_all(future).unwrap();
+        let mut rt = Runtime::new().unwrap();
+        let _ = rt.block_on(future).unwrap();
     });
 
     let transport = TcpConfig::new().with_upgrade(multiplex::MplexConfig::new());
@@ -130,6 +134,7 @@ fn client_to_server_inbound() {
         .and_then(|server| server.send("hello world".into()))
         .map(|_| ());
 
-    tokio_current_thread::block_on_all(future).unwrap();
+    let mut rt = Runtime::new().unwrap();
+    let _ = rt.block_on(future).unwrap();
     bg_thread.join().unwrap();
 }

protocols/identify/Cargo.toml

@@ -22,4 +22,4 @@ void = "1.0"
 
 [dev-dependencies]
 libp2p-tcp-transport = { path = "../../transports/tcp" }
-tokio-current-thread = "0.1"
+tokio = "0.1"

protocols/identify/src/protocol.rs

@@ -211,8 +211,9 @@ fn parse_proto_msg(msg: BytesMut) -> Result<(IdentifyInfo, Multiaddr), IoError>
 #[cfg(test)]
 mod tests {
     extern crate libp2p_tcp_transport;
-    extern crate tokio_current_thread;
+    extern crate tokio;
 
+    use self::tokio::runtime::current_thread::Runtime;
     use self::libp2p_tcp_transport::TcpConfig;
     use futures::{Future, Stream};
     use libp2p_core::{PublicKey, Transport};
@@ -255,8 +256,8 @@ mod tests {
                     ),
                     _ => panic!(),
                 });
-
-            let _ = tokio_current_thread::block_on_all(future).unwrap();
+            let mut rt = Runtime::new().unwrap();
+            let _ = rt.block_on(future).unwrap();
         });
 
         let transport = TcpConfig::new().with_upgrade(IdentifyProtocolConfig);
@@ -291,8 +292,8 @@ mod tests {
                 }
                 _ => panic!(),
             });
-
-        let _ = tokio_current_thread::block_on_all(future).unwrap();
+        let mut rt = Runtime::new().unwrap();
+        let _ = rt.block_on(future).unwrap();
         bg_thread.join().unwrap();
     }
 }

protocols/kad/Cargo.toml

@@ -30,4 +30,4 @@ unsigned-varint = { version = "0.2.1", features = ["codec"] }
 [dev-dependencies]
 libp2p-tcp-transport = { path = "../../transports/tcp" }
 rand = "0.4.2"
-tokio-current-thread = "0.1"
+tokio = "0.1"

protocols/kad/src/protocol.rs

@@ -409,7 +409,7 @@ fn proto_to_msg(mut message: protobuf_structs::dht::Message) -> Result<KadMsg, I
 #[cfg(test)]
 mod tests {
     extern crate libp2p_tcp_transport;
-    extern crate tokio_current_thread;
+    extern crate tokio;
 
     use self::libp2p_tcp_transport::TcpConfig;
     use futures::{Future, Sink, Stream};
@@ -418,6 +418,8 @@ mod tests {
     use protocol::{KadConnectionType, KadMsg, KademliaProtocolConfig, KadPeer};
     use std::sync::mpsc;
     use std::thread;
+    use self::tokio::runtime::current_thread::Runtime;
+
 
     #[test]
     fn correct_transfer() {
@@ -494,8 +496,8 @@ mod tests {
                         assert_eq!(recv_msg.unwrap(), msg_server);
                         ()
                     });
-
-                let _ = tokio_current_thread::block_on_all(future).unwrap();
+                let mut rt = Runtime::new().unwrap();
+                let _ = rt.block_on(future).unwrap();
             });
 
             let transport = TcpConfig::new().with_upgrade(KademliaProtocolConfig);
@@ -505,8 +507,8 @@ mod tests {
                 .unwrap_or_else(|_| panic!())
                 .and_then(|proto| proto.send(msg_client))
                 .map(|_| ());
-
-            let _ = tokio_current_thread::block_on_all(future).unwrap();
+            let mut rt = Runtime::new().unwrap();
+            let _ = rt.block_on(future).unwrap();
             bg_thread.join().unwrap();
         }
     }