Migrate to tokio 0.2 and futures 0.3

Cargo.toml

@@ -15,8 +15,8 @@ edition = "2018"
 github-actions = { repository = "quininer/tokio-rustls", workflow = "ci" }
 
 [dependencies]
-tokio-io = "=0.2.0-alpha.6"
+tokio = "0.2.0"
-futures-core-preview = "=0.3.0-alpha.19"
+futures-core = "0.3.1"
 rustls = "0.16"
 webpki = "0.21"
 
@@ -25,7 +25,7 @@ early-data = []
 dangerous_configuration = ["rustls/dangerous_configuration"]
 
 [dev-dependencies]
-tokio = "=0.2.0-alpha.6"
+tokio = { version = "0.2.0", features = ["macros", "net", "io-util", "rt-core", "time"] }
-futures-util-preview = "0.3.0-alpha.19"
+futures-util = "0.3.1"
 lazy_static = "1"
 webpki-roots = "0.18"

src/client.rs

@@ -69,7 +69,7 @@ impl<IO> AsyncRead for TlsStream<IO>
 where
     IO: AsyncRead + AsyncWrite + Unpin,
 {
-    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool {
+    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [std::mem::MaybeUninit<u8>]) -> bool {
         self.io.prepare_uninitialized_buffer(buf)
     }
 

src/common/mod.rs

@@ -3,7 +3,7 @@ use std::task::{ Poll, Context };
 use std::marker::Unpin;
 use std::io::{ self, Read, Write };
 use rustls::Session;
-use tokio_io::{ AsyncRead, AsyncWrite };
+use tokio::io::{ AsyncRead, AsyncWrite };
 use futures_core as futures;
 
 

src/common/test_stream.rs

@@ -4,7 +4,7 @@ use std::task::{ Poll, Context };
 use futures_core::ready;
 use futures_util::future::poll_fn;
 use futures_util::task::noop_waker_ref;
-use tokio_io::{ AsyncRead, AsyncWrite, AsyncReadExt, AsyncWriteExt };
+use tokio::io::{ AsyncRead, AsyncWrite, AsyncReadExt, AsyncWriteExt };
 use std::io::{ self, Read, Write, BufReader, Cursor };
 use webpki::DNSNameRef;
 use rustls::internal::pemfile::{ certs, rsa_private_keys };

src/lib.rs

@@ -10,7 +10,7 @@ use std::sync::Arc;
 use std::future::Future;
 use std::task::{ Context, Poll };
 use futures_core as futures;
-use tokio_io::{ AsyncRead, AsyncWrite };
+use tokio::io::{ AsyncRead, AsyncWrite };
 use webpki::DNSNameRef;
 use rustls::{ ClientConfig, ClientSession, ServerConfig, ServerSession, Session };
 use common::Stream;

src/server.rs

@@ -67,7 +67,7 @@ impl<IO> AsyncRead for TlsStream<IO>
 where
     IO: AsyncRead + AsyncWrite + Unpin,
 {
-    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool {
+    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [std::mem::MaybeUninit<u8>]) -> bool {
         self.io.prepare_uninitialized_buffer(buf)
     }
 

tests/test.rs

@@ -3,12 +3,12 @@ use std::io::{ BufReader, Cursor };
 use std::sync::Arc;
 use std::sync::mpsc::channel;
 use std::net::SocketAddr;
+use futures_util::future::TryFutureExt;
 use lazy_static::lazy_static;
 use tokio::prelude::*;
-use tokio::runtime::current_thread;
+use tokio::runtime;
 use tokio::net::{ TcpListener, TcpStream };
-use tokio::io::split;
+use tokio::io::{copy, split};
-use futures_util::try_future::TryFutureExt;
 use rustls::{ ServerConfig, ClientConfig };
 use rustls::internal::pemfile::{ certs, rsa_private_keys };
 use tokio_rustls::{ TlsConnector, TlsAcceptor };
@@ -30,32 +30,36 @@ lazy_static!{
         let (send, recv) = channel();
 
         thread::spawn(move || {
-            let mut runtime = current_thread::Runtime::new().unwrap();
+            let mut runtime = runtime::Builder::new()
-            let handle = runtime.handle();
+                .basic_scheduler()
+                .enable_io()
+                .build()
+                .unwrap();
+
+            let handle = runtime.handle().clone();
 
             let done = async move {
                 let addr = SocketAddr::from(([127, 0, 0, 1], 0));
-                let listener = TcpListener::bind(&addr).await?;
+                let mut listener = TcpListener::bind(&addr).await?;
 
                 send.send(listener.local_addr()?).unwrap();
 
-                let mut incoming = listener.incoming();
+                loop {
-                while let Some(stream) = incoming.next().await {
+                    let (stream, _) = listener.accept().await?;
+
                     let acceptor = acceptor.clone();
                     let fut = async move {
-                        let stream = acceptor.accept(stream?).await?;
+                        let stream = acceptor.accept(stream).await?;
 
                         let (mut reader, mut writer) = split(stream);
-                        reader.copy(&mut writer).await?;
+                        copy(&mut reader, &mut writer).await?;
 
                         Ok(()) as io::Result<()>
                     }.unwrap_or_else(|err| eprintln!("server: {:?}", err));
 
-                    handle.spawn(fut).unwrap();
+                    handle.spawn(fut);
                 }
-
+            }.unwrap_or_else(|err: io::Error| eprintln!("server: {:?}", err));
-                Ok(()) as io::Result<()>
-            }.unwrap_or_else(|err| eprintln!("server: {:?}", err));
 
             runtime.block_on(done);
         });
@@ -95,8 +99,7 @@ async fn pass() -> io::Result<()> {
     // TcpStream::bind now returns a future it creates a race
     // condition until its ready sometimes.
     use std::time::*;
-    let deadline = Instant::now() + Duration::from_secs(1);
+    tokio::time::delay_for(Duration::from_secs(1)).await;
-    tokio::timer::delay(deadline);
 
     let mut config = ClientConfig::new();
     let mut chain = BufReader::new(Cursor::new(chain));