actix/actix-web
1
0
Fork 0
mirror of https://github.com/fafhrd91/actix-web synced 2025-08-31 08:57:00 +02:00
Code Issues Releases Wiki Activity

fix drain support for actor; make pattern more reusable

Browse Source
This commit is contained in:
Nikolay Kim
2018-01-02 23:43:17 -08:00
parent 9e6d090fd0
commit 70ea43b3c0
6 changed files with 82 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
src/pipeline.rs
View File

@@ -40,6 +40,7 @@ struct PipelineInfo<S> {
mws: Rc<Vec<Box<Middleware<S>>>>,
context: Option<Box<ActorHttpContext>>,
error: Option<Error>,
disconnected: Option<bool>,
}
impl<S> PipelineInfo<S> {
@@ -50,6 +51,7 @@ impl<S> PipelineInfo<S> {
mws: Rc::new(Vec::new()),
error: None,
context: None,
disconnected: None,
}
}
@@ -84,6 +86,7 @@ impl<S, H: PipelineHandler<S>> Pipeline<S, H> {
mws: mws,
error: None,
context: None,
disconnected: None,
};
let state = StartMiddlewares::init(&mut info, handler);
@@ -114,9 +117,7 @@ impl<S, H> Pipeline<S, H> {
impl<S, H: PipelineHandler<S>> HttpHandlerTask for Pipeline<S, H> {
fn disconnected(&mut self) {
if let Some(ref mut context) = self.0.context {
context.disconnected();
}
self.0.disconnected = Some(true);
}
fn poll_io(&mut self, io: &mut Writer) -> Poll<bool, Error> {
@@ -592,10 +593,14 @@ impl<S, H> ProcessResponse<S, H> {
}
},
IOState::Actor(mut ctx) => {
if info.disconnected.take().is_some() {
ctx.disconnected();
}
match ctx.poll() {
Ok(Async::Ready(Some(frame))) => {
match frame {
Frame::Payload(None) => {
println!("ACTOR PAYLOAD EOF");
info.context = Some(ctx);
self.iostate = IOState::Done;
if let Err(err) = io.write_eof() {
@@ -606,6 +611,7 @@ impl<S, H> ProcessResponse<S, H> {
break
},
Frame::Payload(Some(chunk)) => {
println!("ACTOR PAYLOAD");
self.iostate = IOState::Actor(ctx);
match io.write(chunk.as_ref()) {
Err(err) => {
@@ -617,7 +623,9 @@ impl<S, H> ProcessResponse<S, H> {
}
},
Frame::Drain(fut) => {
println!("ACTOR DRAIN");
self.drain = Some(fut);
self.iostate = IOState::Actor(ctx);
break
}
}