actix/actix-extras
1
0
Fork 0
mirror of https://github.com/actix/actix-extras.git synced 2024-11-27 17:22:57 +01:00
Code Issues Releases Wiki Activity

update actix

Browse Source
This commit is contained in:
Nikolay Kim 2017-10-17 14:44:00 -07:00
parent 92b81899dd
commit fb92d5552d
2 changed files with 11 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
Cargo.toml
View File

@ -56,7 +56,8 @@ env_logger = "*"
[dependencies.actix] [dependencies.actix]
#path = "../actix" #path = "../actix"
git = "https://github.com/fafhrd91/actix.git" #git = "https://github.com/fafhrd91/actix.git"
version = "0.2"
default-features = false default-features = false
features = [] features = []

23
src/context.rs
View File

@ -6,7 +6,7 @@ use futures::{Async, Stream, Poll};
use bytes::Bytes; use bytes::Bytes;
use actix::{Actor, ActorState, ActorContext, AsyncContext}; use actix::{Actor, ActorState, ActorContext, AsyncContext};
use actix::fut::ActorFuture; use actix::fut::ActorFuture;
use actix::dev::{AsyncContextApi, ActorAddressCell, ActorItemsCell, SpawnHandle}; use actix::dev::{AsyncContextApi, ActorAddressCell, ActorItemsCell, ActorWaitCell, SpawnHandle};
use route::{Route, Frame}; use route::{Route, Frame};
use httpresponse::HttpResponse; use httpresponse::HttpResponse;
@ -20,7 +20,7 @@ pub struct HttpContext<A> where A: Actor<Context=HttpContext<A>> + Route,
items: ActorItemsCell<A>, items: ActorItemsCell<A>,
address: ActorAddressCell<A>, address: ActorAddressCell<A>,
stream: VecDeque<Frame>, stream: VecDeque<Frame>,
wait: Option<Box<ActorFuture<Item=(), Error=(), Actor=A>>>, wait: ActorWaitCell<A>,
app_state: Rc<<A as Route>::State>, app_state: Rc<<A as Route>::State>,
} }
@ -60,7 +60,7 @@ impl<A> AsyncContext<A> for HttpContext<A> where A: Actor<Context=Self> + Route
fn wait<F>(&mut self, fut: F) fn wait<F>(&mut self, fut: F)
where F: ActorFuture<Item=(), Error=(), Actor=A> + 'static where F: ActorFuture<Item=(), Error=(), Actor=A> + 'static
{ {
self.wait = Some(Box::new(fut)); self.wait.add(fut);
} }
fn cancel_future(&mut self, handle: SpawnHandle) -> bool { fn cancel_future(&mut self, handle: SpawnHandle) -> bool {
@ -84,8 +84,8 @@ impl<A> HttpContext<A> where A: Actor<Context=Self> + Route {
state: ActorState::Started, state: ActorState::Started,
items: ActorItemsCell::default(), items: ActorItemsCell::default(),
address: ActorAddressCell::default(), address: ActorAddressCell::default(),
wait: ActorWaitCell::default(),
stream: VecDeque::new(), stream: VecDeque::new(),
wait: None,
app_state: state, app_state: state,
} }
} }
@ -124,7 +124,7 @@ impl<A> Stream for HttpContext<A> where A: Actor<Context=Self> + Route
type Item = Frame; type Item = Frame;
type Error = std::io::Error; type Error = std::io::Error;
fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> { fn poll(&mut self) -> Poll<Option<Frame>, std::io::Error> {
if self.act.is_none() { if self.act.is_none() {
return Ok(Async::NotReady) return Ok(Async::NotReady)
} }
@ -148,16 +148,11 @@ impl<A> Stream for HttpContext<A> where A: Actor<Context=Self> + Route
_ => () _ => ()
} }
// check wait future // check wait futures
if self.wait.is_some() && self.act.is_some() { if let Some(ref mut act) = self.act {
if let Some(ref mut act) = self.act { if let Ok(Async::NotReady) = self.wait.poll(act, ctx) {
if let Some(ref mut fut) = self.wait { return Ok(Async::NotReady)
if let Ok(Async::NotReady) = fut.poll(act, ctx) {
return Ok(Async::NotReady);
}
}
} }
self.wait = None;
} }
let mut prep_stop = false; let mut prep_stop = false;