sync with latest actix

src/client/connector.rs

@@ -5,7 +5,7 @@ use std::collections::VecDeque;
 use std::time::Duration;
 
 use actix::{fut, Actor, ActorFuture, Arbiter, Context,
-            Handler, Response, ResponseType, Supervised};
+            Handler, Message, ActorResponse, Supervised};
 use actix::registry::ArbiterService;
 use actix::fut::WrapFuture;
 use actix::actors::{Connector, ConnectorError, Connect as ResolveConnect};
@@ -37,9 +37,8 @@ impl Connect {
     }
 }
 
-impl ResponseType for Connect {
-    type Item = Connection;
-    type Error = ClientConnectorError;
+impl Message for Connect {
+    type Result = Result<Connection, ClientConnectorError>;
 }
 
 /// A set of errors that can occur during connecting to a http host
@@ -163,34 +162,34 @@ impl ClientConnector {
 }
 
 impl Handler<Connect> for ClientConnector {
-    type Result = Response<ClientConnector, Connect>;
+    type Result = ActorResponse<ClientConnector, Connection, ClientConnectorError>;
 
     fn handle(&mut self, msg: Connect, _: &mut Self::Context) -> Self::Result {
         let uri = &msg.0;
 
         // host name is required
         if uri.host().is_none() {
-            return Response::reply(Err(ClientConnectorError::InvalidUrl))
+            return ActorResponse::reply(Err(ClientConnectorError::InvalidUrl))
         }
 
         // supported protocols
         let proto = match uri.scheme_part() {
             Some(scheme) => match Protocol::from(scheme.as_str()) {
                 Some(proto) => proto,
-                None => return Response::reply(Err(ClientConnectorError::InvalidUrl)),
+                None => return ActorResponse::reply(Err(ClientConnectorError::InvalidUrl)),
             },
-            None => return Response::reply(Err(ClientConnectorError::InvalidUrl)),
+            None => return ActorResponse::reply(Err(ClientConnectorError::InvalidUrl)),
         };
 
         // check ssl availability
         if proto.is_secure() && !HAS_OPENSSL { //&& !HAS_TLS {
-            return Response::reply(Err(ClientConnectorError::SslIsNotSupported))
+            return ActorResponse::reply(Err(ClientConnectorError::SslIsNotSupported))
         }
 
         let host = uri.host().unwrap().to_owned();
         let port = uri.port().unwrap_or_else(|| proto.port());
 
-        Response::async_reply(
+        ActorResponse::async(
             Connector::from_registry()
                 .call(self, ResolveConnect::host_and_port(&host, port))
                 .map_err(|_, _, _| ClientConnectorError::Disconnected)

src/context.rs

@@ -1,4 +1,4 @@
-use std;
+use std::mem;
 use std::marker::PhantomData;
 use futures::{Async, Future, Poll};
 use futures::sync::oneshot::Sender;
@@ -6,7 +6,7 @@ use futures::unsync::oneshot;
 use smallvec::SmallVec;
 
 use actix::{Actor, ActorState, ActorContext, AsyncContext,
-            Addr, Handler, ResponseType, MessageResult, SpawnHandle, Syn, Unsync};
+            Addr, Handler, Message, SpawnHandle, Syn, Unsync};
 use actix::fut::ActorFuture;
 use actix::dev::{ContextImpl, ToEnvelope, RemoteEnvelope};
 
@@ -184,7 +184,7 @@ impl<A, S> ActorHttpContext for HttpContext<A, S> where A: Actor<Context=Self>,
 
     fn poll(&mut self) -> Poll<Option<SmallVec<[Frame; 4]>>, Error> {
         let ctx: &mut HttpContext<A, S> = unsafe {
-            std::mem::transmute(self as &mut HttpContext<A, S>)
+            mem::transmute(self as &mut HttpContext<A, S>)
         };
 
         if self.inner.alive() {
@@ -207,9 +207,9 @@ impl<A, S> ActorHttpContext for HttpContext<A, S> where A: Actor<Context=Self>,
 
 impl<A, M, S> ToEnvelope<Syn<A>, M> for HttpContext<A, S>
     where A: Actor<Context=HttpContext<A, S>> + Handler<M>,
-          M: ResponseType + Send + 'static, M::Item: Send, M::Error: Send,
+          M: Message + Send + 'static, M::Result: Send,
 {
-    fn pack(msg: M, tx: Option<Sender<MessageResult<M>>>) -> Syn<A> {
+    fn pack(msg: M, tx: Option<Sender<M::Result>>) -> Syn<A> {
         Syn::new(Box::new(RemoteEnvelope::envelope(msg, tx)))
     }
 }

src/server/mod.rs

@@ -2,6 +2,7 @@
 use std::{time, io};
 use std::net::Shutdown;
 
+use actix;
 use futures::Poll;
 use tokio_io::{AsyncRead, AsyncWrite};
 use tokio_core::net::TcpStream;
@@ -43,11 +44,14 @@ pub struct ResumeServer;
 /// Stop incoming connection processing, stop all workers and exit.
 ///
 /// If server starts with `spawn()` method, then spawned thread get terminated.
-#[derive(Message)]
 pub struct StopServer {
     pub graceful: bool
 }
 
+impl actix::Message for StopServer {
+    type Result = Result<(), ()>;
+}
+
 /// Low level http request handler
 #[allow(unused_variables)]
 pub trait HttpHandler: 'static {

src/server/srv.rs

@@ -352,7 +352,7 @@ impl<H: IntoHttpHandler> HttpServer<H>
             let signals = self.subscribe_to_signals();
             let addr: SyncAddress<_> = Actor::start(self);
             signals.map(|signals| signals.send(
-                signal::Subscribe(addr.clone().into())));
+                signal::Subscribe(addr.clone().subscriber())));
             Ok(addr)
         }
     }
@@ -396,7 +396,7 @@ impl<H: IntoHttpHandler> HttpServer<H>
             let signals = self.subscribe_to_signals();
             let addr: SyncAddress<_> = Actor::start(self);
             signals.map(|signals| signals.send(
-                signal::Subscribe(addr.clone().into())));
+                signal::Subscribe(addr.clone().subscriber())));
             Ok(addr)
         }
     }
@@ -477,24 +477,6 @@ impl<H: IntoHttpHandler> Handler<signal::Signal> for HttpServer<H>
     }
 }
 
-impl<T, H> Handler<io::Result<Conn<T>>> for HttpServer<H>
-    where T: IoStream,
-          H: IntoHttpHandler,
-{
-    type Result = ();
-
-    fn handle(&mut self, msg: io::Result<Conn<T>>, _: &mut Context<Self>) -> Self::Result {
-        match msg {
-            Ok(msg) =>
-                Arbiter::handle().spawn(
-                    HttpChannel::new(
-                        Rc::clone(self.h.as_ref().unwrap()), msg.io, msg.peer, msg.http2)),
-            Err(err) =>
-                debug!("Error handling request: {}", err),
-        }
-    }
-}
-
 impl<T, H> Handler<Conn<T>> for HttpServer<H>
     where T: IoStream,
           H: IntoHttpHandler,
@@ -535,7 +517,7 @@ impl<H: IntoHttpHandler> Handler<ResumeServer> for HttpServer<H>
 
 impl<H: IntoHttpHandler> Handler<StopServer> for HttpServer<H>
 {
-    type Result = actix::Response<Self, StopServer>;
+    type Result = actix::Response<(), ()>;
 
     fn handle(&mut self, msg: StopServer, ctx: &mut Context<Self>) -> Self::Result {
         // stop accept threads
@@ -570,8 +552,8 @@ impl<H: IntoHttpHandler> Handler<StopServer> for HttpServer<H>
         }
 
         if !self.workers.is_empty() {
-            Response::async_reply(
-                rx.into_future().map(|_| ()).map_err(|_| ()).actfuture())
+            Response::async(
+                rx.into_future().map(|_| ()).map_err(|_| ()))
         } else {
             // we need to stop system if server was spawned
             if self.exit {

src/server/worker.rs

@@ -37,12 +37,14 @@ pub(crate) struct Conn<T> {
 
 /// Stop worker message. Returns `true` on successful shutdown
 /// and `false` if some connections still alive.
-#[derive(Message)]
-#[rtype(bool)]
 pub(crate) struct StopWorker {
     pub graceful: Option<time::Duration>,
 }
 
+impl Message for StopWorker {
+    type Result = Result<bool, ()>;
+}
+
 /// Http worker
 ///
 /// Worker accepts Socket objects via unbounded channel and start requests processing.
@@ -117,7 +119,7 @@ impl<H> Handler<Conn<net::TcpStream>> for Worker<H>
 impl<H> Handler<StopWorker> for Worker<H>
     where H: HttpHandler + 'static,
 {
-    type Result = Response<Self, StopWorker>;
+    type Result = Response<bool, ()>;
 
     fn handle(&mut self, msg: StopWorker, ctx: &mut Context<Self>) -> Self::Result {
         let num = self.settings.num_channels();
@@ -128,7 +130,7 @@ impl<H> Handler<StopWorker> for Worker<H>
             info!("Graceful http worker shutdown, {} connections", num);
             let (tx, rx) = oneshot::channel();
             self.shutdown_timeout(ctx, tx, dur);
-            Response::async_reply(rx.map_err(|_| ()).actfuture())
+            Response::async(rx.map_err(|_| ()))
         } else {
             info!("Force shutdown http worker, {} connections", num);
             self.settings.head().traverse::<TcpStream, H>();

src/ws/context.rs

@@ -5,7 +5,7 @@ use futures::unsync::oneshot;
 use smallvec::SmallVec;
 
 use actix::{Actor, ActorState, ActorContext, AsyncContext,
-            Addr, Handler, ResponseType, SpawnHandle, MessageResult, Syn, Unsync};
+            Addr, Handler, Message, Syn, Unsync, SpawnHandle};
 use actix::fut::ActorFuture;
 use actix::dev::{ContextImpl, ToEnvelope, RemoteEnvelope};
 
@@ -219,9 +219,9 @@ impl<A, S> ActorHttpContext for WebsocketContext<A, S> where A: Actor<Context=Se
 
 impl<A, M, S> ToEnvelope<Syn<A>, M> for WebsocketContext<A, S>
     where A: Actor<Context=WebsocketContext<A, S>> + Handler<M>,
-          M: ResponseType + Send + 'static, M::Item: Send, M::Error: Send,
+          M: Message + Send + 'static, M::Result: Send
 {
-    fn pack(msg: M, tx: Option<Sender<MessageResult<M>>>) -> Syn<A> {
+    fn pack(msg: M, tx: Option<Sender<M::Result>>) -> Syn<A> {
         Syn::new(Box::new(RemoteEnvelope::envelope(msg, tx)))
     }
 }

src/ws/mod.rs

@@ -47,7 +47,7 @@ use bytes::BytesMut;
 use http::{Method, StatusCode, header};
 use futures::{Async, Poll, Stream};
 
-use actix::{Actor, AsyncContext, ResponseType, Handler};
+use actix::{Actor, AsyncContext, Handler};
 
 use body::Binary;
 use payload::ReadAny;
@@ -74,7 +74,7 @@ const SEC_WEBSOCKET_VERSION: &str = "SEC-WEBSOCKET-VERSION";
 
 
 /// `WebSocket` Message
-#[derive(Debug, PartialEq)]
+#[derive(Debug, PartialEq, Message)]
 pub enum Message {
     Text(String),
     Binary(Binary),
@@ -85,11 +85,6 @@ pub enum Message {
     Error
 }
 
-impl ResponseType for Message {
-    type Item = ();
-    type Error = ();
-}
-
 /// Do websocket handshake and start actor
 pub fn start<A, S>(mut req: HttpRequest<S>, actor: A) -> Result<HttpResponse, Error>
     where A: Actor<Context=WebsocketContext<A, S>> + Handler<Message>,