diff --git a/src/executor/mod.rs b/src/executor/mod.rs index c69558032..52f43934e 100644 --- a/src/executor/mod.rs +++ b/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 + 'static + Send Spawn(()) } +/// Like `spawn`, but compatible with futures 0.2 +#[cfg(feature = "unstable-futures")] +pub fn spawn2(f: F) -> Spawn + where F: futures2::Future + '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(()) + } +} diff --git a/src/lib.rs b/src/lib.rs index 2512cacb4..7a10b3201 100644 --- a/src/lib.rs +++ b/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 { diff --git a/tests/echo2.rs b/tests/echo2.rs new file mode 100755 index 000000000..c05a5f478 --- /dev/null +++ b/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); +} diff --git a/tests/global2.rs b/tests/global2.rs new file mode 100755 index 000000000..2ad52b8b0 --- /dev/null +++ b/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::>(); + for thread in threads { + thread.join().unwrap(); + } +} + +struct Rd(Arc); +struct Wr(Arc); + +impl AsyncRead for Rd { + fn poll_read(&mut self, cx: &mut task::Context, dst: &mut [u8]) -> Poll { + <&TcpStream>::poll_read(&mut &*self.0, cx, dst) + } +} + +impl AsyncWrite for Wr { + fn poll_write(&mut self, cx: &mut task::Context, src: &[u8]) -> Poll { + <&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(); +} diff --git a/tests/tcp2.rs b/tests/tcp2.rs new file mode 100755 index 000000000..464939c8c --- /dev/null +++ b/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(); + } +} diff --git a/tokio-tcp/src/stream.rs b/tokio-tcp/src/stream.rs index c0ae87659..d402952c5 100644 --- a/tokio-tcp/src/stream.rs +++ b/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 + { + 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 + { + self.io.poll_write_ready2(cx) + } + /// Returns the local address that this stream is bound to. pub fn local_addr(&self) -> io::Result { self.io.get_ref().local_addr()