Improve `poll_read_ready` implementation (#193)

This patch updates `poll_read_ready` to take a `mask` argument, enabling
the caller to specify the desired readiness. `need_read` is renamed to
`clear_read_ready` and also takes a mask.

This enables a caller to listen for HUP events without requiring reading
from the I/O resource.

1 files changed, 54 insertions, 5 deletions

diff --git a/tests/tcp.rs b/tests/tcp.rs
index 83cc425c..5694e8a6 100644
--- a/tests/tcp.rs
+++ b/tests/tcp.rs
@@ -1,14 +1,14 @@
 extern crate env_logger;
-extern crate futures;
 extern crate tokio;
+extern crate mio;
+extern crate futures;
 
-use std::net;
+use std::{net, thread};
 use std::sync::mpsc::channel;
-use std::thread;
 
-use futures::Future;
-use futures::stream::Stream;
 use tokio::net::{TcpListener, TcpStream};
+use tokio::prelude::*;
+
 
 macro_rules! t {
     ($e:expr) => (match $e {
@@ -79,3 +79,52 @@ fn accept2() {
     mine.unwrap();
     t.join().unwrap();
 }
+
+#[cfg(unix)]
+mod unix {
+    use tokio::net::TcpStream;
+    use tokio::prelude::*;
+
+    use env_logger;
+    use futures::future;
+    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!(TcpStream::connect(&addr).wait());
+
+        // Poll for HUP before reading.
+        future::poll_fn(|| {
+            stream.poll_read_ready(UnixReady::hup().into())
+        }).wait().unwrap();
+
+        // Same for write half
+        future::poll_fn(|| {
+            stream.poll_write_ready()
+        }).wait().unwrap();
+
+        let mut buf = vec![0; 11];
+
+        // Read the data
+        future::poll_fn(|| {
+            stream.poll_read(&mut buf)
+        }).wait().unwrap();
+
+        assert_eq!(b"hello world", &buf[..]);
+
+        t.join().unwrap();
+    }
+}