actix compatibility

Cargo.toml

@@ -75,7 +75,8 @@ openssl = { version="0.10", optional = true }
 tokio-openssl = { version="0.2", optional = true }
 
 [dependencies.actix]
-version = "^0.4.5"
+#version = "^0.4.6"
+git = "https://github.com/actix/actix.git"
 
 [dev-dependencies]
 env_logger = "0.5"

src/context.rs

@@ -6,13 +6,12 @@ use futures::unsync::oneshot;
 use smallvec::SmallVec;
 
 use actix::{Actor, ActorState, ActorContext, AsyncContext,
-            Address, SyncAddress, Handler, Subscriber, ResponseType, SpawnHandle};
+            Address, SyncAddress, Handler, ResponseType, MessageResult, SpawnHandle};
 use actix::fut::ActorFuture;
-use actix::dev::{queue, AsyncContextApi,
+use actix::dev::{AsyncContextApi, ContextImpl, Envelope, ToEnvelope, RemoteEnvelope};
-                 ContextImpl, ContextProtocol, Envelope, ToEnvelope, RemoteEnvelope};
 
 use body::{Body, Binary};
-use error::{Error, Result, ErrorInternalServerError};
+use error::{Error, ErrorInternalServerError};
 use httprequest::HttpRequest;
 
 
@@ -86,10 +85,6 @@ impl<A, S> AsyncContext<A> for HttpContext<A, S> where A: Actor<Context=Self>
 
 #[doc(hidden)]
 impl<A, S> AsyncContextApi<A> for HttpContext<A, S> where A: Actor<Context=Self> {
-    #[inline]
-    fn unsync_sender(&mut self) -> queue::unsync::UnboundedSender<ContextProtocol<A>> {
-        self.inner.unsync_sender()
-    }
     #[inline]
     fn unsync_address(&mut self) -> Address<A> {
         self.inner.unsync_address()
@@ -174,26 +169,6 @@ impl<A, S> HttpContext<A, S> where A: Actor<Context=Self> {
     }
 }
 
-impl<A, S> HttpContext<A, S> where A: Actor<Context=Self> {
-
-    #[inline]
-    #[doc(hidden)]
-    pub fn subscriber<M>(&mut self) -> Box<Subscriber<M>>
-        where A: Handler<M>, M: ResponseType + 'static
-    {
-        self.inner.subscriber()
-    }
-
-    #[inline]
-    #[doc(hidden)]
-    pub fn sync_subscriber<M>(&mut self) -> Box<Subscriber<M> + Send>
-        where A: Handler<M>,
-              M: ResponseType + Send + 'static, M::Item: Send, M::Error: Send,
-    {
-        self.inner.sync_subscriber()
-    }
-}
-
 impl<A, S> ActorHttpContext for HttpContext<A, S> where A: Actor<Context=Self>, S: 'static {
 
     #[inline]
@@ -229,12 +204,11 @@ impl<A, S> ToEnvelope<A> for HttpContext<A, S>
     where A: Actor<Context=HttpContext<A, S>>,
 {
     #[inline]
-    fn pack<M>(msg: M, tx: Option<Sender<Result<M::Item, M::Error>>>,
+    fn pack_msg<M>(msg: M, tx: Option<Sender<MessageResult<M>>>) -> Envelope<A>
-               channel_on_drop: bool) -> Envelope<A>
         where A: Handler<M>,
               M: ResponseType + Send + 'static, M::Item: Send, M::Error: Send
     {
-        RemoteEnvelope::new(msg, tx, channel_on_drop).into()
+        RemoteEnvelope::envelope(msg, tx).into()
     }
 }
 

src/server/srv.rs

@@ -313,7 +313,8 @@ impl<H: HttpHandler, U, V> HttpServer<TcpStream, net::SocketAddr, H, U>
             // start http server actor
             let signals = self.subscribe_to_signals();
             let addr: SyncAddress<_> = Actor::start(self);
-            signals.map(|signals| signals.send(signal::Subscribe(addr.subscriber())));
+            signals.map(|signals| signals.send(
+                signal::Subscribe(addr.clone().into_subscriber())));
             addr
         }
     }
@@ -478,7 +479,8 @@ impl<T, A, H, U, V> HttpServer<WrapperStream<T>, A, H, U>
                 move |(t, _)| Conn{io: WrapperStream::new(t), peer: None, http2: false}));
             self
         });
-        signals.map(|signals| signals.send(signal::Subscribe(addr.subscriber())));
+        signals.map(|signals| signals.send(
+            signal::Subscribe(addr.clone().into_subscriber())));
         addr
     }
 }

src/ws/context.rs

@@ -5,13 +5,12 @@ use futures::unsync::oneshot;
 use smallvec::SmallVec;
 
 use actix::{Actor, ActorState, ActorContext, AsyncContext,
-            Address, SyncAddress, Handler, Subscriber, ResponseType, SpawnHandle};
+            Address, SyncAddress, Handler, ResponseType, SpawnHandle, MessageResult};
 use actix::fut::ActorFuture;
-use actix::dev::{queue, AsyncContextApi,
+use actix::dev::{AsyncContextApi, ContextImpl, Envelope, ToEnvelope, RemoteEnvelope};
-                 ContextImpl, ContextProtocol, Envelope, ToEnvelope, RemoteEnvelope};
 
 use body::{Body, Binary};
-use error::{Error, Result, ErrorInternalServerError};
+use error::{Error, ErrorInternalServerError};
 use httprequest::HttpRequest;
 use context::{Frame as ContextFrame, ActorHttpContext, Drain};
 
@@ -69,10 +68,6 @@ impl<A, S> AsyncContext<A> for WebsocketContext<A, S> where A: Actor<Context=Sel
 
 #[doc(hidden)]
 impl<A, S> AsyncContextApi<A> for WebsocketContext<A, S> where A: Actor<Context=Self> {
-    #[inline]
-    fn unsync_sender(&mut self) -> queue::unsync::UnboundedSender<ContextProtocol<A>> {
-        self.inner.unsync_sender()
-    }
 
     #[inline]
     fn unsync_address(&mut self) -> Address<A> {
@@ -198,26 +193,6 @@ impl<A, S> WebsocketContext<A, S> where A: Actor<Context=Self> {
     }
 }
 
-impl<A, S> WebsocketContext<A, S> where A: Actor<Context=Self> {
-
-    #[inline]
-    #[doc(hidden)]
-    pub fn subscriber<M>(&mut self) -> Box<Subscriber<M>>
-        where A: Handler<M>, M: ResponseType + 'static
-    {
-        self.inner.subscriber()
-    }
-
-    #[inline]
-    #[doc(hidden)]
-    pub fn sync_subscriber<M>(&mut self) -> Box<Subscriber<M> + Send>
-        where A: Handler<M>,
-              M: ResponseType + Send + 'static, M::Item: Send, M::Error: Send,
-    {
-        self.inner.sync_subscriber()
-    }
-}
-
 impl<A, S> ActorHttpContext for WebsocketContext<A, S> where A: Actor<Context=Self>, S: 'static {
 
     #[inline]
@@ -253,11 +228,10 @@ impl<A, S> ToEnvelope<A> for WebsocketContext<A, S>
     where A: Actor<Context=WebsocketContext<A, S>>,
 {
     #[inline]
-    fn pack<M>(msg: M, tx: Option<Sender<Result<M::Item, M::Error>>>,
+    fn pack_msg<M>(msg: M, tx: Option<Sender<MessageResult<M>>>) -> Envelope<A>
-               channel_on_drop: bool) -> Envelope<A>
         where A: Handler<M>,
               M: ResponseType + Send + 'static, M::Item: Send, M::Error: Send {
-        RemoteEnvelope::new(msg, tx, channel_on_drop).into()
+        RemoteEnvelope::envelope(msg, tx).into()
     }
 }