fix framed dispatcher
@@ -149,7 +149,7 @@ where
self.state = FramedState::Error(ServiceError::Service(err));
return true;
}
- Poll::Ready(None) | Poll::Pending => (),
+ Poll::Ready(None) | Poll::Pending => {}
}
if self.sink.is_some() {
@@ -195,8 +195,9 @@ where
&mut self,
cx: &mut Context<'_>,
) -> Poll<Result<(), ServiceError<S::Error, U>>> {
+ loop {
match self.state {
- FramedState::Processing => loop {
+ FramedState::Processing => {
let read = self.poll_read(cx);
let write = self.poll_write(cx);
if read || write {
@@ -204,7 +205,7 @@ where
} else {
return Poll::Pending;
}
- },
+ }
FramedState::Error(_) => {
// flush write buffer
if !self.framed.is_write_buf_empty() {
@@ -213,21 +214,21 @@ where
}
}
self.state = FramedState::Shutdown(Some(self.state.take_error()));
- self.poll(cx)
}
FramedState::FlushAndStop => {
// drain service responses
match Pin::new(&mut self.rx).poll_next(cx) {
Poll::Ready(Some(Ok(msg))) => {
if let Err(err) = self.framed.write(msg) {
- self.state =
- FramedState::Shutdown(Some(ServiceError::Encoder(err)));
- return self.poll(cx);
+ self.state = FramedState::Shutdown(Some(
+ ServiceError::Encoder(err),
+ ));
+ continue;
}
}
Poll::Ready(Some(Err(err))) => {
self.state = FramedState::Shutdown(Some(err.into()));
- return self.poll(cx);
+ continue;
}
Poll::Ready(None) | Poll::Pending => (),
}
@@ -245,18 +246,16 @@ where
}
};
self.state = FramedState::Shutdown(None);
- self.poll(cx)
}
FramedState::FramedError(_) => {
- self.state = FramedState::Shutdown(Some(self.state.take_framed_error()));
- self.poll(cx)
+ self.state =
+ FramedState::Shutdown(Some(self.state.take_framed_error()));
}
FramedState::Stopping => {
self.state = FramedState::Shutdown(None);
- self.poll(cx)
}
FramedState::Shutdown(ref mut err) => {
- if self.service.poll_shutdown(cx, err.is_some()).is_ready() {
+ return if self.service.poll_shutdown(cx, err.is_some()).is_ready() {
if let Some(err) = err.take() {
Poll::Ready(Err(err))
} else {
@@ -269,3 +268,4 @@ where
}
}
}
+}
