actix/actix-extras
1
0
Fork 0
mirror of https://github.com/actix/actix-extras.git synced 2024-11-30 18:34:36 +01:00
Code Issues Releases Wiki Activity

Check Upgrade service readiness before calling it

Browse Source
This commit is contained in:
Nikolay Kim 2019-12-09 17:40:15 +06:00
parent e4382e4fc1
commit 0c1f5f9edc
2 changed files with 37 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
actix-http/CHANGES.md
View File

@ -1,5 +1,10 @@
# Changes # Changes
## [1.0.0-alpha.5] - 2019-12-xx
### Fixed
* Check `Upgrade` service readiness before calling it
### Changed ### Changed

42
actix-http/src/h1/dispatcher.rs
View File

@ -66,6 +66,7 @@ where
U::Error: fmt::Display, U::Error: fmt::Display,
{ {
Normal(InnerDispatcher<T, S, B, X, U>), Normal(InnerDispatcher<T, S, B, X, U>),
UpgradeReadiness(InnerDispatcher<T, S, B, X, U>, Request),
Upgrade(U::Future), Upgrade(U::Future),
None, None,
} }
@ -761,16 +762,8 @@ where
if let DispatcherState::Normal(inner) = if let DispatcherState::Normal(inner) =
std::mem::replace(&mut self.inner, DispatcherState::None) std::mem::replace(&mut self.inner, DispatcherState::None)
{ {
let mut parts = FramedParts::with_read_buf( self.inner =
inner.io, DispatcherState::UpgradeReadiness(inner, req);
inner.codec,
inner.read_buf,
);
parts.write_buf = inner.write_buf;
let framed = Framed::from_parts(parts);
self.inner = DispatcherState::Upgrade(
inner.upgrade.unwrap().call((req, framed)),
);
return self.poll(cx); return self.poll(cx);
} else { } else {
panic!() panic!()
@ -820,6 +813,35 @@ where
} }
} }
} }
DispatcherState::UpgradeReadiness(ref mut inner, _) => {
let upgrade = inner.upgrade.as_mut().unwrap();
match upgrade.poll_ready(cx) {
Poll::Ready(Ok(_)) => {
if let DispatcherState::UpgradeReadiness(inner, req) =
std::mem::replace(&mut self.inner, DispatcherState::None)
{
let mut parts = FramedParts::with_read_buf(
inner.io,
inner.codec,
inner.read_buf,
);
parts.write_buf = inner.write_buf;
let framed = Framed::from_parts(parts);
self.inner = DispatcherState::Upgrade(
inner.upgrade.unwrap().call((req, framed)),
);
self.poll(cx)
} else {
panic!()
}
}
Poll::Pending => Poll::Pending,
Poll::Ready(Err(e)) => {
error!("Upgrade handler readiness check error: {}", e);
Poll::Ready(Err(DispatchError::Upgrade))
}
}
}
DispatcherState::Upgrade(ref mut fut) => { DispatcherState::Upgrade(ref mut fut) => {
unsafe { Pin::new_unchecked(fut) }.poll(cx).map_err(|e| { unsafe { Pin::new_unchecked(fut) }.poll(cx).map_err(|e| {
error!("Upgrade handler error: {}", e); error!("Upgrade handler error: {}", e);