actix/actix-web
1
0
Fork 0
mirror of https://github.com/fafhrd91/actix-web synced 2024-11-27 17:52:56 +01:00
Code Issues Releases Wiki Activity

provide wait method

Browse Source
This commit is contained in:
Nikolay Kim 2017-10-11 19:20:05 -07:00
parent 8b1fdeb8c9
commit c0e73c7275
1 changed files with 23 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
src/context.rs
View File

@ -4,7 +4,7 @@ use std::collections::VecDeque;
use futures::{Async, Stream, Poll}; use futures::{Async, Stream, Poll};
use bytes::Bytes; use bytes::Bytes;
use actix::{Actor, ActorState, ActorContext, AsyncActorContext}; 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, SpawnHandle};
@ -20,6 +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>>>,
app_state: Rc<<A as Route>::State>, app_state: Rc<<A as Route>::State>,
} }
@ -47,7 +48,7 @@ impl<A> ActorContext<A> for HttpContext<A> where A: Actor<Context=Self> + Route
} }
} }
impl<A> AsyncActorContext<A> for HttpContext<A> where A: Actor<Context=Self> + Route impl<A> AsyncContext<A> for HttpContext<A> where A: Actor<Context=Self> + Route
{ {
fn spawn<F>(&mut self, fut: F) -> SpawnHandle fn spawn<F>(&mut self, fut: F) -> SpawnHandle
where F: ActorFuture<Item=(), Error=(), Actor=A> + 'static where F: ActorFuture<Item=(), Error=(), Actor=A> + 'static
@ -55,6 +56,12 @@ impl<A> AsyncActorContext<A> for HttpContext<A> where A: Actor<Context=Self> + R
self.items.spawn(fut) self.items.spawn(fut)
} }
fn wait<F>(&mut self, fut: F)
where F: ActorFuture<Item=(), Error=(), Actor=A> + 'static
{
self.wait = Some(Box::new(fut));
}
fn cancel_future(&mut self, handle: SpawnHandle) -> bool { fn cancel_future(&mut self, handle: SpawnHandle) -> bool {
self.items.cancel_future(handle) self.items.cancel_future(handle)
} }
@ -77,6 +84,7 @@ impl<A> HttpContext<A> where A: Actor<Context=Self> + Route {
items: ActorItemsCell::default(), items: ActorItemsCell::default(),
address: ActorAddressCell::default(), address: ActorAddressCell::default(),
stream: VecDeque::new(), stream: VecDeque::new(),
wait: None,
app_state: state, app_state: state,
} }
} }
@ -139,6 +147,19 @@ impl<A> Stream for HttpContext<A> where A: Actor<Context=Self> + Route
_ => () _ => ()
} }
// check wait future
if self.wait.is_some() && self.act.is_some() {
if let Some(ref mut act) = self.act {
if let Some(ref mut fut) = self.wait {
match fut.poll(act, ctx) {
Ok(Async::NotReady) => return Ok(Async::NotReady),
_ => (),
}
}
}
self.wait = None;
}
let mut prep_stop = false; let mut prep_stop = false;
loop { loop {
let mut not_ready = true; let mut not_ready = true;