Update tokio to 0.3

Cargo.toml

@@ -21,7 +21,7 @@ stream = []
 log = "0.4"
 futures-util = { version = "0.3", default-features = false, features = ["async-await", "sink", "std"] }
 pin-project = "0.4.17"
-tokio = { version = "0.2", default-features = false, features = ["io-util"] }
+tokio = { version = "0.3", default-features = false, features = ["io-util"] }
 
 [dependencies.tungstenite]
 version = "0.11.1"
@@ -33,10 +33,10 @@ version = "0.2.0"
 
 [dependencies.tokio-native-tls]
 optional = true
-version = "0.1"
+version = "0.2"
 
 [dev-dependencies]
 futures-channel = "0.3"
-tokio = { version = "0.2", default-features = false, features = ["io-std", "macros", "stream", "time"] }
+tokio = { version = "0.3", default-features = false, features = ["io-std", "macros", "rt-multi-thread", "stream", "time"] }
 url = "2.0.0"
 env_logger = "0.7"

examples/autobahn-server.rs

@@ -34,7 +34,7 @@ async fn main() {
     env_logger::init();
 
     let addr = "127.0.0.1:9002";
-    let mut listener = TcpListener::bind(&addr).await.expect("Can't listen");
+    let listener = TcpListener::bind(&addr).await.expect("Can't listen");
     info!("Listening on: {}", addr);
 
     while let Ok((stream, _)) = listener.accept().await {

examples/echo-server.rs

@@ -23,7 +23,7 @@ async fn main() -> Result<(), Error> {
 
     // Create the event loop and TCP listener we'll accept connections on.
     let try_socket = TcpListener::bind(&addr).await;
-    let mut listener = try_socket.expect("Failed to bind");
+    let listener = try_socket.expect("Failed to bind");
     info!("Listening on: {}", addr);
 
     while let Ok((stream, _)) = listener.accept().await {

examples/interval-server.rs

@@ -59,7 +59,7 @@ async fn main() {
     env_logger::init();
 
     let addr = "127.0.0.1:9002";
-    let mut listener = TcpListener::bind(&addr).await.expect("Can't listen");
+    let listener = TcpListener::bind(&addr).await.expect("Can't listen");
     info!("Listening on: {}", addr);
 
     while let Ok((stream, _)) = listener.accept().await {

examples/server.rs

@@ -92,7 +92,7 @@ async fn main() -> Result<(), IoError> {
 
     // Create the event loop and TCP listener we'll accept connections on.
     let try_socket = TcpListener::bind(&addr).await;
-    let mut listener = try_socket.expect("Failed to bind");
+    let listener = try_socket.expect("Failed to bind");
     println!("Listening on: {}", addr);
 
     // Let's spawn the handling of each connection in a separate task.

src/compat.rs

@@ -5,7 +5,7 @@ use std::task::{Context, Poll};
 
 use futures_util::task;
 use std::sync::Arc;
-use tokio::io::{AsyncRead, AsyncWrite};
+use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 use tungstenite::Error as WsError;
 
 pub(crate) enum ContextWaker {
@@ -145,15 +145,17 @@ where
 {
     fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
         trace!("{}:{} Read.read", file!(), line!());
+        let mut buf = ReadBuf::new(buf);
         match self.with_context(ContextWaker::Read, |ctx, stream| {
             trace!(
                 "{}:{} Read.with_context read -> poll_read",
                 file!(),
                 line!()
             );
-            stream.poll_read(ctx, buf)
+            stream.poll_read(ctx, &mut buf)
         }) {
-            Poll::Ready(r) => r,
+            Poll::Ready(Ok(_)) => Ok(buf.filled().len()),
+            Poll::Ready(Err(err)) => Err(err),
             Poll::Pending => Err(std::io::Error::from(std::io::ErrorKind::WouldBlock)),
         }
     }

src/stream.rs

@@ -7,7 +7,7 @@ use pin_project::pin_project;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
-use tokio::io::{AsyncRead, AsyncWrite};
+use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 
 /// Stream, either plain TCP or TLS.
 #[pin_project(project = StreamProj)]
@@ -22,8 +22,8 @@ impl<S: AsyncRead + Unpin, T: AsyncRead + Unpin> AsyncRead for Stream<S, T> {
     fn poll_read(
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
-        buf: &mut [u8],
-    ) -> Poll<std::io::Result<usize>> {
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<std::io::Result<()>> {
         match self.project() {
             StreamProj::Plain(ref mut s) => Pin::new(s).poll_read(cx, buf),
             StreamProj::Tls(ref mut s) => Pin::new(s).poll_read(cx, buf),

tests/communication.rs

@@ -30,7 +30,7 @@ async fn communication() {
     let (msg_tx, msg_rx) = futures_channel::oneshot::channel();
 
     let f = async move {
-        let mut listener = TcpListener::bind("0.0.0.0:12345").await.unwrap();
+        let listener = TcpListener::bind("0.0.0.0:12345").await.unwrap();
         info!("Server ready");
         con_tx.send(()).unwrap();
         info!("Waiting on next connection");
@@ -76,7 +76,7 @@ async fn split_communication() {
     let (msg_tx, msg_rx) = futures_channel::oneshot::channel();
 
     let f = async move {
-        let mut listener = TcpListener::bind("0.0.0.0:12346").await.unwrap();
+        let listener = TcpListener::bind("0.0.0.0:12346").await.unwrap();
         info!("Server ready");
         con_tx.send(()).unwrap();
         info!("Waiting on next connection");

tests/handshakes.rs

@@ -6,7 +6,7 @@ async fn handshakes() {
     let (tx, rx) = futures_channel::oneshot::channel();
 
     let f = async move {
-        let mut listener = TcpListener::bind("0.0.0.0:12345").await.unwrap();
+        let listener = TcpListener::bind("0.0.0.0:12345").await.unwrap();
         tx.send(()).unwrap();
         while let Ok((connection, _)) = listener.accept().await {
             let stream = accept_async(connection).await;