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

refactor(client): clean up client config fields

Browse Source
This commit is contained in:
Sean McArthur
2018-12-05 12:17:03 -08:00
parent 155dc41e70
commit 4b3747110a
3 changed files with 62 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/client/conn.rs
View File

@@ -71,7 +71,7 @@ where
/// After setting options, the builder is used to create a `Handshake` future. /// After setting options, the builder is used to create a `Handshake` future.
#[derive(Clone, Debug)] #[derive(Clone, Debug)]
pub struct Builder { pub struct Builder {
exec: Exec, pub(super) exec: Exec,
h1_writev: bool, h1_writev: bool,
h1_title_case_headers: bool, h1_title_case_headers: bool,
h1_read_buf_exact_size: Option<usize>, h1_read_buf_exact_size: Option<usize>,

126
src/client/mod.rs
View File

@@ -90,7 +90,7 @@ use http::header::{HeaderValue, HOST};
use http::uri::Scheme; use http::uri::Scheme;
use body::{Body, Payload}; use body::{Body, Payload};
use common::{Exec, lazy as hyper_lazy, Lazy}; use common::{lazy as hyper_lazy, Lazy};
use self::connect::{Alpn, Connect, Connected, Destination}; use self::connect::{Alpn, Connect, Connected, Destination};
use self::pool::{Key as PoolKey, Pool, Poolable, Pooled, Reservation}; use self::pool::{Key as PoolKey, Pool, Poolable, Pooled, Reservation};
@@ -105,12 +105,14 @@ mod tests;
/// A Client to make outgoing HTTP requests. /// A Client to make outgoing HTTP requests.
pub struct Client<C, B = Body> { pub struct Client<C, B = Body> {
config: Config,
conn_builder: conn::Builder,
connector: Arc<C>, connector: Arc<C>,
executor: Exec,
h1_writev: bool,
h1_title_case_headers: bool,
pool: Pool<PoolClient<B>>, pool: Pool<PoolClient<B>>,
h1_read_buf_exact_size: Option<usize>, }
#[derive(Clone, Copy, Debug)]
struct Config {
retry_canceled_requests: bool, retry_canceled_requests: bool,
set_host: bool, set_host: bool,
ver: Ver, ver: Ver,
@@ -200,7 +202,7 @@ where C: Connect + Sync + 'static,
debug!("CONNECT is not allowed for HTTP/1.0"); debug!("CONNECT is not allowed for HTTP/1.0");
return ResponseFuture::new(Box::new(future::err(::Error::new_user_unsupported_request_method()))); return ResponseFuture::new(Box::new(future::err(::Error::new_user_unsupported_request_method())));
}, },
other => if self.ver != Ver::Http2 { other => if self.config.ver != Ver::Http2 {
error!("Request has unsupported version \"{:?}\"", other); error!("Request has unsupported version \"{:?}\"", other);
return ResponseFuture::new(Box::new(future::err(::Error::new_user_unsupported_version()))); return ResponseFuture::new(Box::new(future::err(::Error::new_user_unsupported_version())));
} }
@@ -250,7 +252,7 @@ where C: Connect + Sync + 'static,
mut req, mut req,
reason, reason,
}) => { }) => {
if !client.retry_canceled_requests || !connection_reused { if !client.config.retry_canceled_requests || !connection_reused {
// if client disabled, don't retry // if client disabled, don't retry
// a fresh connection means we definitely can't retry // a fresh connection means we definitely can't retry
return Err(reason); return Err(reason);
@@ -267,8 +269,8 @@ where C: Connect + Sync + 'static,
fn send_request(&self, mut req: Request<B>, pool_key: PoolKey) -> impl Future<Item=Response<Body>, Error=ClientError<B>> { fn send_request(&self, mut req: Request<B>, pool_key: PoolKey) -> impl Future<Item=Response<Body>, Error=ClientError<B>> {
let conn = self.connection_for(req.uri().clone(), pool_key); let conn = self.connection_for(req.uri().clone(), pool_key);
let set_host = self.set_host; let set_host = self.config.set_host;
let executor = self.executor.clone(); let executor = self.conn_builder.exec.clone();
conn.and_then(move |mut pooled| { conn.and_then(move |mut pooled| {
if pooled.is_http1() { if pooled.is_http1() {
if set_host { if set_host {
@@ -392,7 +394,7 @@ where C: Connect + Sync + 'static,
let checkout = self.pool.checkout(pool_key.clone()); let checkout = self.pool.checkout(pool_key.clone());
let connect = self.connect_to(uri, pool_key); let connect = self.connect_to(uri, pool_key);
let executor = self.executor.clone(); let executor = self.conn_builder.exec.clone();
checkout checkout
// The order of the `select` is depended on below... // The order of the `select` is depended on below...
.select2(connect) .select2(connect)
@@ -458,13 +460,11 @@ where C: Connect + Sync + 'static,
fn connect_to(&self, uri: Uri, pool_key: PoolKey) fn connect_to(&self, uri: Uri, pool_key: PoolKey)
-> impl Lazy<Item=Pooled<PoolClient<B>>, Error=::Error> -> impl Lazy<Item=Pooled<PoolClient<B>>, Error=::Error>
{ {
let executor = self.executor.clone(); let executor = self.conn_builder.exec.clone();
let pool = self.pool.clone(); let pool = self.pool.clone();
let h1_writev = self.h1_writev; let mut conn_builder = self.conn_builder.clone();
let h1_title_case_headers = self.h1_title_case_headers; let ver = self.config.ver;
let h1_read_buf_exact_size = self.h1_read_buf_exact_size; let is_ver_h2 = ver == Ver::Http2;
let ver = self.ver;
let is_ver_h2 = self.ver == Ver::Http2;
let connector = self.connector.clone(); let connector = self.connector.clone();
let dst = Destination { let dst = Destination {
uri, uri,
@@ -505,11 +505,7 @@ where C: Connect + Sync + 'static,
connecting connecting
}; };
let is_h2 = is_ver_h2 || connected.alpn == Alpn::H2; let is_h2 = is_ver_h2 || connected.alpn == Alpn::H2;
Either::A(conn::Builder::new() Either::A(conn_builder
.exec(executor.clone())
.h1_writev(h1_writev)
.h1_title_case_headers(h1_title_case_headers)
.h1_read_buf_exact_size(h1_read_buf_exact_size)
.http2_only(is_h2) .http2_only(is_h2)
.handshake(io) .handshake(io)
.and_then(move |(tx, conn)| { .and_then(move |(tx, conn)| {
@@ -546,15 +542,10 @@ where C: Connect + Sync + 'static,
impl<C, B> Clone for Client<C, B> { impl<C, B> Clone for Client<C, B> {
fn clone(&self) -> Client<C, B> { fn clone(&self) -> Client<C, B> {
Client { Client {
config: self.config.clone(),
conn_builder: self.conn_builder.clone(),
connector: self.connector.clone(), connector: self.connector.clone(),
executor: self.executor.clone(),
h1_writev: self.h1_writev,
h1_read_buf_exact_size: self.h1_read_buf_exact_size,
h1_title_case_headers: self.h1_title_case_headers,
pool: self.pool.clone(), pool: self.pool.clone(),
retry_canceled_requests: self.retry_canceled_requests,
set_host: self.set_host,
ver: self.ver,
} }
} }
} }
@@ -790,32 +781,25 @@ fn set_scheme(uri: &mut Uri, scheme: Scheme) {
/// Builder for a Client /// Builder for a Client
#[derive(Clone)] #[derive(Clone)]
pub struct Builder { pub struct Builder {
//connect_timeout: Duration, client_config: Config,
exec: Exec, conn_builder: conn::Builder,
keep_alive: bool, pool_config: pool::Config,
keep_alive_timeout: Option<Duration>,
h1_writev: bool,
h1_title_case_headers: bool,
h1_read_buf_exact_size: Option<usize>,
max_idle_per_host: usize,
retry_canceled_requests: bool,
set_host: bool,
ver: Ver,
} }
impl Default for Builder { impl Default for Builder {
fn default() -> Self { fn default() -> Self {
Self { Self {
exec: Exec::Default, client_config: Config {
keep_alive: true,
keep_alive_timeout: Some(Duration::from_secs(90)),
h1_writev: true,
h1_title_case_headers: false,
h1_read_buf_exact_size: None,
max_idle_per_host: ::std::usize::MAX,
retry_canceled_requests: true, retry_canceled_requests: true,
set_host: true, set_host: true,
ver: Ver::Http1, ver: Ver::Http1,
},
conn_builder: conn::Builder::new(),
pool_config: pool::Config {
enabled: true,
keep_alive_timeout: Some(Duration::from_secs(90)),
max_idle_per_host: ::std::usize::MAX,
},
} }
} }
} }
@@ -826,7 +810,7 @@ impl Builder {
/// Default is enabled. /// Default is enabled.
#[inline] #[inline]
pub fn keep_alive(&mut self, val: bool) -> &mut Self { pub fn keep_alive(&mut self, val: bool) -> &mut Self {
self.keep_alive = val; self.pool_config.enabled = val;
self self
} }
@@ -840,7 +824,7 @@ impl Builder {
where where
D: Into<Option<Duration>>, D: Into<Option<Duration>>,
{ {
self.keep_alive_timeout = val.into(); self.pool_config.keep_alive_timeout = val.into();
self self
} }
@@ -854,7 +838,7 @@ impl Builder {
/// Default is `true`. /// Default is `true`.
#[inline] #[inline]
pub fn http1_writev(&mut self, val: bool) -> &mut Self { pub fn http1_writev(&mut self, val: bool) -> &mut Self {
self.h1_writev = val; self.conn_builder.h1_writev(val);
self self
} }
@@ -863,7 +847,7 @@ impl Builder {
/// Default is an adaptive read buffer. /// Default is an adaptive read buffer.
#[inline] #[inline]
pub fn http1_read_buf_exact_size(&mut self, sz: usize) -> &mut Self { pub fn http1_read_buf_exact_size(&mut self, sz: usize) -> &mut Self {
self.h1_read_buf_exact_size = Some(sz); self.conn_builder.h1_read_buf_exact_size(Some(sz));
self self
} }
@@ -874,7 +858,7 @@ impl Builder {
/// ///
/// Default is false. /// Default is false.
pub fn http1_title_case_headers(&mut self, val: bool) -> &mut Self { pub fn http1_title_case_headers(&mut self, val: bool) -> &mut Self {
self.h1_title_case_headers = val; self.conn_builder.h1_title_case_headers(val);
self self
} }
@@ -884,7 +868,7 @@ impl Builder {
/// ///
/// Default is false. /// Default is false.
pub fn http2_only(&mut self, val: bool) -> &mut Self { pub fn http2_only(&mut self, val: bool) -> &mut Self {
self.ver = if val { self.client_config.ver = if val {
Ver::Http2 Ver::Http2
} else { } else {
Ver::Http1 Ver::Http1
@@ -896,7 +880,7 @@ impl Builder {
/// ///
/// Default is `usize::MAX` (no limit). /// Default is `usize::MAX` (no limit).
pub fn max_idle_per_host(&mut self, max_idle: usize) -> &mut Self { pub fn max_idle_per_host(&mut self, max_idle: usize) -> &mut Self {
self.max_idle_per_host = max_idle; self.pool_config.max_idle_per_host = max_idle;
self self
} }
@@ -913,7 +897,7 @@ impl Builder {
/// Default is `true`. /// Default is `true`.
#[inline] #[inline]
pub fn retry_canceled_requests(&mut self, val: bool) -> &mut Self { pub fn retry_canceled_requests(&mut self, val: bool) -> &mut Self {
self.retry_canceled_requests = val; self.client_config.retry_canceled_requests = val;
self self
} }
@@ -925,7 +909,7 @@ impl Builder {
/// Default is `true`. /// Default is `true`.
#[inline] #[inline]
pub fn set_host(&mut self, val: bool) -> &mut Self { pub fn set_host(&mut self, val: bool) -> &mut Self {
self.set_host = val; self.client_config.set_host = val;
self self
} }
@@ -934,7 +918,7 @@ impl Builder {
where where
E: Executor<Box<Future<Item=(), Error=()> + Send>> + Send + Sync + 'static, E: Executor<Box<Future<Item=(), Error=()> + Send>> + Send + Sync + 'static,
{ {
self.exec = Exec::Executor(Arc::new(exec)); self.conn_builder.executor(exec);
self self
} }
@@ -946,8 +930,8 @@ impl Builder {
B::Data: Send, B::Data: Send,
{ {
let mut connector = HttpConnector::new(4); let mut connector = HttpConnector::new(4);
if self.keep_alive { if self.pool_config.enabled {
connector.set_keepalive(self.keep_alive_timeout); connector.set_keepalive(self.pool_config.keep_alive_timeout);
} }
self.build(connector) self.build(connector)
} }
@@ -962,20 +946,10 @@ impl Builder {
B::Data: Send, B::Data: Send,
{ {
Client { Client {
config: self.client_config,
conn_builder: self.conn_builder.clone(),
connector: Arc::new(connector), connector: Arc::new(connector),
executor: self.exec.clone(), pool: Pool::new(self.pool_config, &self.conn_builder.exec),
h1_writev: self.h1_writev,
h1_title_case_headers: self.h1_title_case_headers,
h1_read_buf_exact_size: self.h1_read_buf_exact_size,
pool: Pool::new(
pool::Enabled(self.keep_alive),
pool::IdleTimeout(self.keep_alive_timeout),
pool::MaxIdlePerHost(self.max_idle_per_host),
&self.exec,
),
retry_canceled_requests: self.retry_canceled_requests,
set_host: self.set_host,
ver: self.ver,
} }
} }
} }
@@ -983,13 +957,9 @@ impl Builder {
impl fmt::Debug for Builder { impl fmt::Debug for Builder {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.debug_struct("Builder") f.debug_struct("Builder")
.field("keep_alive", &self.keep_alive) .field("client_config", &self.client_config)
.field("keep_alive_timeout", &self.keep_alive_timeout) .field("conn_builder", &self.conn_builder)
.field("http1_read_buf_exact_size", &self.h1_read_buf_exact_size) .field("pool_config", &self.pool_config)
.field("http1_writev", &self.h1_writev)
.field("max_idle_per_host", &self.max_idle_per_host)
.field("set_host", &self.set_host)
.field("version", &self.ver)
.finish() .finish()
} }
} }

27
src/client/pool.rs
View File

@@ -85,33 +85,26 @@ struct PoolInner<T> {
// doesn't need it! // doesn't need it!
struct WeakOpt<T>(Option<Weak<T>>); struct WeakOpt<T>(Option<Weak<T>>);
// Newtypes to act as keyword arguments for `Pool::new`... #[derive(Clone, Copy, Debug)]
pub(super) struct Config {
// FIXME: allow() required due to `impl Trait` leaking types to this lint pub(super) enabled: bool,
#[allow(missing_debug_implementations)] pub(super) keep_alive_timeout: Option<Duration>,
pub(super) struct Enabled(pub(super) bool); pub(super) max_idle_per_host: usize,
}
// FIXME: allow() required due to `impl Trait` leaking types to this lint
#[allow(missing_debug_implementations)]
pub(super) struct IdleTimeout(pub(super) Option<Duration>);
// FIXME: allow() required due to `impl Trait` leaking types to this lint
#[allow(missing_debug_implementations)]
pub(super) struct MaxIdlePerHost(pub(super) usize);
impl<T> Pool<T> { impl<T> Pool<T> {
pub fn new(enabled: Enabled, timeout: IdleTimeout, max_idle: MaxIdlePerHost, __exec: &Exec) -> Pool<T> { pub fn new(config: Config, __exec: &Exec) -> Pool<T> {
let inner = if enabled.0 { let inner = if config.enabled {
Some(Arc::new(Mutex::new(PoolInner { Some(Arc::new(Mutex::new(PoolInner {
connecting: HashSet::new(), connecting: HashSet::new(),
idle: HashMap::new(), idle: HashMap::new(),
#[cfg(feature = "runtime")] #[cfg(feature = "runtime")]
idle_interval_ref: None, idle_interval_ref: None,
max_idle_per_host: max_idle.0, max_idle_per_host: config.max_idle_per_host,
waiters: HashMap::new(), waiters: HashMap::new(),
#[cfg(feature = "runtime")] #[cfg(feature = "runtime")]
exec: __exec.clone(), exec: __exec.clone(),
timeout: timeout.0, timeout: config.keep_alive_timeout,
}))) })))
} else { } else {
None None