Add top-level tests for futures 0.2 integration (#231)

src/executor/mod.rs

@@ -136,6 +136,9 @@ pub use tokio_executor::{Executor, DefaultExecutor, SpawnError};
 use futures::{Future, IntoFuture};
 use futures::future::{self, FutureResult};
 
+#[cfg(feature = "unstable-futures")]
+use futures2;
+
 /// Return value from the `spawn` function.
 ///
 /// Currently this value doesn't actually provide any functionality. However, it
@@ -205,6 +208,15 @@ where F: Future<Item = (), Error = ()> + 'static + Send
     Spawn(())
 }
 
+/// Like `spawn`, but compatible with futures 0.2
+#[cfg(feature = "unstable-futures")]
+pub fn spawn2<F>(f: F) -> Spawn
+    where F: futures2::Future<Item = (), Error = futures2::Never> + 'static + Send
+{
+    ::tokio_executor::spawn2(f);
+    Spawn(())
+}
+
 impl IntoFuture for Spawn {
     type Future = FutureResult<(), ()>;
     type Item = ();
@@ -214,3 +226,14 @@ impl IntoFuture for Spawn {
         future::ok(())
     }
 }
+
+#[cfg(feature = "unstable-futures")]
+impl futures2::IntoFuture for Spawn {
+    type Future = futures2::future::FutureResult<(), ()>;
+    type Item = ();
+    type Error = ();
+
+    fn into_future(self) -> Self::Future {
+        futures2::future::ok(())
+    }
+}

src/lib.rs

@@ -84,6 +84,9 @@ pub mod reactor;
 pub mod runtime;
 
 pub use executor::spawn;
+#[cfg(feature = "unstable-futures")]
+pub use executor::spawn2;
+
 pub use runtime::run;
 
 pub mod io {

tests/echo2.rs

@@ -0,0 +1,53 @@
+#![cfg(feature = "unstable-futures")]
+
+// This test is the same as `echo.rs`, but ported to futures 0.2
+
+extern crate env_logger;
+extern crate futures2;
+extern crate tokio;
+extern crate tokio_io;
+
+use std::io::{Read, Write};
+use std::net::TcpStream;
+use std::thread;
+
+use futures2::prelude::*;
+use futures2::executor::block_on;
+use tokio::net::TcpListener;
+
+macro_rules! t {
+    ($e:expr) => (match $e {
+        Ok(e) => e,
+        Err(e) => panic!("{} failed with {:?}", stringify!($e), e),
+    })
+}
+
+#[test]
+fn echo_server() {
+    drop(env_logger::init());
+
+    let srv = t!(TcpListener::bind(&t!("127.0.0.1:0".parse())));
+    let addr = t!(srv.local_addr());
+
+    let msg = "foo bar baz";
+    let t = thread::spawn(move || {
+        let mut s = TcpStream::connect(&addr).unwrap();
+
+        for _i in 0..1024 {
+            assert_eq!(t!(s.write(msg.as_bytes())), msg.len());
+            let mut buf = [0; 1024];
+            assert_eq!(t!(s.read(&mut buf)), msg.len());
+            assert_eq!(&buf[..msg.len()], msg.as_bytes());
+        }
+    });
+
+    let clients = srv.incoming();
+    let client = clients.into_future().map(|e| e.0.unwrap()).map_err(|e| e.0);
+    let halves = client.map(|s| s.split());
+    let copied = halves.and_then(|(a, b)| a.copy_into(b));
+
+    let (amt, _, _) = t!(block_on(copied));
+    t.join().unwrap();
+
+    assert_eq!(amt, msg.len() as u64 * 1024);
+}

tests/global2.rs

@@ -0,0 +1,122 @@
+#![cfg(feature = "unstable-futures")]
+
+// This test is the same as `global.rs`, but ported to futures 0.2
+
+extern crate futures;
+extern crate futures2;
+extern crate tokio;
+extern crate tokio_io;
+extern crate env_logger;
+
+use std::{io, thread};
+use std::sync::Arc;
+
+use futures2::prelude::*;
+use futures2::executor::block_on;
+use futures2::task;
+
+use tokio::net::{TcpStream, TcpListener};
+use tokio::runtime::Runtime;
+
+macro_rules! t {
+    ($e:expr) => (match $e {
+        Ok(e) => e,
+        Err(e) => panic!("{} failed with {:?}", stringify!($e), e),
+    })
+}
+
+#[test]
+fn hammer() {
+    let _ = env_logger::init();
+
+    let threads = (0..10).map(|_| {
+        thread::spawn(|| {
+            let srv = t!(TcpListener::bind(&"127.0.0.1:0".parse().unwrap()));
+            let addr = t!(srv.local_addr());
+            let mine = TcpStream::connect(&addr);
+            let theirs = srv.incoming().into_future()
+                .map(|(s, _)| s.unwrap())
+                .map_err(|(s, _)| s);
+            let (mine, theirs) = t!(block_on(mine.join(theirs)));
+
+            assert_eq!(t!(mine.local_addr()), t!(theirs.peer_addr()));
+            assert_eq!(t!(theirs.local_addr()), t!(mine.peer_addr()));
+        })
+    }).collect::<Vec<_>>();
+    for thread in threads {
+        thread.join().unwrap();
+    }
+}
+
+struct Rd(Arc<TcpStream>);
+struct Wr(Arc<TcpStream>);
+
+impl AsyncRead for Rd {
+    fn poll_read(&mut self, cx: &mut task::Context, dst: &mut [u8]) -> Poll<usize, io::Error> {
+        <&TcpStream>::poll_read(&mut &*self.0, cx, dst)
+    }
+}
+
+impl AsyncWrite for Wr {
+    fn poll_write(&mut self, cx: &mut task::Context, src: &[u8]) -> Poll<usize, io::Error> {
+        <&TcpStream>::poll_write(&mut &*self.0, cx, src)
+    }
+
+    fn poll_flush(&mut self, _cx: &mut task::Context) -> Poll<(), io::Error> {
+        Ok(().into())
+    }
+
+    fn poll_close(&mut self, _cx: &mut task::Context) -> Poll<(), io::Error> {
+        Ok(().into())
+    }
+}
+
+#[test]
+fn hammer_split() {
+    const N: usize = 100;
+
+    let _ = env_logger::init();
+
+    let srv = t!(TcpListener::bind(&"127.0.0.1:0".parse().unwrap()));
+    let addr = t!(srv.local_addr());
+
+    let mut rt = Runtime::new().unwrap();
+
+    fn split(socket: TcpStream) {
+        let socket = Arc::new(socket);
+        let rd = Rd(socket.clone());
+        let wr = Wr(socket);
+
+        let rd = rd.read(vec![0; 1])
+            .map(|_| ())
+            .map_err(|e| panic!("read error = {:?}", e));
+
+        let wr = wr.write_all(b"1")
+            .map(|_| ())
+            .map_err(|e| panic!("write error = {:?}", e));
+
+        tokio::spawn2(rd);
+        tokio::spawn2(wr);
+    }
+
+    rt.spawn2({
+        srv.incoming()
+            .map_err(|e| panic!("accept error = {:?}", e))
+            .take(N as u64)
+            .for_each(|socket| {
+                split(socket);
+                Ok(())
+            })
+            .map(|_| ())
+    });
+
+    for _ in 0..N {
+        rt.spawn2({
+            TcpStream::connect(&addr)
+                .map_err(|e| panic!("connect error = {:?}", e))
+                .map(|socket| split(socket))
+        });
+    }
+
+    futures::Future::wait(rt.shutdown_on_idle()).unwrap();
+}

tests/tcp2.rs

@@ -0,0 +1,136 @@
+#![cfg(feature = "unstable-futures")]
+
+// This test is the same as `tcp.rs`, but ported to futures 0.2
+
+extern crate env_logger;
+extern crate tokio;
+extern crate mio;
+extern crate futures2;
+
+use std::{net, thread};
+use std::sync::mpsc::channel;
+
+use tokio::net::{TcpListener, TcpStream};
+use futures2::executor::block_on;
+use futures2::prelude::*;
+
+macro_rules! t {
+    ($e:expr) => (match $e {
+        Ok(e) => e,
+        Err(e) => panic!("{} failed with {:?}", stringify!($e), e),
+    })
+}
+
+#[test]
+fn connect() {
+    drop(env_logger::init());
+    let srv = t!(net::TcpListener::bind("127.0.0.1:0"));
+    let addr = t!(srv.local_addr());
+    let t = thread::spawn(move || {
+        t!(srv.accept()).0
+    });
+
+    let stream = TcpStream::connect(&addr);
+    let mine = t!(block_on(stream));
+    let theirs = t.join().unwrap();
+
+    assert_eq!(t!(mine.local_addr()), t!(theirs.peer_addr()));
+    assert_eq!(t!(theirs.local_addr()), t!(mine.peer_addr()));
+}
+
+#[test]
+fn accept() {
+    drop(env_logger::init());
+    let srv = t!(TcpListener::bind(&t!("127.0.0.1:0".parse())));
+    let addr = t!(srv.local_addr());
+
+    let (tx, rx) = channel();
+    let client = srv.incoming().map(move |t| {
+        tx.send(()).unwrap();
+        t
+    }).into_future().map_err(|e| e.0);
+    assert!(rx.try_recv().is_err());
+    let t = thread::spawn(move || {
+        net::TcpStream::connect(&addr).unwrap()
+    });
+
+    let (mine, _remaining) = t!(block_on(client));
+    let mine = mine.unwrap();
+    let theirs = t.join().unwrap();
+
+    assert_eq!(t!(mine.local_addr()), t!(theirs.peer_addr()));
+    assert_eq!(t!(theirs.local_addr()), t!(mine.peer_addr()));
+}
+
+#[test]
+fn accept2() {
+    drop(env_logger::init());
+    let srv = t!(TcpListener::bind(&t!("127.0.0.1:0".parse())));
+    let addr = t!(srv.local_addr());
+
+    let t = thread::spawn(move || {
+        net::TcpStream::connect(&addr).unwrap()
+    });
+
+    let (tx, rx) = channel();
+    let client = srv.incoming().map(move |t| {
+        tx.send(()).unwrap();
+        t
+    }).into_future().map_err(|e| e.0);
+    assert!(rx.try_recv().is_err());
+
+    let (mine, _remaining) = t!(block_on(client));
+    mine.unwrap();
+    t.join().unwrap();
+}
+
+#[cfg(unix)]
+mod unix {
+    use tokio::net::TcpStream;
+    use tokio::prelude::*;
+
+    use env_logger;
+    use futures2::future;
+    use futures2::executor::block_on;
+    use futures2::io::AsyncRead;
+    use mio::unix::UnixReady;
+
+    use std::{net, thread};
+    use std::time::Duration;
+
+    #[test]
+    fn poll_hup() {
+        drop(env_logger::init());
+
+        let srv = t!(net::TcpListener::bind("127.0.0.1:0"));
+        let addr = t!(srv.local_addr());
+        let t = thread::spawn(move || {
+            let mut client = t!(srv.accept()).0;
+            client.write(b"hello world").unwrap();
+            thread::sleep(Duration::from_millis(200));
+        });
+
+        let mut stream = t!(block_on(TcpStream::connect(&addr)));
+
+        // Poll for HUP before reading.
+        block_on(future::poll_fn(|cx| {
+            stream.poll_read_ready2(cx, UnixReady::hup().into())
+        })).unwrap();
+
+        // Same for write half
+        block_on(future::poll_fn(|cx| {
+            stream.poll_write_ready2(cx)
+        })).unwrap();
+
+        let mut buf = vec![0; 11];
+
+        // Read the data
+        block_on(future::poll_fn(|cx| {
+            stream.poll_read(cx, &mut buf)
+        })).unwrap();
+
+        assert_eq!(b"hello world", &buf[..]);
+
+        t.join().unwrap();
+    }
+}

tokio-tcp/src/stream.rs

@@ -139,6 +139,14 @@ impl TcpStream {
         self.io.poll_read_ready(mask)
     }
 
+    /// Like `poll_read_ready`, but compatible with futures 0.2
+    #[cfg(feature = "unstable-futures")]
+    pub fn poll_read_ready2(&self, cx: &mut futures2::task::Context, mask: mio::Ready)
+        -> futures2::Poll<mio::Ready, io::Error>
+    {
+        self.io.poll_read_ready2(cx, mask)
+    }
+
     /// Check the TCP stream's write readiness state.
     ///
     /// This always checks for writable readiness and also checks for HUP
@@ -160,6 +168,14 @@ impl TcpStream {
         self.io.poll_write_ready()
     }
 
+    /// Like `poll_write_ready`, but compatible with futures 0.2.
+    #[cfg(feature = "unstable-futures")]
+    pub fn poll_write_ready2(&self, cx: &mut futures2::task::Context)
+        -> futures2::Poll<mio::Ready, io::Error>
+    {
+        self.io.poll_write_ready2(cx)
+    }
+
     /// Returns the local address that this stream is bound to.
     pub fn local_addr(&self) -> io::Result<SocketAddr> {
         self.io.get_ref().local_addr()