migrate to tokio

src/client/connector.rs

@@ -9,16 +9,16 @@ use actix::actors::{Connect as ResolveConnect, Connector, ConnectorError};
 use actix::fut::WrapFuture;
 use actix::registry::ArbiterService;
 use actix::{
-    fut, Actor, ActorFuture, ActorResponse, Arbiter, AsyncContext, Context,
-    ContextFutureSpawner, Handler, Message, Recipient, Supervised, Syn,
+    fut, Actor, ActorFuture, ActorResponse, AsyncContext, Context, ContextFutureSpawner,
+    Handler, Message, Recipient, Supervised, Syn,
 };
 
 use futures::task::{current as current_task, Task};
 use futures::unsync::oneshot;
 use futures::{Async, Future, Poll};
 use http::{Error as HttpError, HttpTryFrom, Uri};
-use tokio_core::reactor::Timeout;
 use tokio_io::{AsyncRead, AsyncWrite};
+use tokio_timer::Delay;
 
 #[cfg(feature = "alpn")]
 use openssl::ssl::{Error as OpensslError, SslConnector, SslMethod};
@@ -190,8 +190,8 @@ pub struct ClientConnector {
     available: HashMap<Key, VecDeque<Conn>>,
     to_close: Vec<Connection>,
     waiters: HashMap<Key, VecDeque<Waiter>>,
-    wait_timeout: Option<(Instant, Timeout)>,
-    paused: Option<Option<(Instant, Timeout)>>,
+    wait_timeout: Option<(Instant, Delay)>,
+    paused: Option<Option<(Instant, Delay)>>,
 }
 
 impl Actor for ClientConnector {
@@ -563,8 +563,7 @@ impl ClientConnector {
             }
         }
 
-        let mut timeout =
-            Timeout::new(time - Instant::now(), Arbiter::handle()).unwrap();
+        let mut timeout = Delay::new(time);
         let _ = timeout.poll();
         self.wait_timeout = Some((time, timeout));
     }
@@ -597,7 +596,7 @@ impl Handler<Pause> for ClientConnector {
     fn handle(&mut self, msg: Pause, _: &mut Self::Context) {
         if let Some(time) = msg.time {
             let when = Instant::now() + time;
-            let mut timeout = Timeout::new(time, Arbiter::handle()).unwrap();
+            let mut timeout = Delay::new(when);
             let _ = timeout.poll();
             self.paused = Some(Some((when, timeout)));
         } else if self.paused.is_none() {

src/client/mod.rs

@@ -10,7 +10,7 @@
 //! fn main() {
 //!     let sys = actix::System::new("test");
 //!
-//!     actix::Arbiter::handle().spawn({
+//!     actix::Arbiter::spawn({
 //!         client::get("http://www.rust-lang.org")   // <- Create request builder
 //!             .header("User-Agent", "Actix-web")
 //!             .finish().unwrap()
@@ -70,7 +70,7 @@ impl ResponseError for SendRequestError {
 /// fn main() {
 ///     let sys = actix::System::new("test");
 ///
-///     actix::Arbiter::handle().spawn({
+///     actix::Arbiter::spawn({
 ///         client::get("http://www.rust-lang.org")   // <- Create request builder
 ///             .header("User-Agent", "Actix-web")
 ///             .finish().unwrap()

src/client/pipeline.rs

@@ -2,9 +2,9 @@ use bytes::{Bytes, BytesMut};
 use futures::unsync::oneshot;
 use futures::{Async, Future, Poll};
 use http::header::CONTENT_ENCODING;
-use std::time::Duration;
+use std::time::{Duration, Instant};
 use std::{io, mem};
-use tokio_core::reactor::Timeout;
+use tokio_timer::Delay;
 
 use actix::prelude::*;
 
@@ -71,7 +71,7 @@ pub struct SendRequest {
     conn: Addr<Unsync, ClientConnector>,
     conn_timeout: Duration,
     wait_timeout: Duration,
-    timeout: Option<Timeout>,
+    timeout: Option<Delay>,
 }
 
 impl SendRequest {
@@ -108,7 +108,7 @@ impl SendRequest {
     /// Request timeout is the total time before a response must be received.
     /// Default value is 5 seconds.
     pub fn timeout(mut self, timeout: Duration) -> Self {
-        self.timeout = Some(Timeout::new(timeout, Arbiter::handle()).unwrap());
+        self.timeout = Some(Delay::new(Instant::now() + timeout));
         self
     }
 
@@ -174,7 +174,7 @@ impl Future for SendRequest {
                     };
 
                     let timeout = self.timeout.take().unwrap_or_else(|| {
-                        Timeout::new(Duration::from_secs(5), Arbiter::handle()).unwrap()
+                        Delay::new(Instant::now() + Duration::from_secs(5))
                     });
 
                     let pl = Box::new(Pipeline {
@@ -229,7 +229,7 @@ pub(crate) struct Pipeline {
     decompress: Option<PayloadStream>,
     should_decompress: bool,
     write_state: RunningState,
-    timeout: Option<Timeout>,
+    timeout: Option<Delay>,
 }
 
 enum IoBody {

src/client/request.rs

@@ -34,7 +34,7 @@ use httprequest::HttpRequest;
 /// fn main() {
 ///     let sys = actix::System::new("test");
 ///
-///     actix::Arbiter::handle().spawn({
+///     actix::Arbiter::spawn({
 ///         ClientRequest::get("http://www.rust-lang.org") // <- Create request builder
 ///             .header("User-Agent", "Actix-web")
 ///             .finish().unwrap()

src/error.rs

@@ -16,6 +16,7 @@ use http_range::HttpRangeParseError;
 use httparse;
 use serde::de::value::Error as DeError;
 use serde_json::error::Error as JsonError;
+use tokio_timer::Error as TimerError;
 pub use url::ParseError as UrlParseError;
 
 // re-exports
@@ -126,6 +127,9 @@ impl From<failure::Error> for Error {
 /// `InternalServerError` for `JsonError`
 impl ResponseError for JsonError {}
 
+/// `InternalServerError` for `TimerError`
+impl ResponseError for TimerError {}
+
 /// `InternalServerError` for `UrlParseError`
 impl ResponseError for UrlParseError {}
 

src/lib.rs

@@ -112,8 +112,11 @@ extern crate mio;
 extern crate net2;
 extern crate rand;
 extern crate slab;
-extern crate tokio_core;
+extern crate tokio;
 extern crate tokio_io;
+extern crate tokio_reactor;
+extern crate tokio_tcp;
+extern crate tokio_timer;
 extern crate url;
 #[macro_use]
 extern crate serde;

src/multipart.rs

@@ -663,7 +663,7 @@ mod tests {
     use bytes::Bytes;
     use futures::future::{lazy, result};
     use payload::{Payload, PayloadWriter};
-    use tokio_core::reactor::Core;
+    use tokio::runtime::current_thread::Runtime;
 
     #[test]
     fn test_boundary() {
@@ -710,9 +710,9 @@ mod tests {
 
     #[test]
     fn test_multipart() {
-        Core::new()
+        Runtime::new()
             .unwrap()
-            .run(lazy(|| {
+            .block_on(lazy(|| {
                 let (mut sender, payload) = Payload::new(false);
 
                 let bytes = Bytes::from(

src/payload.rs

@@ -524,7 +524,7 @@ mod tests {
     use failure::Fail;
     use futures::future::{lazy, result};
     use std::io;
-    use tokio_core::reactor::Core;
+    use tokio::runtime::current_thread::Runtime;
 
     #[test]
     fn test_error() {
@@ -542,9 +542,9 @@ mod tests {
 
     #[test]
     fn test_basic() {
-        Core::new()
+        Runtime::new()
             .unwrap()
-            .run(lazy(|| {
+            .block_on(lazy(|| {
                 let (_, payload) = Payload::new(false);
                 let mut payload = PayloadHelper::new(payload);
 
@@ -559,9 +559,9 @@ mod tests {
 
     #[test]
     fn test_eof() {
-        Core::new()
+        Runtime::new()
             .unwrap()
-            .run(lazy(|| {
+            .block_on(lazy(|| {
                 let (mut sender, payload) = Payload::new(false);
                 let mut payload = PayloadHelper::new(payload);
 
@@ -584,9 +584,9 @@ mod tests {
 
     #[test]
     fn test_err() {
-        Core::new()
+        Runtime::new()
             .unwrap()
-            .run(lazy(|| {
+            .block_on(lazy(|| {
                 let (mut sender, payload) = Payload::new(false);
                 let mut payload = PayloadHelper::new(payload);
 
@@ -602,9 +602,9 @@ mod tests {
 
     #[test]
     fn test_readany() {
-        Core::new()
+        Runtime::new()
             .unwrap()
-            .run(lazy(|| {
+            .block_on(lazy(|| {
                 let (mut sender, payload) = Payload::new(false);
                 let mut payload = PayloadHelper::new(payload);
 
@@ -631,9 +631,9 @@ mod tests {
 
     #[test]
     fn test_readexactly() {
-        Core::new()
+        Runtime::new()
             .unwrap()
-            .run(lazy(|| {
+            .block_on(lazy(|| {
                 let (mut sender, payload) = Payload::new(false);
                 let mut payload = PayloadHelper::new(payload);
 
@@ -665,9 +665,9 @@ mod tests {
 
     #[test]
     fn test_readuntil() {
-        Core::new()
+        Runtime::new()
             .unwrap()
-            .run(lazy(|| {
+            .block_on(lazy(|| {
                 let (mut sender, payload) = Payload::new(false);
                 let mut payload = PayloadHelper::new(payload);
 
@@ -699,9 +699,9 @@ mod tests {
 
     #[test]
     fn test_unread_data() {
-        Core::new()
+        Runtime::new()
             .unwrap()
-            .run(lazy(|| {
+            .block_on(lazy(|| {
                 let (_, mut payload) = Payload::new(false);
 
                 payload.unread_data(Bytes::from("data"));

src/pipeline.rs

@@ -770,7 +770,7 @@ mod tests {
     use actix::*;
     use context::HttpContext;
     use futures::future::{lazy, result};
-    use tokio_core::reactor::Core;
+    use tokio::runtime::current_thread::Runtime;
 
     impl<S, H> PipelineState<S, H> {
         fn is_none(&self) -> Option<bool> {
@@ -796,9 +796,9 @@ mod tests {
 
     #[test]
     fn test_completed() {
-        Core::new()
+        Runtime::new()
             .unwrap()
-            .run(lazy(|| {
+            .block_on(lazy(|| {
                 let mut info = PipelineInfo::new(HttpRequest::default());
                 Completed::<(), Inner<()>>::init(&mut info)
                     .is_none()

src/server/h1.rs

@@ -2,12 +2,11 @@ use std::collections::VecDeque;
 use std::io;
 use std::net::SocketAddr;
 use std::rc::Rc;
-use std::time::Duration;
+use std::time::{Duration, Instant};
 
-use actix::Arbiter;
 use bytes::{BufMut, BytesMut};
 use futures::{Async, Future, Poll};
-use tokio_core::reactor::Timeout;
+use tokio_timer::Delay;
 
 use error::PayloadError;
 use httprequest::HttpRequest;
@@ -53,7 +52,7 @@ pub(crate) struct Http1<T: IoStream, H: 'static> {
     payload: Option<PayloadType>,
     buf: BytesMut,
     tasks: VecDeque<Entry>,
-    keepalive_timer: Option<Timeout>,
+    keepalive_timer: Option<Delay>,
 }
 
 struct Entry {
@@ -295,8 +294,7 @@ where
             if self.keepalive_timer.is_none() && keep_alive > 0 {
                 trace!("Start keep-alive timer");
                 let mut timer =
-                    Timeout::new(Duration::new(keep_alive, 0), Arbiter::handle())
-                        .unwrap();
+                    Delay::new(Instant::now() + Duration::new(keep_alive, 0));
                 // register timer
                 let _ = timer.poll();
                 self.keepalive_timer = Some(timer);

src/server/h2.rs

@@ -4,17 +4,16 @@ use std::collections::VecDeque;
 use std::io::{Read, Write};
 use std::net::SocketAddr;
 use std::rc::Rc;
-use std::time::Duration;
+use std::time::{Duration, Instant};
 use std::{cmp, io, mem};
 
-use actix::Arbiter;
 use bytes::{Buf, Bytes};
 use futures::{Async, Future, Poll, Stream};
 use http2::server::{self, Connection, Handshake, SendResponse};
 use http2::{Reason, RecvStream};
 use modhttp::request::Parts;
-use tokio_core::reactor::Timeout;
 use tokio_io::{AsyncRead, AsyncWrite};
+use tokio_timer::Delay;
 
 use error::PayloadError;
 use httpmessage::HttpMessage;
@@ -46,7 +45,7 @@ where
     addr: Option<SocketAddr>,
     state: State<IoWrapper<T>>,
     tasks: VecDeque<Entry<H>>,
-    keepalive_timer: Option<Timeout>,
+    keepalive_timer: Option<Delay>,
 }
 
 enum State<T: AsyncRead + AsyncWrite> {
@@ -218,9 +217,10 @@ where
                                     let keep_alive = self.settings.keep_alive();
                                     if keep_alive > 0 && self.keepalive_timer.is_none() {
                                         trace!("Start keep-alive timer");
-                                        let mut timeout = Timeout::new(
-                                            Duration::new(keep_alive, 0),
-                                            Arbiter::handle()).unwrap();
+                                        let mut timeout = Delay::new(
+                                            Instant::now()
+                                                + Duration::new(keep_alive, 0),
+                                        );
                                         // register timeout
                                         let _ = timeout.poll();
                                         self.keepalive_timer = Some(timeout);

src/server/mod.rs

@@ -5,8 +5,8 @@ use std::{io, time};
 use actix;
 use bytes::BytesMut;
 use futures::{Async, Poll};
-use tokio_core::net::TcpStream;
 use tokio_io::{AsyncRead, AsyncWrite};
+use tokio_tcp::TcpStream;
 
 mod channel;
 pub(crate) mod encoding;

src/server/srv.rs

@@ -563,11 +563,10 @@ impl<H: IntoHttpHandler> HttpServer<H> {
     /// Start listening for incoming connections from a stream.
     ///
     /// This method uses only one thread for handling incoming connections.
-    pub fn start_incoming<T, A, S>(mut self, stream: S, secure: bool) -> Addr<Syn, Self>
+    pub fn start_incoming<T, S>(mut self, stream: S, secure: bool) -> Addr<Syn, Self>
     where
-        S: Stream<Item = (T, A), Error = io::Error> + 'static,
+        S: Stream<Item = T, Error = io::Error> + 'static,
         T: AsyncRead + AsyncWrite + 'static,
-        A: 'static,
     {
         // set server settings
         let addr: net::SocketAddr = "127.0.0.1:8080".parse().unwrap();
@@ -581,7 +580,7 @@ impl<H: IntoHttpHandler> HttpServer<H> {
         // start server
         let signals = self.subscribe_to_signals();
         let addr: Addr<Syn, _> = HttpServer::create(move |ctx| {
-            ctx.add_message_stream(stream.map_err(|_| ()).map(move |(t, _)| Conn {
+            ctx.add_message_stream(stream.map_err(|_| ()).map(move |t| Conn {
                 io: WrapperStream::new(t),
                 token: 0,
                 peer: None,
@@ -687,7 +686,7 @@ where
     type Result = ();
 
     fn handle(&mut self, msg: Conn<T>, _: &mut Context<Self>) -> Self::Result {
-        Arbiter::handle().spawn(HttpChannel::new(
+        Arbiter::spawn(HttpChannel::new(
             Rc::clone(self.h.as_ref().unwrap()),
             msg.io,
             msg.peer,

src/server/worker.rs

@@ -4,8 +4,8 @@ use net2::TcpStreamExt;
 use slab::Slab;
 use std::rc::Rc;
 use std::{net, time};
-use tokio_core::net::TcpStream;
-use tokio_core::reactor::Handle;
+use tokio_reactor::Handle;
+use tokio_tcp::TcpStream;
 
 #[cfg(any(feature = "tls", feature = "alpn"))]
 use futures::future;
@@ -60,7 +60,6 @@ where
     H: HttpHandler + 'static,
 {
     settings: Rc<WorkerSettings<H>>,
-    hnd: Handle,
     socks: Slab<SocketInfo>,
     tcp_ka: Option<time::Duration>,
 }
@@ -77,7 +76,6 @@ impl<H: HttpHandler + 'static> Worker<H> {
 
         Worker {
             settings: Rc::new(WorkerSettings::new(h, keep_alive)),
-            hnd: Arbiter::handle().clone(),
             socks,
             tcp_ka,
         }
@@ -130,11 +128,11 @@ where
         if self.tcp_ka.is_some() && msg.io.set_keepalive(self.tcp_ka).is_err() {
             error!("Can not set socket keep-alive option");
         }
-        self.socks.get_mut(msg.token).unwrap().htype.handle(
-            Rc::clone(&self.settings),
-            &self.hnd,
-            msg,
-        );
+        self.socks
+            .get_mut(msg.token)
+            .unwrap()
+            .htype
+            .handle(Rc::clone(&self.settings), msg);
     }
 }
 
@@ -174,15 +172,15 @@ pub(crate) enum StreamHandlerType {
 
 impl StreamHandlerType {
     fn handle<H: HttpHandler>(
-        &mut self, h: Rc<WorkerSettings<H>>, hnd: &Handle, msg: Conn<net::TcpStream>,
+        &mut self, h: Rc<WorkerSettings<H>>, msg: Conn<net::TcpStream>,
     ) {
         match *self {
             StreamHandlerType::Normal => {
                 let _ = msg.io.set_nodelay(true);
-                let io = TcpStream::from_stream(msg.io, hnd)
+                let io = TcpStream::from_std(msg.io, &Handle::default())
                     .expect("failed to associate TCP stream");
 
-                hnd.spawn(HttpChannel::new(h, io, msg.peer, msg.http2));
+                Arbiter::spawn(HttpChannel::new(h, io, msg.peer, msg.http2));
             }
             #[cfg(feature = "tls")]
             StreamHandlerType::Tls(ref acceptor) => {
@@ -190,47 +188,50 @@ impl StreamHandlerType {
                     io, peer, http2, ..
                 } = msg;
                 let _ = io.set_nodelay(true);
-                let io = TcpStream::from_stream(io, hnd)
+                let io = TcpStream::from_std(io, &Handle::default())
                     .expect("failed to associate TCP stream");
 
-                hnd.spawn(TlsAcceptorExt::accept_async(acceptor, io).then(move |res| {
-                    match res {
-                        Ok(io) => {
-                            Arbiter::handle().spawn(HttpChannel::new(h, io, peer, http2))
-                        }
-                        Err(err) => {
-                            trace!("Error during handling tls connection: {}", err)
-                        }
-                    };
-                    future::result(Ok(()))
-                }));
+                Arbiter::spawn(TlsAcceptorExt::accept_async(acceptor, io).then(
+                    move |res| {
+                        match res {
+                            Ok(io) => {
+                                Arbiter::spawn(HttpChannel::new(h, io, peer, http2))
+                            }
+                            Err(err) => {
+                                trace!("Error during handling tls connection: {}", err)
+                            }
+                        };
+                        future::result(Ok(()))
+                    },
+                ));
             }
             #[cfg(feature = "alpn")]
             StreamHandlerType::Alpn(ref acceptor) => {
                 let Conn { io, peer, .. } = msg;
                 let _ = io.set_nodelay(true);
-                let io = TcpStream::from_stream(io, hnd)
+                let io = TcpStream::from_std(io, &Handle::default())
                     .expect("failed to associate TCP stream");
 
-                hnd.spawn(SslAcceptorExt::accept_async(acceptor, io).then(move |res| {
-                    match res {
-                        Ok(io) => {
-                            let http2 = if let Some(p) =
-                                io.get_ref().ssl().selected_alpn_protocol()
-                            {
-                                p.len() == 2 && &p == b"h2"
-                            } else {
-                                false
-                            };
-                            Arbiter::handle()
-                                .spawn(HttpChannel::new(h, io, peer, http2));
-                        }
-                        Err(err) => {
-                            trace!("Error during handling tls connection: {}", err)
-                        }
-                    };
-                    future::result(Ok(()))
-                }));
+                Arbiter::spawn(SslAcceptorExt::accept_async(acceptor, io).then(
+                    move |res| {
+                        match res {
+                            Ok(io) => {
+                                let http2 = if let Some(p) =
+                                    io.get_ref().ssl().selected_alpn_protocol()
+                                {
+                                    p.len() == 2 && &p == b"h2"
+                                } else {
+                                    false
+                                };
+                                Arbiter::spawn(HttpChannel::new(h, io, peer, http2));
+                            }
+                            Err(err) => {
+                                trace!("Error during handling tls connection: {}", err)
+                            }
+                        };
+                        future::result(Ok(()))
+                    },
+                ));
             }
         }
     }

src/test.rs

@@ -11,8 +11,9 @@ use futures::Future;
 use http::header::HeaderName;
 use http::{HeaderMap, HttpTryFrom, Method, Uri, Version};
 use net2::TcpBuilder;
-use tokio_core::net::TcpListener;
-use tokio_core::reactor::Core;
+use tokio::runtime::current_thread::Runtime;
+use tokio_reactor::Handle;
+use tokio_tcp::TcpListener;
 
 #[cfg(feature = "alpn")]
 use openssl::ssl::SslAcceptor;
@@ -112,8 +113,7 @@ impl TestServer {
             let sys = System::new("actix-test-server");
             let tcp = net::TcpListener::bind("127.0.0.1:0").unwrap();
             let local_addr = tcp.local_addr().unwrap();
-            let tcp =
-                TcpListener::from_listener(tcp, &local_addr, Arbiter::handle()).unwrap();
+            let tcp = TcpListener::from_std(tcp, &Handle::default()).unwrap();
 
             HttpServer::new(factory)
                 .disable_signals()
@@ -289,8 +289,7 @@ impl<S: 'static> TestServerBuilder<S> {
 
             let tcp = net::TcpListener::bind("127.0.0.1:0").unwrap();
             let local_addr = tcp.local_addr().unwrap();
-            let tcp =
-                TcpListener::from_listener(tcp, &local_addr, Arbiter::handle()).unwrap();
+            let tcp = TcpListener::from_std(tcp, &Handle::default()).unwrap();
 
             let state = self.state;
 
@@ -309,10 +308,9 @@ impl<S: 'static> TestServerBuilder<S> {
                 let ssl = self.ssl.take();
                 if let Some(ssl) = ssl {
                     srv.start_incoming(
-                        tcp.incoming().and_then(move |(sock, addr)| {
+                        tcp.incoming().and_then(move |sock| {
                             ssl.accept_async(sock)
                                 .map_err(|e| io::Error::new(io::ErrorKind::Other, e))
-                                .map(move |s| (s, addr))
                         }),
                         false,
                     );
@@ -616,8 +614,8 @@ impl<S: 'static> TestRequest<S> {
         let req = self.finish();
         let fut = h(req.clone());
 
-        let mut core = Core::new().unwrap();
-        match core.run(fut) {
+        let mut core = Runtime::new().unwrap();
+        match core.block_on(fut) {
             Ok(r) => match r.respond_to(&req) {
                 Ok(reply) => match reply.into().into() {
                     AsyncResultItem::Ok(resp) => Ok(resp),