actix/actix-net
1
0
Fork 0
mirror of https://github.com/fafhrd91/actix-net synced 2025-01-31 09:12:08 +01:00
Code Issues Releases Wiki Activity

Merge pull request #110 from Aaron1011/fix/better-pin

Browse Source 
Replace calls to `Pin::new_unchecked` with `pin_project`.
This commit is contained in:
Yuki Okushi 2020-03-05 21:52:55 +09:00 committed by GitHub
commit 8d3d58b3b7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 116 additions and 104 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
actix-codec/Cargo.toml
View File

@ -23,4 +23,5 @@ futures-core = "0.3.1"
futures-sink = "0.3.1" futures-sink = "0.3.1"
tokio = { version = "0.2.4", default-features=false } tokio = { version = "0.2.4", default-features=false }
tokio-util = { version = "0.2.0", default-features=false, features=["codec"] } tokio-util = { version = "0.2.0", default-features=false, features=["codec"] }
log = "0.4" log = "0.4"
pin-project = "0.4.8"

75
actix-codec/src/framed.rs
View File

@ -5,6 +5,7 @@ use std::{fmt, io};
use bytes::{Buf, BytesMut}; use bytes::{Buf, BytesMut};
use futures_core::{ready, Stream}; use futures_core::{ready, Stream};
use futures_sink::Sink; use futures_sink::Sink;
use pin_project::pin_project;
use crate::{AsyncRead, AsyncWrite, Decoder, Encoder}; use crate::{AsyncRead, AsyncWrite, Decoder, Encoder};
@ -20,7 +21,9 @@ bitflags::bitflags! {
/// A unified `Stream` and `Sink` interface to an underlying I/O object, using /// A unified `Stream` and `Sink` interface to an underlying I/O object, using
/// the `Encoder` and `Decoder` traits to encode and decode frames. /// the `Encoder` and `Decoder` traits to encode and decode frames.
#[pin_project]
pub struct Framed<T, U> { pub struct Framed<T, U> {
#[pin]
io: T, io: T,
codec: U, codec: U,
flags: Flags, flags: Flags,
@ -28,8 +31,6 @@ pub struct Framed<T, U> {
write_buf: BytesMut, write_buf: BytesMut,
} }
impl<T, U> Unpin for Framed<T, U> {}
impl<T, U> Framed<T, U> impl<T, U> Framed<T, U>
where where
T: AsyncRead + AsyncWrite, T: AsyncRead + AsyncWrite,
@ -185,17 +186,18 @@ impl<T, U> Framed<T, U> {
impl<T, U> Framed<T, U> { impl<T, U> Framed<T, U> {
/// Serialize item and Write to the inner buffer /// Serialize item and Write to the inner buffer
pub fn write(&mut self, item: <U as Encoder>::Item) -> Result<(), <U as Encoder>::Error> pub fn write(mut self: Pin<&mut Self>, item: <U as Encoder>::Item) -> Result<(), <U as Encoder>::Error>
where where
T: AsyncWrite, T: AsyncWrite,
U: Encoder, U: Encoder,
{ {
let remaining = self.write_buf.capacity() - self.write_buf.len(); let this = self.as_mut().project();
let remaining = this.write_buf.capacity() - this.write_buf.len();
if remaining < LW { if remaining < LW {
self.write_buf.reserve(HW - remaining); this.write_buf.reserve(HW - remaining);
} }
self.codec.encode(item, &mut self.write_buf)?; this.codec.encode(item, this.write_buf)?;
Ok(()) Ok(())
} }
@ -207,21 +209,22 @@ impl<T, U> Framed<T, U> {
} }
/// Try to read underlying I/O stream and decode item. /// Try to read underlying I/O stream and decode item.
pub fn next_item(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<U::Item, U::Error>>> pub fn next_item(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Result<U::Item, U::Error>>>
where where
T: AsyncRead, T: AsyncRead,
U: Decoder, U: Decoder,
{ {
loop { loop {
let mut this = self.as_mut().project();
// Repeatedly call `decode` or `decode_eof` as long as it is // Repeatedly call `decode` or `decode_eof` as long as it is
// "readable". Readable is defined as not having returned `None`. If // "readable". Readable is defined as not having returned `None`. If
// the upstream has returned EOF, and the decoder is no longer // the upstream has returned EOF, and the decoder is no longer
// readable, it can be assumed that the decoder will never become // readable, it can be assumed that the decoder will never become
// readable again, at which point the stream is terminated. // readable again, at which point the stream is terminated.
if self.flags.contains(Flags::READABLE) { if this.flags.contains(Flags::READABLE) {
if self.flags.contains(Flags::EOF) { if this.flags.contains(Flags::EOF) {
match self.codec.decode_eof(&mut self.read_buf) { match this.codec.decode_eof(&mut this.read_buf) {
Ok(Some(frame)) => return Poll::Ready(Some(Ok(frame))), Ok(Some(frame)) => return Poll::Ready(Some(Ok(frame))),
Ok(None) => return Poll::Ready(None), Ok(None) => return Poll::Ready(None),
Err(e) => return Poll::Ready(Some(Err(e))), Err(e) => return Poll::Ready(Some(Err(e))),
@ -230,7 +233,7 @@ impl<T, U> Framed<T, U> {
log::trace!("attempting to decode a frame"); log::trace!("attempting to decode a frame");
match self.codec.decode(&mut self.read_buf) { match this.codec.decode(&mut this.read_buf) {
Ok(Some(frame)) => { Ok(Some(frame)) => {
log::trace!("frame decoded from buffer"); log::trace!("frame decoded from buffer");
return Poll::Ready(Some(Ok(frame))); return Poll::Ready(Some(Ok(frame)));
@ -239,45 +242,44 @@ impl<T, U> Framed<T, U> {
_ => (), // Need more data _ => (), // Need more data
} }
self.flags.remove(Flags::READABLE); this.flags.remove(Flags::READABLE);
} }
debug_assert!(!self.flags.contains(Flags::EOF)); debug_assert!(!this.flags.contains(Flags::EOF));
// Otherwise, try to read more data and try again. Make sure we've got room // Otherwise, try to read more data and try again. Make sure we've got room
let remaining = self.read_buf.capacity() - self.read_buf.len(); let remaining = this.read_buf.capacity() - this.read_buf.len();
if remaining < LW { if remaining < LW {
self.read_buf.reserve(HW - remaining) this.read_buf.reserve(HW - remaining)
} }
let cnt = match unsafe { let cnt = match this.io.poll_read_buf(cx, &mut this.read_buf) {
Pin::new_unchecked(&mut self.io).poll_read_buf(cx, &mut self.read_buf)
} {
Poll::Pending => return Poll::Pending, Poll::Pending => return Poll::Pending,
Poll::Ready(Err(e)) => return Poll::Ready(Some(Err(e.into()))), Poll::Ready(Err(e)) => return Poll::Ready(Some(Err(e.into()))),
Poll::Ready(Ok(cnt)) => cnt, Poll::Ready(Ok(cnt)) => cnt,
}; };
if cnt == 0 { if cnt == 0 {
self.flags.insert(Flags::EOF); this.flags.insert(Flags::EOF);
} }
self.flags.insert(Flags::READABLE); this.flags.insert(Flags::READABLE);
} }
} }
/// Flush write buffer to underlying I/O stream. /// Flush write buffer to underlying I/O stream.
pub fn flush(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), U::Error>> pub fn flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), U::Error>>
where where
T: AsyncWrite, T: AsyncWrite,
U: Encoder, U: Encoder,
{ {
let mut this = self.as_mut().project();
log::trace!("flushing framed transport"); log::trace!("flushing framed transport");
while !self.write_buf.is_empty() { while !this.write_buf.is_empty() {
log::trace!("writing; remaining={}", self.write_buf.len()); log::trace!("writing; remaining={}", this.write_buf.len());
let n = ready!(unsafe { let n = ready!(
Pin::new_unchecked(&mut self.io).poll_write(cx, &self.write_buf) this.io.as_mut().poll_write(cx, this.write_buf)
})?; )?;
if n == 0 { if n == 0 {
return Poll::Ready(Err(io::Error::new( return Poll::Ready(Err(io::Error::new(
@ -288,26 +290,25 @@ impl<T, U> Framed<T, U> {
} }
// remove written data // remove written data
self.write_buf.advance(n); this.write_buf.advance(n);
} }
// Try flushing the underlying IO // Try flushing the underlying IO
ready!(unsafe { Pin::new_unchecked(&mut self.io).poll_flush(cx) })?; ready!(this.io.poll_flush(cx))?;
log::trace!("framed transport flushed"); log::trace!("framed transport flushed");
Poll::Ready(Ok(())) Poll::Ready(Ok(()))
} }
/// Flush write buffer and shutdown underlying I/O stream. /// Flush write buffer and shutdown underlying I/O stream.
pub fn close(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), U::Error>> pub fn close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), U::Error>>
where where
T: AsyncWrite, T: AsyncWrite,
U: Encoder, U: Encoder,
{ {
unsafe { let mut this = self.as_mut().project();
ready!(Pin::new_unchecked(&mut self.io).poll_flush(cx))?; ready!(this.io.as_mut().poll_flush(cx))?;
ready!(Pin::new_unchecked(&mut self.io).poll_shutdown(cx))?; ready!(this.io.as_mut().poll_shutdown(cx))?;
}
Poll::Ready(Ok(())) Poll::Ready(Ok(()))
} }
} }
@ -319,7 +320,7 @@ where
{ {
type Item = Result<U::Item, U::Error>; type Item = Result<U::Item, U::Error>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
self.next_item(cx) self.next_item(cx)
} }
} }
@ -341,21 +342,21 @@ where
} }
fn start_send( fn start_send(
mut self: Pin<&mut Self>, self: Pin<&mut Self>,
item: <U as Encoder>::Item, item: <U as Encoder>::Item,
) -> Result<(), Self::Error> { ) -> Result<(), Self::Error> {
self.write(item) self.write(item)
} }
fn poll_flush( fn poll_flush(
mut self: Pin<&mut Self>, self: Pin<&mut Self>,
cx: &mut Context<'_>, cx: &mut Context<'_>,
) -> Poll<Result<(), Self::Error>> { ) -> Poll<Result<(), Self::Error>> {
self.flush(cx) self.flush(cx)
} }
fn poll_close( fn poll_close(
mut self: Pin<&mut Self>, self: Pin<&mut Self>,
cx: &mut Context<'_>, cx: &mut Context<'_>,
) -> Poll<Result<(), Self::Error>> { ) -> Poll<Result<(), Self::Error>> {
self.close(cx) self.close(cx)

13
actix-ioframe/src/connect.rs
View File

@ -42,6 +42,7 @@ where
pub struct ConnectResult<Io, St, Codec: Encoder + Decoder, Out> { pub struct ConnectResult<Io, St, Codec: Encoder + Decoder, Out> {
pub(crate) state: St, pub(crate) state: St,
pub(crate) out: Option<Out>, pub(crate) out: Option<Out>,
#[pin]
pub(crate) framed: Framed<Io, Codec>, pub(crate) framed: Framed<Io, Codec>,
} }
@ -97,8 +98,8 @@ where
{ {
type Error = <Codec as Encoder>::Error; type Error = <Codec as Encoder>::Error;
fn poll_ready(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> { fn poll_ready(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
if self.framed.is_write_ready() { if self.as_mut().project().framed.is_write_ready() {
Poll::Ready(Ok(())) Poll::Ready(Ok(()))
} else { } else {
Poll::Pending Poll::Pending
@ -112,11 +113,11 @@ where
self.project().framed.write(item) self.project().framed.write(item)
} }
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> { fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.get_mut().framed.flush(cx) self.as_mut().project().framed.flush(cx)
} }
fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> { fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.get_mut().framed.close(cx) self.as_mut().project().framed.close(cx)
} }
} }

76
actix-ioframe/src/dispatcher.rs
View File

@ -6,6 +6,7 @@ use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
use actix_service::Service; use actix_service::Service;
use actix_utils::mpsc; use actix_utils::mpsc;
use futures::Stream; use futures::Stream;
use pin_project::pin_project;
use log::debug; use log::debug;
use crate::error::ServiceError; use crate::error::ServiceError;
@ -15,6 +16,7 @@ type Response<U> = <U as Encoder>::Item;
/// FramedTransport - is a future that reads frames from Framed object /// FramedTransport - is a future that reads frames from Framed object
/// and pass then to the service. /// and pass then to the service.
#[pin_project]
pub(crate) struct Dispatcher<S, T, U, Out> pub(crate) struct Dispatcher<S, T, U, Out>
where where
S: Service<Request = Request<U>, Response = Option<Response<U>>>, S: Service<Request = Request<U>, Response = Option<Response<U>>>,
@ -29,6 +31,7 @@ where
service: S, service: S,
sink: Option<Out>, sink: Option<Out>,
state: FramedState<S, U>, state: FramedState<S, U>,
#[pin]
framed: Framed<T, U>, framed: Framed<T, U>,
rx: mpsc::Receiver<Result<<U as Encoder>::Item, S::Error>>, rx: mpsc::Receiver<Result<<U as Encoder>::Item, S::Error>>,
} }
@ -90,26 +93,27 @@ where
<U as Encoder>::Error: std::fmt::Debug, <U as Encoder>::Error: std::fmt::Debug,
Out: Stream<Item = <U as Encoder>::Item> + Unpin, Out: Stream<Item = <U as Encoder>::Item> + Unpin,
{ {
fn poll_read(&mut self, cx: &mut Context<'_>) -> bool { fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> bool {
loop { loop {
match self.service.poll_ready(cx) { let this = self.as_mut().project();
match this.service.poll_ready(cx) {
Poll::Ready(Ok(_)) => { Poll::Ready(Ok(_)) => {
let item = match self.framed.next_item(cx) { let item = match this.framed.next_item(cx) {
Poll::Ready(Some(Ok(el))) => el, Poll::Ready(Some(Ok(el))) => el,
Poll::Ready(Some(Err(err))) => { Poll::Ready(Some(Err(err))) => {
self.state = FramedState::FramedError(ServiceError::Decoder(err)); *this.state = FramedState::FramedError(ServiceError::Decoder(err));
return true; return true;
} }
Poll::Pending => return false, Poll::Pending => return false,
Poll::Ready(None) => { Poll::Ready(None) => {
log::trace!("Client disconnected"); log::trace!("Client disconnected");
self.state = FramedState::Stopping; *this.state = FramedState::Stopping;
return true; return true;
} }
}; };
let tx = self.rx.sender(); let tx = this.rx.sender();
let fut = self.service.call(item); let fut = this.service.call(item);
actix_rt::spawn(async move { actix_rt::spawn(async move {
let item = fut.await; let item = fut.await;
let item = match item { let item = match item {
@ -122,7 +126,7 @@ where
} }
Poll::Pending => return false, Poll::Pending => return false,
Poll::Ready(Err(err)) => { Poll::Ready(Err(err)) => {
self.state = FramedState::Error(ServiceError::Service(err)); *this.state = FramedState::Error(ServiceError::Service(err));
return true; return true;
} }
} }
@ -130,37 +134,38 @@ where
} }
/// write to framed object /// write to framed object
fn poll_write(&mut self, cx: &mut Context<'_>) -> bool { fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> bool {
loop { loop {
while !self.framed.is_write_buf_full() { let mut this = self.as_mut().project();
match Pin::new(&mut self.rx).poll_next(cx) { while !this.framed.is_write_buf_full() {
match Pin::new(&mut this.rx).poll_next(cx) {
Poll::Ready(Some(Ok(msg))) => { Poll::Ready(Some(Ok(msg))) => {
if let Err(err) = self.framed.write(msg) { if let Err(err) = this.framed.as_mut().write(msg) {
self.state = FramedState::FramedError(ServiceError::Encoder(err)); *this.state = FramedState::FramedError(ServiceError::Encoder(err));
return true; return true;
} }
continue; continue;
} }
Poll::Ready(Some(Err(err))) => { Poll::Ready(Some(Err(err))) => {
self.state = FramedState::Error(ServiceError::Service(err)); *this.state = FramedState::Error(ServiceError::Service(err));
return true; return true;
} }
Poll::Ready(None) | Poll::Pending => (), Poll::Ready(None) | Poll::Pending => (),
} }
if self.sink.is_some() { if this.sink.is_some() {
match Pin::new(self.sink.as_mut().unwrap()).poll_next(cx) { match Pin::new(this.sink.as_mut().unwrap()).poll_next(cx) {
Poll::Ready(Some(msg)) => { Poll::Ready(Some(msg)) => {
if let Err(err) = self.framed.write(msg) { if let Err(err) = this.framed.as_mut().write(msg) {
self.state = *this.state =
FramedState::FramedError(ServiceError::Encoder(err)); FramedState::FramedError(ServiceError::Encoder(err));
return true; return true;
} }
continue; continue;
} }
Poll::Ready(None) => { Poll::Ready(None) => {
let _ = self.sink.take(); let _ = this.sink.take();
self.state = FramedState::FlushAndStop; *this.state = FramedState::FlushAndStop;
return true; return true;
} }
Poll::Pending => (), Poll::Pending => (),
@ -169,13 +174,13 @@ where
break; break;
} }
if !self.framed.is_write_buf_empty() { if !this.framed.is_write_buf_empty() {
match self.framed.flush(cx) { match this.framed.as_mut().flush(cx) {
Poll::Pending => break, Poll::Pending => break,
Poll::Ready(Ok(_)) => (), Poll::Ready(Ok(_)) => (),
Poll::Ready(Err(err)) => { Poll::Ready(Err(err)) => {
debug!("Error sending data: {:?}", err); debug!("Error sending data: {:?}", err);
self.state = FramedState::FramedError(ServiceError::Encoder(err)); *this.state = FramedState::FramedError(ServiceError::Encoder(err));
return true; return true;
} }
} }
@ -187,13 +192,14 @@ where
} }
pub(crate) fn poll( pub(crate) fn poll(
&mut self, mut self: Pin<&mut Self>,
cx: &mut Context<'_>, cx: &mut Context<'_>,
) -> Poll<Result<(), ServiceError<S::Error, U>>> { ) -> Poll<Result<(), ServiceError<S::Error, U>>> {
match self.state { let mut this = self.as_mut().project();
match this.state {
FramedState::Processing => loop { FramedState::Processing => loop {
let read = self.poll_read(cx); let read = self.as_mut().poll_read(cx);
let write = self.poll_write(cx); let write = self.as_mut().poll_write(cx);
if read || write { if read || write {
continue; continue;
} else { } else {
@ -202,18 +208,18 @@ where
}, },
FramedState::Error(_) => { FramedState::Error(_) => {
// flush write buffer // flush write buffer
if !self.framed.is_write_buf_empty() { if !this.framed.is_write_buf_empty() {
if let Poll::Pending = self.framed.flush(cx) { if let Poll::Pending = this.framed.flush(cx) {
return Poll::Pending; return Poll::Pending;
} }
} }
Poll::Ready(Err(self.state.take_error())) Poll::Ready(Err(this.state.take_error()))
} }
FramedState::FlushAndStop => { FramedState::FlushAndStop => {
// drain service responses // drain service responses
match Pin::new(&mut self.rx).poll_next(cx) { match Pin::new(this.rx).poll_next(cx) {
Poll::Ready(Some(Ok(msg))) => { Poll::Ready(Some(Ok(msg))) => {
if self.framed.write(msg).is_err() { if this.framed.as_mut().write(msg).is_err() {
return Poll::Ready(Ok(())); return Poll::Ready(Ok(()));
} }
} }
@ -222,8 +228,8 @@ where
} }
// flush io // flush io
if !self.framed.is_write_buf_empty() { if !this.framed.is_write_buf_empty() {
match self.framed.flush(cx) { match this.framed.flush(cx) {
Poll::Ready(Err(err)) => { Poll::Ready(Err(err)) => {
debug!("Error sending data: {:?}", err); debug!("Error sending data: {:?}", err);
} }
@ -235,7 +241,7 @@ where
}; };
Poll::Ready(Ok(())) Poll::Ready(Ok(()))
} }
FramedState::FramedError(_) => Poll::Ready(Err(self.state.take_framed_error())), FramedState::FramedError(_) => Poll::Ready(Err(this.state.take_framed_error())),
FramedState::Stopping => Poll::Ready(Ok(())), FramedState::Stopping => Poll::Ready(Ok(())),
} }
} }

4
actix-ioframe/src/service.rs
View File

@ -357,7 +357,7 @@ where
{ {
Connect(#[pin] C::Future, Rc<T>), Connect(#[pin] C::Future, Rc<T>),
Handler(#[pin] T::Future, Option<Framed<Io, Codec>>, Option<Out>), Handler(#[pin] T::Future, Option<Framed<Io, Codec>>, Option<Out>),
Dispatcher(Dispatcher<T::Service, Io, Codec, Out>), Dispatcher(#[pin] Dispatcher<T::Service, Io, Codec, Out>),
} }
impl<St, Io, Codec, Out, C, T> FramedServiceImplResponseInner<St, Io, Codec, Out, C, T> impl<St, Io, Codec, Out, C, T> FramedServiceImplResponseInner<St, Io, Codec, Out, C, T>
@ -408,7 +408,7 @@ where
Poll::Ready(Err(e)) => Either::Right(Poll::Ready(Err(e.into()))), Poll::Ready(Err(e)) => Either::Right(Poll::Ready(Err(e.into()))),
} }
} }
FramedServiceImplResponseInner::Dispatcher(ref mut fut) => { FramedServiceImplResponseInner::Dispatcher(fut) => {
Either::Right(fut.poll(cx)) Either::Right(fut.poll(cx))
} }
} }

2
actix-rt/src/arbiter.rs
View File

@ -181,7 +181,7 @@ impl Arbiter {
// because the executor boxes the future again, but works for now // because the executor boxes the future again, but works for now
Q.with(move |cell| { Q.with(move |cell| {
cell.borrow_mut() cell.borrow_mut()
.push(unsafe { Pin::new_unchecked(Box::alloc().init(future)) }) .push(Pin::from(Box::alloc().init(future)))
}); });
} }
}); });

47
actix-utils/src/framed.rs
View File

@ -77,6 +77,7 @@ where
{ {
service: S, service: S,
state: State<S, U>, state: State<S, U>,
#[pin]
framed: Framed<T, U>, framed: Framed<T, U>,
rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, S::Error>>, rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, S::Error>>,
tx: mpsc::Sender<Result<Message<<U as Encoder>::Item>, S::Error>>, tx: mpsc::Sender<Result<Message<<U as Encoder>::Item>, S::Error>>,
@ -169,7 +170,7 @@ where
&mut self.framed &mut self.framed
} }
fn poll_read(&mut self, cx: &mut Context<'_>) -> bool fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> bool
where where
S: Service<Request = Request<U>, Response = Response<U>>, S: Service<Request = Request<U>, Response = Response<U>>,
S::Error: 'static, S::Error: 'static,
@ -180,29 +181,30 @@ where
<U as Encoder>::Error: std::fmt::Debug, <U as Encoder>::Error: std::fmt::Debug,
{ {
loop { loop {
match self.service.poll_ready(cx) { let this = self.as_mut().project();
match this.service.poll_ready(cx) {
Poll::Ready(Ok(_)) => { Poll::Ready(Ok(_)) => {
let item = match self.framed.next_item(cx) { let item = match this.framed.next_item(cx) {
Poll::Ready(Some(Ok(el))) => el, Poll::Ready(Some(Ok(el))) => el,
Poll::Ready(Some(Err(err))) => { Poll::Ready(Some(Err(err))) => {
self.state = State::FramedError(DispatcherError::Decoder(err)); *this.state = State::FramedError(DispatcherError::Decoder(err));
return true; return true;
} }
Poll::Pending => return false, Poll::Pending => return false,
Poll::Ready(None) => { Poll::Ready(None) => {
self.state = State::Stopping; *this.state = State::Stopping;
return true; return true;
} }
}; };
let tx = self.tx.clone(); let tx = this.tx.clone();
actix_rt::spawn(self.service.call(item).map(move |item| { actix_rt::spawn(this.service.call(item).map(move |item| {
let _ = tx.send(item.map(Message::Item)); let _ = tx.send(item.map(Message::Item));
})); }));
} }
Poll::Pending => return false, Poll::Pending => return false,
Poll::Ready(Err(err)) => { Poll::Ready(Err(err)) => {
self.state = State::Error(DispatcherError::Service(err)); *this.state = State::Error(DispatcherError::Service(err));
return true; return true;
} }
} }
@ -210,7 +212,7 @@ where
} }
/// write to framed object /// write to framed object
fn poll_write(&mut self, cx: &mut Context<'_>) -> bool fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> bool
where where
S: Service<Request = Request<U>, Response = Response<U>>, S: Service<Request = Request<U>, Response = Response<U>>,
S::Error: 'static, S::Error: 'static,
@ -221,33 +223,34 @@ where
<U as Encoder>::Error: std::fmt::Debug, <U as Encoder>::Error: std::fmt::Debug,
{ {
loop { loop {
while !self.framed.is_write_buf_full() { let mut this = self.as_mut().project();
match Pin::new(&mut self.rx).poll_next(cx) { while !this.framed.is_write_buf_full() {
match Pin::new(&mut this.rx).poll_next(cx) {
Poll::Ready(Some(Ok(Message::Item(msg)))) => { Poll::Ready(Some(Ok(Message::Item(msg)))) => {
if let Err(err) = self.framed.write(msg) { if let Err(err) = this.framed.as_mut().write(msg) {
self.state = State::FramedError(DispatcherError::Encoder(err)); *this.state = State::FramedError(DispatcherError::Encoder(err));
return true; return true;
} }
} }
Poll::Ready(Some(Ok(Message::Close))) => { Poll::Ready(Some(Ok(Message::Close))) => {
self.state = State::FlushAndStop; *this.state = State::FlushAndStop;
return true; return true;
} }
Poll::Ready(Some(Err(err))) => { Poll::Ready(Some(Err(err))) => {
self.state = State::Error(DispatcherError::Service(err)); *this.state = State::Error(DispatcherError::Service(err));
return true; return true;
} }
Poll::Ready(None) | Poll::Pending => break, Poll::Ready(None) | Poll::Pending => break,
} }
} }
if !self.framed.is_write_buf_empty() { if !this.framed.is_write_buf_empty() {
match self.framed.flush(cx) { match this.framed.flush(cx) {
Poll::Pending => break, Poll::Pending => break,
Poll::Ready(Ok(_)) => (), Poll::Ready(Ok(_)) => (),
Poll::Ready(Err(err)) => { Poll::Ready(Err(err)) => {
debug!("Error sending data: {:?}", err); debug!("Error sending data: {:?}", err);
self.state = State::FramedError(DispatcherError::Encoder(err)); *this.state = State::FramedError(DispatcherError::Encoder(err));
return true; return true;
} }
} }
@ -279,7 +282,7 @@ where
return match this.state { return match this.state {
State::Processing => { State::Processing => {
if self.poll_read(cx) || self.poll_write(cx) { if self.as_mut().poll_read(cx) || self.as_mut().poll_write(cx) {
continue; continue;
} else { } else {
Poll::Pending Poll::Pending
@ -287,12 +290,12 @@ where
} }
State::Error(_) => { State::Error(_) => {
// flush write buffer // flush write buffer
if !self.framed.is_write_buf_empty() { if !this.framed.is_write_buf_empty() {
if let Poll::Pending = self.framed.flush(cx) { if let Poll::Pending = this.framed.flush(cx) {
return Poll::Pending; return Poll::Pending;
} }
} }
Poll::Ready(Err(self.state.take_error())) Poll::Ready(Err(this.state.take_error()))
} }
State::FlushAndStop => { State::FlushAndStop => {
if !this.framed.is_write_buf_empty() { if !this.framed.is_write_buf_empty() {