/// Run this dispatcher until HTTP says this connection is done, /// but don't call `AsyncWrite::shutdown` on the underlying IO. /// /// This is useful for old-style HTTP upgrades, but ignores /// newer-style upgrade API. pub(crate) fn poll_without_shutdown(
&mutself,
cx: &mut task::Context<'_>,
) -> Poll<crate::Result<()>> where Self: Unpin,
{
Pin::new(self).poll_catch(cx, false).map_ok(|ds| { iflet Dispatched::Upgrade(pending) = ds {
pending.manual();
}
})
}
fn poll_catch(
&mutself,
cx: &mut task::Context<'_>,
should_shutdown: bool,
) -> Poll<crate::Result<Dispatched>> {
Poll::Ready(ready!(self.poll_inner(cx, should_shutdown)).or_else(|e| { // An error means we're shutting down either way. // We just try to give the error to the user, // and close the connection with an Ok. If we // cannot give it to the user, then return the Err. self.dispatch.recv_msg(Err(e))?;
Ok(Dispatched::Shutdown)
}))
}
fn poll_loop(&mutself, cx: &mut task::Context<'_>) -> Poll<crate::Result<()>> { // Limit the looping on this connection, in case it is ready far too // often, so that other futures don't starve. // // 16 was chosen arbitrarily, as that is number of pipelined requests // benchmarks often use. Perhaps it should be a config option instead. for _ in0..16 { let _ = self.poll_read(cx)?; let _ = self.poll_write(cx)?; let _ = self.poll_flush(cx)?;
// This could happen if reading paused before blocking on IO, // such as getting to the end of a framed message, but then // writing/flushing set the state back to Init. In that case, // if the read buffer still had bytes, we'd want to try poll_read // again, or else we wouldn't ever be woken up again. // // Using this instead of task::current() and notify() inside // the Conn is noticeably faster in pipelined benchmarks. if !self.conn.wants_read_again() { //break; return Poll::Ready(Ok(()));
}
}
trace!("poll_loop yielding (self = {:p})", self);
task::yield_now(cx).map(|never| match never {})
}
fn poll_read(&mutself, cx: &mut task::Context<'_>) -> Poll<crate::Result<()>> { loop { ifself.is_closing { return Poll::Ready(Ok(()));
} elseifself.conn.can_read_head() {
ready!(self.poll_read_head(cx))?;
} elseiflet Some(mut body) = self.body_tx.take() { ifself.conn.can_read_body() { match body.poll_ready(cx) {
Poll::Ready(Ok(())) => (),
Poll::Pending => { self.body_tx = Some(body); return Poll::Pending;
}
Poll::Ready(Err(_canceled)) => { // user doesn't care about the body // so we should stop reading
trace!("body receiver dropped before eof, draining or closing"); self.conn.poll_drain_or_close_read(cx); continue;
}
} matchself.conn.poll_read_body(cx) {
Poll::Ready(Some(Ok(chunk))) => match body.try_send_data(chunk) {
Ok(()) => { self.body_tx = Some(body);
}
Err(_canceled) => { ifself.conn.can_read_body() {
trace!("body receiver dropped before eof, closing"); self.conn.close_read();
}
}
},
Poll::Ready(None) => { // just drop, the body will close automatically
}
Poll::Pending => { self.body_tx = Some(body); return Poll::Pending;
}
Poll::Ready(Some(Err(e))) => {
body.send_error(crate::Error::new_body(e));
}
}
} else { // just drop, the body will close automatically
}
} else { returnself.conn.poll_read_keep_alive(cx);
}
}
}
fn poll_read_head(&mutself, cx: &mut task::Context<'_>) -> Poll<crate::Result<()>> { // can dispatch receive, or does it still care about, an incoming message? match ready!(self.dispatch.poll_ready(cx)) {
Ok(()) => (),
Err(()) => {
trace!("dispatch no longer receiving messages"); self.close(); return Poll::Ready(Ok(()));
}
} // dispatch is ready for a message, try to read one match ready!(self.conn.poll_read_head(cx)) {
Some(Ok((mut head, body_len, wants))) => { let body = match body_len {
DecodedLength::ZERO => Body::empty(),
other => { let (tx, rx) = Body::new_channel(other, wants.contains(Wants::EXPECT)); self.body_tx = Some(tx);
rx
}
}; if wants.contains(Wants::UPGRADE) { let upgrade = self.conn.on_upgrade();
debug_assert!(!upgrade.is_none(), "empty upgrade");
debug_assert!(head.extensions.get::<OnUpgrade>().is_none(), "OnUpgrade already set");
head.extensions.insert(upgrade);
} self.dispatch.recv_msg(Ok((head, body)))?;
Poll::Ready(Ok(()))
}
Some(Err(err)) => {
debug!("read_head error: {}", err); self.dispatch.recv_msg(Err(err))?; // if here, the dispatcher gave the user the error // somewhere else. we still need to shutdown, but // not as a second error. self.close();
Poll::Ready(Ok(()))
}
None => { // read eof, the write side will have been closed too unless // allow_read_close was set to true, in which case just do // nothing...
debug_assert!(self.conn.is_read_closed()); ifself.conn.is_write_closed() { self.close();
}
Poll::Ready(Ok(()))
}
}
}
// Check if the body knows its full data immediately. // // If so, we can skip a bit of bookkeeping that streaming // bodies need to do. iflet Some(full) = crate::body::take_full_data(&mut body) { self.conn.write_full_msg(head, full); return Poll::Ready(Ok(()));
}
let body_type = if body.is_end_stream() { self.body_rx.set(None);
None
} else { let btype = body
.size_hint()
.exact()
.map(BodyLength::Known)
.or_else(|| Some(BodyLength::Unknown)); self.body_rx.set(Some(body));
btype
}; self.conn.write_head(head, body_type);
} else { self.close(); return Poll::Ready(Ok(()));
}
} elseif !self.conn.can_buffer_body() {
ready!(self.poll_flush(cx))?;
} else { // A new scope is needed :( iflet (Some(mut body), clear_body) =
OptGuard::new(self.body_rx.as_mut()).guard_mut()
{
debug_assert!(!*clear_body, "opt guard defaults to keeping body"); if !self.conn.can_write_body() {
trace!( "no more write body allowed, user body is_end_stream = {}",
body.is_end_stream(),
);
*clear_body = true; continue;
}
let item = ready!(body.as_mut().poll_data(cx)); iflet Some(item) = item { let chunk = item.map_err(|e| {
*clear_body = true; crate::Error::new_user_body(e)
})?; let eos = body.is_end_stream(); if eos {
*clear_body = true; if chunk.remaining() == 0 {
trace!("discarding empty chunk"); self.conn.end_body()?;
} else { self.conn.write_body_and_end(chunk);
}
} else { if chunk.remaining() == 0 {
trace!("discarding empty chunk"); continue;
} self.conn.write_body(chunk);
}
} else {
*clear_body = true; self.conn.end_body()?;
}
} else { return Poll::Pending;
}
}
}
}
if !T::should_read_first() && read_done { // a client that cannot read may was well be done. true
} else { let write_done = self.conn.is_write_closed()
|| (!self.dispatch.should_poll() && self.body_rx.is_none());
read_done && write_done
}
}
}
/// A drop guard to allow a mutable borrow of an Option while being able to /// set whether the `Option` should be cleared on drop. struct OptGuard<'a, T>(Pin<&'a mut Option<T>>, bool);
impl<B> Dispatch for Client<B> where
B: HttpBody,
{ type PollItem = RequestHead; type PollBody = B; type PollError = crate::common::Never; type RecvItem = crate::proto::ResponseHead;
fn poll_msg( mutself: Pin<&mutSelf>,
cx: &mut task::Context<'_>,
) -> Poll<Option<Result<(Self::PollItem, Self::PollBody), crate::common::Never>>> { letmut this = self.as_mut();
debug_assert!(!this.rx_closed); match this.rx.poll_recv(cx) {
Poll::Ready(Some((req, mut cb))) => { // check that future hasn't been canceled already match cb.poll_canceled(cx) {
Poll::Ready(()) => {
trace!("request canceled");
Poll::Ready(None)
}
Poll::Pending => { let (parts, body) = req.into_parts(); let head = RequestHead {
version: parts.version,
subject: crate::proto::RequestLine(parts.method, parts.uri),
headers: parts.headers,
extensions: parts.extensions,
};
this.callback = Some(cb);
Poll::Ready(Some(Ok((head, body))))
}
}
}
Poll::Ready(None) => { // user has dropped sender handle
trace!("client tx closed");
this.rx_closed = true;
Poll::Ready(None)
}
Poll::Pending => Poll::Pending,
}
}
fn recv_msg(&mutself, msg: crate::Result<(Self::RecvItem, Body)>) -> crate::Result<()> { match msg {
Ok((msg, body)) => { iflet Some(cb) = self.callback.take() { let res = msg.into_response(body);
cb.send(Ok(res));
Ok(())
} else { // Getting here is likely a bug! An error should have happened // in Conn::require_empty_read() before ever parsing a // full message!
Err(crate::Error::new_unexpected_message())
}
}
Err(err) => { iflet Some(cb) = self.callback.take() {
cb.send(Err((err, None)));
Ok(())
} elseif !self.rx_closed { self.rx.close(); iflet Some((req, cb)) = self.rx.try_recv() {
trace!("canceling queued request with connection error: {}", err); // in this case, the message was never even started, so it's safe to tell // the user that the request was completely canceled
cb.send(Err((crate::Error::new_canceled().with(err), Some(req))));
Ok(())
} else {
Err(err)
}
} else {
Err(err)
}
}
}
}
#[cfg(test)] mod tests { usesuper::*; usecrate::proto::h1::ClientTransaction; use std::time::Duration;
#[test] fn client_read_bytes_before_writing_request() { let _ = pretty_env_logger::try_init();
tokio_test::task::spawn(()).enter(|cx, _| { let (io, mut handle) = tokio_test::io::Builder::new().build_with_handle();
// Block at 0 for now, but we will release this response before // the request is ready to write later... let (mut tx, rx) = crate::client::dispatch::channel(); let conn = Conn::<_, bytes::Bytes, ClientTransaction>::new(io); letmut dispatcher = Dispatcher::new(Client::new(rx), conn);
// First poll is needed to allow tx to send...
assert!(Pin::new(&mut dispatcher).poll(cx).is_pending());
// Unblock our IO, which has a response before we've sent request! //
handle.read(b"HTTP/1.1 200 OK\r\n\r\n");
Die Informationen auf dieser Webseite wurden
nach bestem Wissen sorgfältig zusammengestellt. Es wird jedoch weder Vollständigkeit, noch Richtigkeit,
noch Qualität der bereit gestellten Informationen zugesichert.
Bemerkung:
Die farbliche Syntaxdarstellung und die Messung sind noch experimentell.
