Merge branch 'master' into master

Cargo.toml

@@ -77,7 +77,8 @@ tokio-tls = { version="0.1", optional = true }
 tokio-openssl = { version="0.1", optional = true }
 
 [dependencies.actix]
-version = "^0.4.1"
+#version = "^0.4.2"
+git = "https://github.com/actix/actix.git"
 
 [dependencies.openssl]
 version = "0.9"

examples/websocket-chat/Cargo.toml

@@ -25,5 +25,6 @@ serde = "1.0"
 serde_json = "1.0"
 serde_derive = "1.0"
 
-actix = "0.4"
+#actix = "0.4"
+actix = { git = "https://github.com/actix/actix" }
 actix-web = { git = "https://github.com/actix/actix-web" }

examples/websocket-chat/src/client.rs

@@ -67,11 +67,13 @@ impl Actor for ChatClient {
         self.hb(ctx)
     }
 
-    fn stopping(&mut self, _: &mut FramedContext<Self>) {
+    fn stopping(&mut self, _: &mut FramedContext<Self>) -> bool {
         println!("Disconnected");
 
         // Stop application on disconnect
         Arbiter::system().send(actix::msgs::SystemExit(0));
+
+        true
     }
 }
 

examples/websocket-chat/src/main.rs

@@ -75,9 +75,10 @@ impl Actor for WsChatSession {
             }).wait(ctx);
     }
 
-    fn stopping(&mut self, ctx: &mut Self::Context) {
+    fn stopping(&mut self, ctx: &mut Self::Context) -> bool {
         // notify chat server
         ctx.state().addr.send(server::Disconnect{id: self.id});
+        true
     }
 }
 

examples/websocket-chat/src/session.rs

@@ -52,10 +52,10 @@ impl Actor for ChatSession {
             }).wait(ctx);
     }
 
-    fn stopping(&mut self, ctx: &mut Self::Context) {
+    fn stopping(&mut self, ctx: &mut Self::Context) -> bool {
         // notify chat server
         self.addr.send(server::Disconnect{id: self.id});
-        ctx.stop()
+        true
     }
 }
 

examples/websocket/Cargo.toml

@@ -10,5 +10,6 @@ path = "src/main.rs"
 [dependencies]
 env_logger = "*"
 futures = "0.1"
-actix = "0.4"
+#actix = "0.4"
+actix = { git = "https://github.com/actix/actix.git" }
 actix-web = { git = "https://github.com/actix/actix-web.git" }

src/context.rs

@@ -8,11 +8,11 @@ use futures::unsync::oneshot;
 use actix::{Actor, ActorState, ActorContext, AsyncContext,
             Handler, Subscriber, ResponseType, SpawnHandle};
 use actix::fut::ActorFuture;
-use actix::dev::{AsyncContextApi, ActorAddressCell, ActorItemsCell, ActorWaitCell,
+use actix::dev::{AsyncContextApi, ActorAddressCell,
-                 Envelope, ToEnvelope, RemoteEnvelope};
+                 ContextImpl, Envelope, ToEnvelope, RemoteEnvelope};
 
 use body::{Body, Binary};
-use error::{Error, Result};
+use error::{Error, Result, ErrorInternalServerError};
 use httprequest::HttpRequest;
 
 
@@ -30,13 +30,8 @@ pub enum Frame {
 /// Http actor execution context
 pub struct HttpContext<A, S=()> where A: Actor<Context=HttpContext<A, S>>,
 {
-    act: Option<A>,
+    inner: ContextImpl<A>,
-    state: ActorState,
-    modified: bool,
-    items: ActorItemsCell<A>,
-    address: ActorAddressCell<A>,
     stream: VecDeque<Frame>,
-    wait: ActorWaitCell<A>,
     request: HttpRequest<S>,
     disconnected: bool,
 }
@@ -46,23 +41,17 @@ impl<A, S> ActorContext for HttpContext<A, S> where A: Actor<Context=Self>
     /// Stop actor execution
     fn stop(&mut self) {
         self.stream.push_back(Frame::Payload(None));
-        self.items.stop();
+        self.inner.stop();
-        self.address.close();
-        if self.state == ActorState::Running {
-            self.state = ActorState::Stopping;
-        }
     }
 
     /// Terminate actor execution
     fn terminate(&mut self) {
-        self.address.close();
+        self.inner.terminate()
-        self.items.close();
-        self.state = ActorState::Stopped;
     }
 
     /// Actor execution state
     fn state(&self) -> ActorState {
-        self.state
+        self.inner.state()
     }
 }
 
@@ -71,31 +60,24 @@ impl<A, S> AsyncContext<A> for HttpContext<A, S> where A: Actor<Context=Self>
     fn spawn<F>(&mut self, fut: F) -> SpawnHandle
         where F: ActorFuture<Item=(), Error=(), Actor=A> + 'static
     {
-        self.modified = true;
+        self.inner.spawn(fut)
-        self.items.spawn(fut)
     }
 
     fn wait<F>(&mut self, fut: F)
         where F: ActorFuture<Item=(), Error=(), Actor=A> + 'static
     {
-        self.modified = true;
+        self.inner.wait(fut)
-        self.wait.add(fut);
     }
 
     fn cancel_future(&mut self, handle: SpawnHandle) -> bool {
-        self.modified = true;
+        self.inner.cancel_future(handle)
-        self.items.cancel_future(handle)
-    }
-
-    fn cancel_future_on_stop(&mut self, handle: SpawnHandle) {
-        self.items.cancel_future_on_stop(handle)
     }
 }
 
 #[doc(hidden)]
 impl<A, S> AsyncContextApi<A> for HttpContext<A, S> where A: Actor<Context=Self> {
     fn address_cell(&mut self) -> &mut ActorAddressCell<A> {
-        &mut self.address
+        self.inner.address_cell()
     }
 }
 
@@ -107,12 +89,7 @@ impl<A, S: 'static> HttpContext<A, S> where A: Actor<Context=Self> {
 
     pub fn from_request(req: HttpRequest<S>) -> HttpContext<A, S> {
         HttpContext {
-            act: None,
+            inner: ContextImpl::new(None),
-            state: ActorState::Started,
-            modified: false,
-            items: ActorItemsCell::default(),
-            address: ActorAddressCell::default(),
-            wait: ActorWaitCell::default(),
             stream: VecDeque::new(),
             request: req,
             disconnected: false,
@@ -120,7 +97,7 @@ impl<A, S: 'static> HttpContext<A, S> where A: Actor<Context=Self> {
     }
 
     pub fn actor(mut self, actor: A) -> HttpContext<A, S> {
-        self.act = Some(actor);
+        self.inner.set_actor(actor);
         self
     }
 }
@@ -154,7 +131,7 @@ impl<A, S> HttpContext<A, S> where A: Actor<Context=Self> {
     /// Returns drain future
     pub fn drain(&mut self) -> Drain<A> {
         let (tx, rx) = oneshot::channel();
-        self.modified = true;
+        self.inner.modify();
         self.stream.push_back(Frame::Drain(tx));
         Drain::new(rx)
     }
@@ -169,120 +146,43 @@ impl<A, S> HttpContext<A, S> where A: Actor<Context=Self> {
 
     #[doc(hidden)]
     pub fn subscriber<M>(&mut self) -> Box<Subscriber<M>>
-        where A: Handler<M>,
+        where A: Handler<M>, M: ResponseType + 'static
-              M: ResponseType + 'static,
     {
-        Box::new(self.address.unsync_address())
+        self.inner.subscriber()
     }
 
     #[doc(hidden)]
     pub fn sync_subscriber<M>(&mut self) -> Box<Subscriber<M> + Send>
         where A: Handler<M>,
-              M: ResponseType + Send + 'static,
+              M: ResponseType + Send + 'static, M::Item: Send, M::Error: Send,
-              M::Item: Send,
-              M::Error: Send,
     {
-        Box::new(self.address.sync_address())
+        self.inner.sync_subscriber()
     }
 }
 
 impl<A, S> ActorHttpContext for HttpContext<A, S> where A: Actor<Context=Self>, S: 'static {
 
     fn disconnected(&mut self) {
-        self.items.stop();
         self.disconnected = true;
-        if self.state == ActorState::Running {
+        self.stop();
-            self.state = ActorState::Stopping;
-        }
     }
 
     fn poll(&mut self) -> Poll<Option<Frame>, Error> {
-        if self.act.is_none() {
-            return Ok(Async::Ready(None))
-        }
-        let act: &mut A = unsafe {
-            std::mem::transmute(self.act.as_mut().unwrap() as &mut A)
-        };
         let ctx: &mut HttpContext<A, S> = unsafe {
             std::mem::transmute(self as &mut HttpContext<A, S>)
         };
 
-        // update state
+        match self.inner.poll(ctx) {
-        match self.state {
+            Ok(Async::NotReady) => {
-            ActorState::Started => {
-                Actor::started(act, ctx);
-                self.state = ActorState::Running;
-            },
-            ActorState::Stopping => {
-                Actor::stopping(act, ctx);
-            }
-            _ => ()
-        }
-
-        let mut prep_stop = false;
-        loop {
-            self.modified = false;
-
-            // check wait futures
-            if self.wait.poll(act, ctx) {
                 // get frame
                 if let Some(frame) = self.stream.pop_front() {
-                    return Ok(Async::Ready(Some(frame)))
+                    Ok(Async::Ready(Some(frame)))
+                } else {
+                    Ok(Async::NotReady)
                 }
-                return Ok(Async::NotReady)
             }
-
+            Ok(Async::Ready(())) => Ok(Async::Ready(None)),
-            // incoming messages
+            Err(_) => Err(ErrorInternalServerError("error").into()),
-            self.address.poll(act, ctx);
-
-            // spawned futures and streams
-            self.items.poll(act, ctx);
-
-            // are we done
-            if self.modified {
-                continue
-            }
-
-            // get frame
-            if let Some(frame) = self.stream.pop_front() {
-                return Ok(Async::Ready(Some(frame)))
-            }
-
-            // check state
-            match self.state {
-                ActorState::Stopped => {
-                    self.state = ActorState::Stopped;
-                    Actor::stopped(act, ctx);
-                    return Ok(Async::Ready(None))
-                },
-                ActorState::Stopping => {
-                    if prep_stop {
-                        if self.address.connected() || !self.items.is_empty() {
-                            self.state = ActorState::Running;
-                            continue
-                        } else {
-                            self.state = ActorState::Stopped;
-                            Actor::stopped(act, ctx);
-                            return Ok(Async::Ready(None))
-                        }
-                    } else {
-                        Actor::stopping(act, ctx);
-                        prep_stop = true;
-                        continue
-                    }
-                },
-                ActorState::Running => {
-                    if !self.address.connected() && self.items.is_empty() {
-                        self.state = ActorState::Stopping;
-                        Actor::stopping(act, ctx);
-                        prep_stop = true;
-                        continue
-                    }
-                },
-                _ => (),
-            }
-
-            return Ok(Async::NotReady)
         }
     }
 }

src/server.rs

@@ -556,12 +556,6 @@ impl<T, A, H, U> Handler<signal::Signal> for HttpServer<T, A, H, U>
     }
 }
 
-impl<T, A, H, U> StreamHandler<io::Result<Conn<T>>> for HttpServer<T, A, H, U>
-    where T: IoStream,
-          H: HttpHandler + 'static,
-          U: 'static,
-          A: 'static {}
-
 impl<T, A, H, U> Handler<io::Result<Conn<T>>> for HttpServer<T, A, H, U>
     where T: IoStream,
           H: HttpHandler + 'static,
@@ -682,7 +676,7 @@ impl<T, A, H, U> Handler<StopServer> for HttpServer<T, A, H, U>
             if self.exit {
                 Arbiter::system().send(actix::msgs::SystemExit(0))
             }
-            Self::empty()
+            Self::reply(Ok(()))
         }
     }
 }

src/worker.rs

@@ -177,7 +177,7 @@ impl<H> Handler<StopWorker> for Worker<H>
         let num = self.settings.channels.get();
         if num == 0 {
             info!("Shutting down http worker, 0 connections");
-            Self::reply(true)
+            Self::reply(Ok(true))
         } else if let Some(dur) = msg.graceful {
             info!("Graceful http worker shutdown, {} connections", num);
             let (tx, rx) = oneshot::channel();
@@ -186,7 +186,7 @@ impl<H> Handler<StopWorker> for Worker<H>
         } else {
             info!("Force shutdown http worker, {} connections", num);
             self.settings.head().traverse::<TcpStream, H>();
-            Self::reply(false)
+            Self::reply(Ok(false))
         }
     }
 }