Update actix-server to tokio 0.2

actix-server/Cargo.toml

@@ -38,7 +38,7 @@ mio = "0.6.19"
 net2 = "0.2"
 futures = "0.3.1"
 slab = "0.4"
-tokio-net = { version = "0.2.0-alpha.6", features = ["signal", "tcp", "uds"] }
+tokio = { version = "0.2", features = ["net", "uds", "signal"] }
 
 futures-core-preview = "0.3.0-alpha.19"
 
@@ -51,7 +51,7 @@ tokio-tls = { version = "0.3.0-alpha.6", optional = true }
 
 # openssl
 open-ssl = { version = "0.10", package = "openssl", optional = true }
-tokio-openssl = { version = "0.4.0-alpha.6", optional = true }
+tokio-openssl = { version = "0.4.0", optional = true }
 
 # rustls
 rust-tls = { version = "0.16.0", package = "rustls", optional = true }
@@ -61,6 +61,6 @@ webpki = { version = "0.21", optional = true }
 webpki-roots = { version = "0.17", optional = true }
 
 [dev-dependencies]
-bytes = "0.4"
+bytes = "0.5"
 actix-codec = "0.2.0-alpha.1"
 env_logger = "0.6"

actix-server/src/accept.rs

@@ -2,7 +2,7 @@ use std::sync::mpsc as sync_mpsc;
 use std::time::{Duration, Instant};
 use std::{io, thread};
 
-use actix_rt::time::delay;
+use actix_rt::time::delay_for;
 use actix_rt::System;
 use futures::FutureExt;
 use log::{error, info};
@@ -442,7 +442,7 @@ impl Accept {
                         let r = self.timer.1.clone();
                         System::current().arbiter().send(
                             async move {
-                                delay(Instant::now() + Duration::from_millis(510)).await;
+                                delay_for(Duration::from_millis(510)).await;
                                 let _ = r.set_readiness(mio::Ready::readable());
                             }
                                 .boxed(),

actix-server/src/builder.rs

@@ -1,9 +1,9 @@
 use std::pin::Pin;
 use std::task::{Context, Poll};
-use std::time::{Duration, Instant};
+use std::time::Duration;
 use std::{io, mem, net};
 
-use actix_rt::{spawn, time::delay, Arbiter, System};
+use actix_rt::{spawn, time::delay_for, Arbiter, System};
 use futures::channel::mpsc::{unbounded, UnboundedReceiver};
 use futures::channel::oneshot;
 use futures::future::ready;
@@ -12,7 +12,7 @@ use futures::{ready, Future, FutureExt, Stream, StreamExt};
 use log::{error, info};
 use net2::TcpBuilder;
 use num_cpus;
-use tokio_net::tcp::TcpStream;
+use tokio::net::TcpStream;
 
 use crate::accept::{AcceptLoop, AcceptNotify, Command};
 use crate::config::{ConfiguredService, ServiceConfig};
@@ -191,7 +191,7 @@ impl ServerBuilder {
     /// Add new unix domain service to the server.
     pub fn bind_uds<F, U, N>(self, name: N, addr: U, factory: F) -> io::Result<Self>
     where
-        F: ServiceFactory<tokio_net::uds::UnixStream>,
+        F: ServiceFactory<tokio::net::UnixStream>,
         N: AsRef<str>,
         U: AsRef<std::path::Path>,
     {
@@ -221,7 +221,7 @@ impl ServerBuilder {
         factory: F,
     ) -> io::Result<Self>
     where
-        F: ServiceFactory<tokio_net::uds::UnixStream>,
+        F: ServiceFactory<tokio::net::UnixStream>,
     {
         use std::net::{IpAddr, Ipv4Addr, SocketAddr};
         let token = self.token.next();
@@ -406,7 +406,7 @@ impl ServerBuilder {
                                 if exit {
                                     spawn(
                                         async {
-                                            delay(Instant::now() + Duration::from_millis(300))
+                                            delay_for(Duration::from_millis(300))
                                                 .await;
                                             System::current().stop();
                                         }
@@ -420,7 +420,7 @@ impl ServerBuilder {
                     // we need to stop system if server was spawned
                     if self.exit {
                         spawn(
-                            delay(Instant::now() + Duration::from_millis(300)).then(|_| {
+                            delay_for(Duration::from_millis(300)).then(|_| {
                                 System::current().stop();
                                 ready(())
                             }),

actix-server/src/config.rs

@@ -5,7 +5,7 @@ use actix_server_config::{Io, ServerConfig};
 use actix_service as actix;
 use futures::future::{Future, FutureExt, LocalBoxFuture};
 use log::error;
-use tokio_net::tcp::TcpStream;
+use tokio::net::TcpStream;
 
 use super::builder::bind_addr;
 use super::service::{

actix-server/src/signals.rs

@@ -3,8 +3,6 @@ use std::io;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
-use futures_core::stream::Stream;
-
 use crate::server::Server;
 
 /// Different types of process signals
@@ -25,7 +23,7 @@ pub(crate) struct Signals {
     #[cfg(not(unix))]
     stream: tokio_net::signal::CtrlC,
     #[cfg(unix)]
-    streams: Vec<(Signal, tokio_net::signal::unix::Signal)>,
+    streams: Vec<(Signal, tokio::signal::unix::Signal)>,
 }
 
 impl Signals {
@@ -39,7 +37,7 @@ impl Signals {
 
             #[cfg(unix)]
             {
-                use tokio_net::signal::unix;
+                use tokio::signal::unix;
 
                 let mut streams = Vec::new();
 
@@ -80,7 +78,7 @@ impl Future for Signals {
         {
             for idx in 0..self.streams.len() {
                 loop {
-                    match Pin::new(&mut self.streams[idx].1).poll_next(cx) {
+                    match Pin::new(&mut self.streams[idx].1).poll_recv(cx) {
                         Poll::Ready(None) => return Poll::Ready(()),
                         Poll::Pending => break,
                         Poll::Ready(Some(_)) => {

actix-server/src/socket.rs

@@ -1,8 +1,7 @@
 use std::{fmt, io, net};
 
 use actix_codec::{AsyncRead, AsyncWrite};
-use tokio_net::driver::Handle;
+use tokio::net::TcpStream;
-use tokio_net::tcp::TcpStream;
 
 pub(crate) enum StdListener {
     Tcp(net::TcpListener),
@@ -151,7 +150,7 @@ pub trait FromStream: AsyncRead + AsyncWrite + Sized {
 impl FromStream for TcpStream {
     fn from_stdstream(sock: StdStream) -> io::Result<Self> {
         match sock {
-            StdStream::Tcp(stream) => TcpStream::from_std(stream, &Handle::default()),
+            StdStream::Tcp(stream) => TcpStream::from_std(stream),
             #[cfg(all(unix))]
             StdStream::Uds(_) => {
                 panic!("Should not happen, bug in server impl");
@@ -161,12 +160,12 @@ impl FromStream for TcpStream {
 }
 
 #[cfg(all(unix))]
-impl FromStream for tokio_net::uds::UnixStream {
+impl FromStream for tokio::net::UnixStream {
     fn from_stdstream(sock: StdStream) -> io::Result<Self> {
         match sock {
             StdStream::Tcp(_) => panic!("Should not happen, bug in server impl"),
             StdStream::Uds(stream) => {
-                tokio_net::uds::UnixStream::from_std(stream, &Handle::default())
+                tokio::net::UnixStream::from_std(stream)
             }
         }
     }

actix-server/src/worker.rs

@@ -4,7 +4,7 @@ use std::sync::Arc;
 use std::task::{Context, Poll};
 use std::{mem, time};
 
-use actix_rt::time::{delay, Delay};
+use actix_rt::time::{delay_for, Delay};
 use actix_rt::{spawn, Arbiter};
 use futures::channel::mpsc::{UnboundedReceiver, UnboundedSender};
 use futures::channel::oneshot;
@@ -277,8 +277,8 @@ impl Future for Worker {
                 if num != 0 {
                     info!("Graceful worker shutdown, {} connections", num);
                     self.state = WorkerState::Shutdown(
-                        delay(time::Instant::now() + time::Duration::from_secs(1)),
+                        delay_for(time::Duration::from_secs(1)),
-                        delay(time::Instant::now() + self.shutdown_timeout),
+                        delay_for(self.shutdown_timeout),
                         result,
                     );
                 } else {
@@ -397,7 +397,7 @@ impl Future for Worker {
                 match Pin::new(&mut t1).poll(cx) {
                     Poll::Pending => (),
                     Poll::Ready(_) => {
-                        t1 = delay(time::Instant::now() + time::Duration::from_secs(1));
+                        t1 = delay_for(time::Duration::from_secs(1));
                         let _ = Pin::new(&mut t1).poll(cx);
                     }
                 }

actix-server/tests/test_server.rs

@@ -8,7 +8,7 @@ use actix_service::{factory_fn_cfg, service_fn, service_fn2};
 use bytes::Bytes;
 use futures::{future::ok, SinkExt};
 use net2::TcpBuilder;
-use tokio_net::tcp::TcpStream;
+use tokio::net::TcpStream;
 
 fn unused_addr() -> net::SocketAddr {
     let addr: net::SocketAddr = "127.0.0.1:0".parse().unwrap();