forked/hyper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(client): Make client an optional feature

Browse Source 
cc #2223

BREAKING CHANGE: The HTTP client of hyper is now an optional feature. To
  enable the client, add `features = ["client"]` to the dependency in
  your `Cargo.toml`.
This commit is contained in:
Sean McArthur
2020-11-17 17:06:25 -08:00
committed by GitHub
parent eb092a7b8c
commit 4e55583d30
19 changed files with 242 additions and 168 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/proto/h1/conn.rs
View File

@@ -1,5 +1,5 @@
use std::fmt;
use std::io::{self};
use std::io;
use std::marker::PhantomData;
use bytes::{Buf, Bytes};
@@ -65,6 +65,7 @@ where
self.io.set_max_buf_size(max);
}
#[cfg(feature = "client")]
pub fn set_read_buf_exact_size(&mut self, sz: usize) {
self.io.set_read_buf_exact_size(sz);
}
@@ -77,6 +78,7 @@ where
self.io.set_write_strategy_queue();
}
#[cfg(feature = "client")]
pub fn set_title_case_headers(&mut self) {
self.state.title_case_headers = true;
}

213
src/proto/h1/dispatch.rs
View File

@@ -1,15 +1,14 @@
use std::error::Error as StdError;
use bytes::{Buf, Bytes};
use http::{Request, Response, StatusCode};
use http::{Request, StatusCode};
use tokio::io::{AsyncRead, AsyncWrite};
use super::{Http1Transaction, Wants};
use crate::body::{Body, DecodedLength, HttpBody};
use crate::common::{task, Future, Never, Pin, Poll, Unpin};
use crate::common::{task, Future, Pin, Poll, Unpin};
use crate::proto::{
BodyLength, Conn, Dispatched, MessageHead, RequestHead, RequestLine,
ResponseHead,
BodyLength, Conn, Dispatched, MessageHead, RequestHead,
};
use crate::service::HttpService;
@@ -40,15 +39,17 @@ pub struct Server<S: HttpService<B>, B> {
pub(crate) service: S,
}
#[pin_project::pin_project]
pub struct Client<B> {
callback: Option<crate::client::dispatch::Callback<Request<B>, Response<Body>>>,
#[pin]
rx: ClientRx<B>,
rx_closed: bool,
}
cfg_client! {
#[pin_project::pin_project]
pub struct Client<B> {
callback: Option<crate::client::dispatch::Callback<Request<B>, http::Response<Body>>>,
#[pin]
rx: ClientRx<B>,
rx_closed: bool,
}
type ClientRx<B> = crate::client::dispatch::Receiver<Request<B>, Response<Body>>;
type ClientRx<B> = crate::client::dispatch::Receiver<Request<B>, http::Response<Body>>;
}
impl<D, Bs, I, T> Dispatcher<D, Bs, I, T>
where
@@ -523,115 +524,117 @@ where
// ===== impl Client =====
impl<B> Client<B> {
pub fn new(rx: ClientRx<B>) -> Client<B> {
Client {
callback: None,
rx,
rx_closed: false,
}
}
}
impl<B> Dispatch for Client<B>
where
B: HttpBody,
{
type PollItem = RequestHead;
type PollBody = B;
type PollError = Never;
type RecvItem = ResponseHead;
fn poll_msg(
self: Pin<&mut Self>,
cx: &mut task::Context<'_>,
) -> Poll<Option<Result<(Self::PollItem, Self::PollBody), Never>>> {
let this = self.project();
debug_assert!(!*this.rx_closed);
match this.rx.poll_next(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: RequestLine(parts.method, parts.uri),
headers: parts.headers,
};
*this.callback = Some(cb);
Poll::Ready(Some(Ok((head, body))))
}
}
cfg_client! {
impl<B> Client<B> {
pub fn new(rx: ClientRx<B>) -> Client<B> {
Client {
callback: None,
rx,
rx_closed: false,
}
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(&mut self, msg: crate::Result<(Self::RecvItem, Body)>) -> crate::Result<()> {
match msg {
Ok((msg, body)) => {
if let Some(cb) = self.callback.take() {
let mut res = Response::new(body);
*res.status_mut() = msg.subject;
*res.headers_mut() = msg.headers;
*res.version_mut() = msg.version;
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())
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(
self: Pin<&mut Self>,
cx: &mut task::Context<'_>,
) -> Poll<Option<Result<(Self::PollItem, Self::PollBody), crate::common::Never>>> {
let this = self.project();
debug_assert!(!*this.rx_closed);
match this.rx.poll_next(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,
};
*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,
}
Err(err) => {
if let Some(cb) = self.callback.take() {
cb.send(Err((err, None)));
Ok(())
} else if !self.rx_closed {
self.rx.close();
if let 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))));
}
fn recv_msg(&mut self, msg: crate::Result<(Self::RecvItem, Body)>) -> crate::Result<()> {
match msg {
Ok((msg, body)) => {
if let Some(cb) = self.callback.take() {
let mut res = http::Response::new(body);
*res.status_mut() = msg.subject;
*res.headers_mut() = msg.headers;
*res.version_mut() = msg.version;
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) => {
if let Some(cb) = self.callback.take() {
cb.send(Err((err, None)));
Ok(())
} else if !self.rx_closed {
self.rx.close();
if let 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)
}
} else {
Err(err)
}
}
}
}
fn poll_ready(&mut self, cx: &mut task::Context<'_>) -> Poll<Result<(), ()>> {
match self.callback {
Some(ref mut cb) => match cb.poll_canceled(cx) {
Poll::Ready(()) => {
trace!("callback receiver has dropped");
Poll::Ready(Err(()))
}
Poll::Pending => Poll::Ready(Ok(())),
},
None => Poll::Ready(Err(())),
fn poll_ready(&mut self, cx: &mut task::Context<'_>) -> Poll<Result<(), ()>> {
match self.callback {
Some(ref mut cb) => match cb.poll_canceled(cx) {
Poll::Ready(()) => {
trace!("callback receiver has dropped");
Poll::Ready(Err(()))
}
Poll::Pending => Poll::Ready(Ok(())),
},
None => Poll::Ready(Err(())),
}
}
}
fn should_poll(&self) -> bool {
self.callback.is_none()
fn should_poll(&self) -> bool {
self.callback.is_none()
}
}
}

57
src/proto/h1/io.rs
View File

@@ -84,6 +84,7 @@ where
self.write_buf.max_buf_size = max;
}
#[cfg(feature = "client")]
pub fn set_read_buf_exact_size(&mut self, sz: usize) {
self.read_buf_strategy = ReadStrategy::Exact(sz);
}
@@ -317,6 +318,7 @@ enum ReadStrategy {
next: usize,
max: usize,
},
#[cfg(feature = "client")]
Exact(usize),
}
@@ -332,6 +334,7 @@ impl ReadStrategy {
fn next(&self) -> usize {
match *self {
ReadStrategy::Adaptive { next, .. } => next,
#[cfg(feature = "client")]
ReadStrategy::Exact(exact) => exact,
}
}
@@ -339,38 +342,42 @@ impl ReadStrategy {
fn max(&self) -> usize {
match *self {
ReadStrategy::Adaptive { max, .. } => max,
#[cfg(feature = "client")]
ReadStrategy::Exact(exact) => exact,
}
}
fn record(&mut self, bytes_read: usize) {
if let ReadStrategy::Adaptive {
ref mut decrease_now,
ref mut next,
max,
..
} = *self
{
if bytes_read >= *next {
*next = cmp::min(incr_power_of_two(*next), max);
*decrease_now = false;
} else {
let decr_to = prev_power_of_two(*next);
if bytes_read < decr_to {
if *decrease_now {
*next = cmp::max(decr_to, INIT_BUFFER_SIZE);
*decrease_now = false;
} else {
// Decreasing is a two "record" process.
*decrease_now = true;
}
} else {
// A read within the current range should cancel
// a potential decrease, since we just saw proof
// that we still need this size.
match *self {
ReadStrategy::Adaptive {
ref mut decrease_now,
ref mut next,
max,
..
} => {
if bytes_read >= *next {
*next = cmp::min(incr_power_of_two(*next), max);
*decrease_now = false;
} else {
let decr_to = prev_power_of_two(*next);
if bytes_read < decr_to {
if *decrease_now {
*next = cmp::max(decr_to, INIT_BUFFER_SIZE);
*decrease_now = false;
} else {
// Decreasing is a two "record" process.
*decrease_now = true;
}
} else {
// A read within the current range should cancel
// a potential decrease, since we just saw proof
// that we still need this size.
*decrease_now = false;
}
}
}
},
#[cfg(feature = "client")]
ReadStrategy::Exact(_) => (),
}
}
}

5
src/proto/h1/mod.rs
View File

@@ -19,7 +19,10 @@ mod io;
mod role;
pub(crate) type ServerTransaction = role::Server;
pub(crate) type ClientTransaction = role::Client;
cfg_client! {
pub(crate) type ClientTransaction = role::Client;
}
pub(crate) trait Http1Transaction {
type Incoming;

2
src/proto/h2/client.rs
View File

@@ -10,7 +10,7 @@ use tokio::io::{AsyncRead, AsyncWrite};
use super::{decode_content_length, ping, PipeToSendStream, SendBuf};
use crate::body::HttpBody;
use crate::common::{task, Exec, Future, Never, Pin, Poll};
use crate::common::{task, exec::Exec, Future, Never, Pin, Poll};
use crate::headers;
use crate::proto::Dispatched;
use crate::{Body, Request, Response};

7
src/proto/h2/mod.rs
View File

@@ -13,11 +13,14 @@ use crate::body::{DecodedLength, HttpBody};
use crate::common::{task, Future, Pin, Poll};
use crate::headers::content_length_parse_all;
pub(crate) mod client;
pub(crate) mod ping;
pub(crate) mod server;
pub(crate) use self::client::ClientTask;
cfg_client! {
pub(crate) mod client;
pub(crate) use self::client::ClientTask;
}
pub(crate) use self::server::Server;
/// Default initial stream window size defined in HTTP2 spec.

1
src/proto/h2/ping.rs
View File

@@ -236,6 +236,7 @@ impl Recorder {
/// If the incoming stream is already closed, convert self into
/// a disabled reporter.
#[cfg(feature = "client")]
pub(super) fn for_stream(self, stream: &h2::RecvStream) -> Self {
if stream.is_end_stream() {
disabled()

6
src/proto/mod.rs
View File

@@ -3,7 +3,10 @@
cfg_http1! {
pub(crate) mod h1;
pub(crate) use self::h1::{dispatch, Conn, ServerTransaction};
pub(crate) use self::h1::{Conn, ServerTransaction};
#[cfg(feature = "client")]
pub(crate) use self::h1::dispatch;
}
cfg_http2! {
@@ -31,6 +34,7 @@ pub struct RequestLine(pub http::Method, pub http::Uri);
/// An incoming response message.
#[cfg(feature = "http1")]
#[cfg(feature = "client")]
pub type ResponseHead = MessageHead<http::StatusCode>;
#[derive(Debug)]