Switch to tokio-io 0.2

Cargo.toml

@@ -17,7 +17,8 @@ appveyor = { repository = "quininer/tokio-rustls" }
 
 [dependencies]
 smallvec = "0.6"
-futures = { package = "futures-preview", version = "0.3.0-alpha.16" }
+tokio-io = { git = "https://github.com/tokio-rs/tokio" }
+tokio-futures = { git = "https://github.com/tokio-rs/tokio" }
 rustls = "0.15"
 webpki = "0.19"
 
@@ -25,6 +26,6 @@ webpki = "0.19"
 early-data = []
 
 [dev-dependencies]
-romio = "0.3.0-alpha.8"
+tokio = { git = "https://github.com/tokio-rs/tokio" }
 lazy_static = "1"
 webpki-roots = "0.16"

src/client.rs

@@ -74,8 +74,8 @@ impl<IO> AsyncRead for TlsStream<IO>
 where
     IO: AsyncRead + AsyncWrite + Unpin,
 {
-    unsafe fn initializer(&self) -> Initializer {
-        Initializer::nop()
+    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool {
+        self.io.prepare_uninitialized_buffer(buf)
     }
 
     fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<io::Result<usize>> {
@@ -192,7 +192,7 @@ where
         stream.as_mut_pin().poll_flush(cx)
     }
 
-    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
         if self.state.writeable() {
             self.session.send_close_notify();
             self.state.shutdown_write();
@@ -201,6 +201,6 @@ where
         let this = self.get_mut();
         let mut stream = Stream::new(&mut this.io, &mut this.session)
             .set_eof(!this.state.readable());
-        stream.as_mut_pin().poll_close(cx)
+        stream.as_mut_pin().poll_shutdown(cx)
     }
 }

src/common/mod.rs

@@ -1,11 +1,10 @@
 use std::pin::Pin;
-use std::task::Poll;
+use std::task::{ Poll, Context };
 use std::marker::Unpin;
-use std::io::{ self, Read };
-use rustls::{ Session, WriteV };
-use futures::task::Context;
-use futures::io::{ AsyncRead, AsyncWrite, IoSlice };
-use smallvec::SmallVec;
+use std::io::{ self, Read, Write };
+use rustls::Session;
+use tokio_io::{ AsyncRead, AsyncWrite };
+use tokio_futures as futures;
 
 
 pub struct Stream<'a, IO, S> {
@@ -154,27 +153,31 @@ impl<'a, IO: AsyncRead + AsyncWrite + Unpin, S: Session> Stream<'a, IO, S> {
 
 impl<'a, IO: AsyncRead + AsyncWrite + Unpin, S: Session> WriteTls<IO, S> for Stream<'a, IO, S> {
     fn write_tls(&mut self, cx: &mut Context) -> io::Result<usize> {
-        struct Writer<'a, 'b, IO> {
-            io: &'a mut IO,
+        // TODO writev
+
+        struct Writer<'a, 'b, T> {
+            io: &'a mut T,
             cx: &'a mut Context<'b>
         }
 
-        impl<'a, 'b, IO: AsyncWrite + Unpin> WriteV for Writer<'a, 'b, IO> {
-            fn writev(&mut self, vbytes: &[&[u8]]) -> io::Result<usize> {
-                let vbytes = vbytes
-                    .into_iter()
-                    .map(|v| IoSlice::new(v))
-                    .collect::<SmallVec<[IoSlice<'_>; 64]>>();
+        impl<'a, 'b, T: AsyncWrite + Unpin> Write for Writer<'a, 'b, T> {
+            fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+                match Pin::new(&mut self.io).poll_write(self.cx, buf) {
+                    Poll::Ready(result) => result,
+                    Poll::Pending => Err(io::ErrorKind::WouldBlock.into())
+                }
+            }
 
-                match Pin::new(&mut self.io).poll_write_vectored(self.cx, &vbytes) {
+            fn flush(&mut self) -> io::Result<()> {
+                match Pin::new(&mut self.io).poll_flush(self.cx) {
                     Poll::Ready(result) => result,
                     Poll::Pending => Err(io::ErrorKind::WouldBlock.into())
                 }
             }
         }
 
-        let mut vecio = Writer { io: self.io, cx };
-        self.session.writev_tls(&mut vecio)
+        let mut writer = Writer { io: self.io, cx };
+        self.session.write_tls(&mut writer)
     }
 }
 
@@ -240,13 +243,13 @@ impl<'a, IO: AsyncRead + AsyncWrite + Unpin, S: Session> AsyncWrite for Stream<'
         Pin::new(&mut this.io).poll_flush(cx)
     }
 
-    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
         let this = self.get_mut();
 
         while this.session.wants_write() {
             futures::ready!(this.complete_inner_io(cx, Focus::Writable))?;
         }
-        Pin::new(&mut this.io).poll_close(cx)
+        Pin::new(&mut this.io).poll_shutdown(cx)
     }
 }
 

src/lib.rs

@@ -12,7 +12,8 @@ use std::sync::Arc;
 use std::pin::Pin;
 use std::future::Future;
 use std::task::{ Poll, Context };
-use futures::io::{ AsyncRead, AsyncWrite, Initializer };
+use tokio_io::{ AsyncRead, AsyncWrite };
+use tokio_futures as futures;
 use rustls::{ ClientConfig, ClientSession, ServerConfig, ServerSession };
 use webpki::DNSNameRef;
 use common::Stream;

src/server.rs

@@ -67,8 +67,8 @@ impl<IO> AsyncRead for TlsStream<IO>
 where
     IO: AsyncRead + AsyncWrite + Unpin,
 {
-    unsafe fn initializer(&self) -> Initializer {
-        Initializer::nop()
+    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool {
+        self.io.prepare_uninitialized_buffer(buf)
     }
 
     fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<io::Result<usize>> {
@@ -119,7 +119,7 @@ where
         stream.as_mut_pin().poll_flush(cx)
     }
 
-    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
         if self.state.writeable() {
             self.session.send_close_notify();
             self.state.shutdown_write();
@@ -128,6 +128,6 @@ where
         let this = self.get_mut();
         let mut stream = Stream::new(&mut this.io, &mut this.session)
             .set_eof(!this.state.readable());
-        stream.as_mut_pin().poll_close(cx)
+        stream.as_mut_pin().poll_shutdown(cx)
     }
 }