fix: futures_impl

Cargo.toml

@@ -15,23 +15,23 @@ travis-ci = { repository = "quininer/tokio-rustls" }
 appveyor = { repository = "quininer/tokio-rustls" }
 
 [dependencies]
-futures = { version = "0.2.0-alpha", optional = true }
+futures = { version = "0.2.0-beta", optional = true }
 tokio = { version = "0.1", optional = true }
 rustls = "0.12"
 webpki = "0.18.0-alpha"
 
 [dev-dependencies]
 tokio = "0.1"
-tokio-io = "0.1"
+# tokio-io = "0.1"
-tokio-core = "0.1"
+# tokio-core = "0.1"
-tokio-file-unix = "0.4"
+# tokio-file-unix = "0.4"
 clap = "2.26"
 webpki-roots = "0.14"
 
 [features]
-default = [ "tokio" ]
+# default = [ "tokio" ]
-# unstable-futures = [ "futures", "tokio/unstable-futures" ]
+default = [ "unstable-futures", "tokio" ]
-# default = [ "unstable-futures", "tokio" ]
+unstable-futures = [ "futures", "tokio/unstable-futures" ]
 
 [patch.crates-io]
-# tokio = { path = "../ref/tokio" }
+tokio = { path = "../ref/tokio" }

src/futures_impl.rs

@@ -80,19 +80,13 @@ impl<S, C> Future for MidHandshake<S, C>
             let stream = self.inner.as_mut().unwrap();
             if !stream.session.is_handshaking() { break };
 
-            let mut taskio = TaskStream { io: &mut stream.io, task: ctx };
+            let (io, session) = stream.get_mut();
+            let mut taskio = TaskStream { io, task: ctx };
 
-            match TlsStream::do_io(&mut stream.session, &mut taskio, &mut stream.eof) {
+            match session.complete_io(&mut taskio) {
-                Ok(()) => match (stream.eof, stream.session.is_handshaking()) {
+                Ok(_) => (),
-                    (true, true) => return Err(io::ErrorKind::UnexpectedEof.into()),
+                Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => return Ok(Async::Pending),
-                    (false, true) => continue,
+                Err(e) => return Err(e)
-                    (..) => break
-                },
-                Err(e) => match (e.kind(), stream.session.is_handshaking()) {
-                    (io::ErrorKind::WouldBlock, true) => return Ok(Async::Pending),
-                    (io::ErrorKind::WouldBlock, false) => break,
-                    (..) => return Err(e)
-                }
             }
         }
 
@@ -106,9 +100,16 @@ impl<S, C> AsyncRead for TlsStream<S, C>
         C: Session
 {
     fn poll_read(&mut self, ctx: &mut Context, buf: &mut [u8]) -> Poll<usize, Error> {
-        let mut taskio = TaskStream { io: &mut self.io, task: ctx };
+        let (io, session) = self.get_mut();
-        // FIXME TlsStream + TaskStream
+        let mut taskio = TaskStream { io, task: ctx };
-        async!(from io::Read::read(&mut taskio, buf))
+        let mut stream = Stream::new(session, &mut taskio);
+
+        match io::Read::read(&mut stream, buf) {
+            Ok(n) => Ok(Async::Ready(n)),
+            Err(ref e) if e.kind() == io::ErrorKind::ConnectionAborted => Ok(Async::Ready(0)),
+            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => Ok(Async::Pending),
+            Err(e) => Err(e)
+        }
     }
 }
 
@@ -118,15 +119,19 @@ impl<S, C> AsyncWrite for TlsStream<S, C>
         C: Session
 {
     fn poll_write(&mut self, ctx: &mut Context, buf: &[u8]) -> Poll<usize, Error> {
-        let mut taskio = TaskStream { io: &mut self.io, task: ctx };
+        let (io, session) = self.get_mut();
-        // FIXME TlsStream + TaskStream
+        let mut taskio = TaskStream { io, task: ctx };
-        async!(from io::Write::write(&mut taskio, buf))
+        let mut stream = Stream::new(session, &mut taskio);
+
+        async!(from io::Write::write(&mut stream, buf))
     }
 
     fn poll_flush(&mut self, ctx: &mut Context) -> Poll<(), Error> {
-        let mut taskio = TaskStream { io: &mut self.io, task: ctx };
+        let (io, session) = self.get_mut();
-        // FIXME TlsStream + TaskStream
+        let mut taskio = TaskStream { io, task: ctx };
-        async!(from io::Write::flush(&mut taskio))
+        let mut stream = Stream::new(session, &mut taskio);
+
+        async!(from io::Write::flush(&mut stream))
     }
 
     fn poll_close(&mut self, ctx: &mut Context) -> Poll<(), Error> {
@@ -136,8 +141,9 @@ impl<S, C> AsyncWrite for TlsStream<S, C>
         }
 
         {
-            let mut taskio = TaskStream { io: &mut self.io, task: ctx };
+            let (io, session) = self.get_mut();
-            while TlsStream::do_write(&mut self.session, &mut taskio)? {};
+            let mut taskio = TaskStream { io, task: ctx };
+            session.complete_io(&mut taskio)?;
         }
 
         self.io.poll_close(ctx)

src/lib.rs

@@ -176,16 +176,6 @@ impl<S, C> TlsStream<S, C>
     }
 }
 
-macro_rules! try_ignore {
-    ( $r:expr ) => {
-        match $r {
-            Ok(_) => (),
-            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => (),
-            Err(e) => return Err(e)
-        }
-    }
-}
-
 impl<S, C> io::Read for TlsStream<S, C>
     where S: io::Read + io::Write, C: Session
 {

src/tokio_impl.rs

@@ -35,7 +35,9 @@ impl<S, C> Future for MidHandshake<S, C>
             let stream = self.inner.as_mut().unwrap();
             if !stream.session.is_handshaking() { break };
 
-            match stream.session.complete_io(&mut stream.io) {
+            let (io, session) = stream.get_mut();
+
+            match session.complete_io(io) {
                 Ok(_) => (),
                 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => return Ok(Async::NotReady),
                 Err(e) => return Err(e)
@@ -62,7 +64,7 @@ impl<S, C> AsyncWrite for TlsStream<S, C>
             self.session.send_close_notify();
             self.is_shutdown = true;
         }
-        while TlsStream::do_write(&mut self.session, &mut self.io)? {};
+        self.session.complete_io(&mut self.io)?;
         self.io.shutdown()
     }
 }

tests/test.rs

@@ -45,8 +45,7 @@ fn start_server(cert: Vec<Certificate>, rsa: PrivateKey) -> SocketAddr {
                         assert_eq!(buf, HELLO_WORLD);
                         aio::write_all(stream, HELLO_WORLD)
                     })
-                    .map(drop)
+                    .then(|_| Ok(()));
-                    .map_err(|err| panic!("{:?}", err));
 
                 tokio::spawn(done);
                 Ok(())
@@ -59,46 +58,6 @@ fn start_server(cert: Vec<Certificate>, rsa: PrivateKey) -> SocketAddr {
     recv.recv().unwrap()
 }
 
-#[cfg(feature = "unstable-futures")]
-fn start_server2(cert: Vec<Certificate>, rsa: PrivateKey) -> SocketAddr {
-    use futures::{ FutureExt, StreamExt };
-    use futures::io::{ AsyncReadExt, AsyncWriteExt };
-
-    let mut config = ServerConfig::new(rustls::NoClientAuth::new());
-    config.set_single_cert(cert, rsa);
-    let config = Arc::new(config);
-
-    let (send, recv) = channel();
-
-    thread::spawn(move || {
-        let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 0);
-        let listener = TcpListener::bind(&addr).unwrap();
-
-        send.send(listener.local_addr().unwrap()).unwrap();
-
-        let done = listener.incoming()
-            .for_each(move |stream| {
-                let done = config.accept_async(stream)
-                    .and_then(|stream| stream.read_exact(vec![0; HELLO_WORLD.len()]))
-                    .and_then(|(stream, buf)| {
-                        assert_eq!(buf, HELLO_WORLD);
-                        stream.write_all(HELLO_WORLD)
-                    })
-                    .map(drop)
-                    .map_err(|err| panic!("{:?}", err));
-
-                tokio::spawn2(done);
-                Ok(())
-            })
-            .map(drop)
-            .map_err(|err| panic!("{:?}", err));
-
-        tokio::runtime::run2(done);
-    });
-
-    recv.recv().unwrap()
-}
-
 fn start_client(addr: &SocketAddr, domain: &str, chain: Option<BufReader<Cursor<&str>>>) -> io::Result<()> {
     use tokio::prelude::*;
     use tokio::io as aio;
@@ -139,7 +98,7 @@ fn start_client2(addr: &SocketAddr, domain: &str, chain: Option<BufReader<Cursor
         .and_then(|stream| config.connect_async(domain, stream))
         .and_then(|stream| stream.write_all(HELLO_WORLD))
         .and_then(|(stream, _)| stream.read_exact(vec![0; HELLO_WORLD.len()]))
-        .and_then(|(stream, buf)| {
+        .and_then(|(_, buf)| {
             assert_eq!(buf, HELLO_WORLD);
             Ok(())
         });
@@ -165,7 +124,7 @@ fn pass2() {
     let mut keys = rsa_private_keys(&mut BufReader::new(Cursor::new(RSA))).unwrap();
     let chain = BufReader::new(Cursor::new(CHAIN));
 
-    let addr = start_server2(cert, keys.pop().unwrap());
+    let addr = start_server(cert, keys.pop().unwrap());
     start_client2(&addr, "localhost", Some(chain)).unwrap();
 }