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

feat(client): update construction of Clients

Browse Source 
- `Client::new()` no longer needs a `Handle`, and instead makes use of
  tokio's implicit default.
- Changed `Client::configure()` to `Client::builder()`.
- `Builder` is a by-ref builder, since all configuration is now
  cloneable pieces.

BREAKING CHANGE: `Client:new(&handle)` and `Client::configure()` are now
  `Client::new()` and `Client::builder()`.
This commit is contained in:
Sean McArthur
2018-04-10 17:30:10 -07:00
parent dfdca25c00
commit fe1578acf6
7 changed files with 156 additions and 266 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
examples/client.rs
View File

@@ -31,7 +31,7 @@ fn main() {
} }
tokio::run(lazy(move || { tokio::run(lazy(move || {
let client = Client::default(); let client = Client::new();
let mut req = Request::new(Body::empty()); let mut req = Request::new(Body::empty());
*req.uri_mut() = url; *req.uri_mut() = url;

11
examples/web_api.rs
View File

@@ -6,9 +6,9 @@ extern crate tokio;
use futures::{Future, Stream}; use futures::{Future, Stream};
use futures::future::lazy; use futures::future::lazy;
use tokio::reactor::Handle;
use hyper::{Body, Chunk, Client, Method, Request, Response, StatusCode}; use hyper::{Body, Chunk, Client, Method, Request, Response, StatusCode};
use hyper::client::HttpConnector;
use hyper::server::{Http, Service}; use hyper::server::{Http, Service};
#[allow(unused, deprecated)] #[allow(unused, deprecated)]
@@ -19,7 +19,7 @@ static URL: &str = "http://127.0.0.1:1337/web_api";
static INDEX: &[u8] = b"<a href=\"test.html\">test.html</a>"; static INDEX: &[u8] = b"<a href=\"test.html\">test.html</a>";
static LOWERCASE: &[u8] = b"i am a lower case string"; static LOWERCASE: &[u8] = b"i am a lower case string";
struct ResponseExamples(Handle); struct ResponseExamples(Client<HttpConnector>);
impl Service for ResponseExamples { impl Service for ResponseExamples {
type Request = Request<Body>; type Request = Request<Body>;
@@ -35,13 +35,12 @@ impl Service for ResponseExamples {
}, },
(&Method::GET, "/test.html") => { (&Method::GET, "/test.html") => {
// Run a web query against the web api below // Run a web query against the web api below
let client = Client::configure().build(&self.0);
let req = Request::builder() let req = Request::builder()
.method(Method::POST) .method(Method::POST)
.uri(URL) .uri(URL)
.body(LOWERCASE.into()) .body(LOWERCASE.into())
.unwrap(); .unwrap();
let web_res_future = client.request(req); let web_res_future = self.0.request(req);
Box::new(web_res_future.map(|web_res| { Box::new(web_res_future.map(|web_res| {
let body = Body::wrap_stream(web_res.into_body().map(|b| { let body = Body::wrap_stream(web_res.into_body().map(|b| {
@@ -79,8 +78,8 @@ fn main() {
let addr = "127.0.0.1:1337".parse().unwrap(); let addr = "127.0.0.1:1337".parse().unwrap();
tokio::run(lazy(move || { tokio::run(lazy(move || {
let handle = Handle::current(); let client = Client::new();
let serve = Http::new().serve_addr(&addr, move || Ok(ResponseExamples(handle.clone()))).unwrap(); let serve = Http::new().serve_addr(&addr, move || Ok(ResponseExamples(client.clone()))).unwrap();
println!("Listening on http://{} with 1 thread.", serve.incoming_ref().local_addr()); println!("Listening on http://{} with 1 thread.", serve.incoming_ref().local_addr());
serve.map_err(|_| ()).for_each(move |conn| { serve.map_err(|_| ()).for_each(move |conn| {

74
src/client/connect.rs
View File

@@ -135,26 +135,30 @@ impl Connected {
*/ */
} }
fn connect(addr: &SocketAddr, handle: &Handle) -> io::Result<ConnectFuture> { fn connect(addr: &SocketAddr, handle: &Option<Handle>) -> io::Result<ConnectFuture> {
let builder = match addr { if let Some(ref handle) = *handle {
&SocketAddr::V4(_) => TcpBuilder::new_v4()?, let builder = match addr {
&SocketAddr::V6(_) => TcpBuilder::new_v6()?, &SocketAddr::V4(_) => TcpBuilder::new_v4()?,
}; &SocketAddr::V6(_) => TcpBuilder::new_v6()?,
if cfg!(windows) {
// Windows requires a socket be bound before calling connect
let any: SocketAddr = match addr {
&SocketAddr::V4(_) => {
([0, 0, 0, 0], 0).into()
},
&SocketAddr::V6(_) => {
([0, 0, 0, 0, 0, 0, 0, 0], 0).into()
}
}; };
builder.bind(any)?;
}
Ok(TcpStream::connect_std(builder.to_tcp_stream()?, addr, handle)) if cfg!(windows) {
// Windows requires a socket be bound before calling connect
let any: SocketAddr = match addr {
&SocketAddr::V4(_) => {
([0, 0, 0, 0], 0).into()
},
&SocketAddr::V6(_) => {
([0, 0, 0, 0, 0, 0, 0, 0], 0).into()
}
};
builder.bind(any)?;
}
Ok(TcpStream::connect_std(builder.to_tcp_stream()?, addr, handle))
} else {
Ok(TcpStream::connect(addr))
}
} }
/// A connector for the `http` scheme. /// A connector for the `http` scheme.
@@ -164,7 +168,7 @@ fn connect(addr: &SocketAddr, handle: &Handle) -> io::Result<ConnectFuture> {
pub struct HttpConnector { pub struct HttpConnector {
executor: HttpConnectExecutor, executor: HttpConnectExecutor,
enforce_http: bool, enforce_http: bool,
handle: Handle, handle: Option<Handle>,
keep_alive_timeout: Option<Duration>, keep_alive_timeout: Option<Duration>,
} }
@@ -173,7 +177,16 @@ impl HttpConnector {
/// ///
/// Takes number of DNS worker threads. /// Takes number of DNS worker threads.
#[inline] #[inline]
pub fn new(threads: usize, handle: &Handle) -> HttpConnector { pub fn new(threads: usize) -> HttpConnector {
HttpConnector::new_with_handle_opt(threads, None)
}
/// Construct a new HttpConnector with a specific Tokio handle.
pub fn new_with_handle(threads: usize, handle: Handle) -> HttpConnector {
HttpConnector::new_with_handle_opt(threads, Some(handle))
}
fn new_with_handle_opt(threads: usize, handle: Option<Handle>) -> HttpConnector {
let pool = CpuPoolBuilder::new() let pool = CpuPoolBuilder::new()
.name_prefix("hyper-dns") .name_prefix("hyper-dns")
.pool_size(threads) .pool_size(threads)
@@ -184,14 +197,13 @@ impl HttpConnector {
/// Construct a new HttpConnector. /// Construct a new HttpConnector.
/// ///
/// Takes an executor to run blocking tasks on. /// Takes an executor to run blocking tasks on.
#[inline] pub fn new_with_executor<E: 'static>(executor: E, handle: Option<Handle>) -> HttpConnector
pub fn new_with_executor<E: 'static>(executor: E, handle: &Handle) -> HttpConnector
where E: Executor<HttpConnectorBlockingTask> + Send + Sync where E: Executor<HttpConnectorBlockingTask> + Send + Sync
{ {
HttpConnector { HttpConnector {
executor: HttpConnectExecutor(Arc::new(executor)), executor: HttpConnectExecutor(Arc::new(executor)),
enforce_http: true, enforce_http: true,
handle: handle.clone(), handle,
keep_alive_timeout: None, keep_alive_timeout: None,
} }
} }
@@ -257,7 +269,7 @@ impl Connect for HttpConnector {
} }
#[inline] #[inline]
fn invalid_url(err: InvalidUrl, handle: &Handle) -> HttpConnecting { fn invalid_url(err: InvalidUrl, handle: &Option<Handle>) -> HttpConnecting {
HttpConnecting { HttpConnecting {
state: State::Error(Some(io::Error::new(io::ErrorKind::InvalidInput, err))), state: State::Error(Some(io::Error::new(io::ErrorKind::InvalidInput, err))),
handle: handle.clone(), handle: handle.clone(),
@@ -292,7 +304,7 @@ impl StdError for InvalidUrl {
#[must_use = "futures do nothing unless polled"] #[must_use = "futures do nothing unless polled"]
pub struct HttpConnecting { pub struct HttpConnecting {
state: State, state: State,
handle: Handle, handle: Option<Handle>,
keep_alive_timeout: Option<Duration>, keep_alive_timeout: Option<Duration>,
} }
@@ -365,7 +377,7 @@ struct ConnectingTcp {
impl ConnectingTcp { impl ConnectingTcp {
// not a Future, since passing a &Handle to poll // not a Future, since passing a &Handle to poll
fn poll(&mut self, handle: &Handle) -> Poll<TcpStream, io::Error> { fn poll(&mut self, handle: &Option<Handle>) -> Poll<TcpStream, io::Error> {
let mut err = None; let mut err = None;
loop { loop {
if let Some(ref mut current) = self.current { if let Some(ref mut current) = self.current {
@@ -431,29 +443,26 @@ mod tests {
#![allow(deprecated)] #![allow(deprecated)]
use std::io; use std::io;
use futures::Future; use futures::Future;
use tokio::runtime::Runtime;
use super::{Connect, Destination, HttpConnector}; use super::{Connect, Destination, HttpConnector};
#[test] #[test]
fn test_errors_missing_authority() { fn test_errors_missing_authority() {
let runtime = Runtime::new().unwrap();
let uri = "/foo/bar?baz".parse().unwrap(); let uri = "/foo/bar?baz".parse().unwrap();
let dst = Destination { let dst = Destination {
uri, uri,
}; };
let connector = HttpConnector::new(1, runtime.handle()); let connector = HttpConnector::new(1);
assert_eq!(connector.connect(dst).wait().unwrap_err().kind(), io::ErrorKind::InvalidInput); assert_eq!(connector.connect(dst).wait().unwrap_err().kind(), io::ErrorKind::InvalidInput);
} }
#[test] #[test]
fn test_errors_enforce_http() { fn test_errors_enforce_http() {
let runtime = Runtime::new().unwrap();
let uri = "https://example.domain/foo/bar?baz".parse().unwrap(); let uri = "https://example.domain/foo/bar?baz".parse().unwrap();
let dst = Destination { let dst = Destination {
uri, uri,
}; };
let connector = HttpConnector::new(1, runtime.handle()); let connector = HttpConnector::new(1);
assert_eq!(connector.connect(dst).wait().unwrap_err().kind(), io::ErrorKind::InvalidInput); assert_eq!(connector.connect(dst).wait().unwrap_err().kind(), io::ErrorKind::InvalidInput);
} }
@@ -461,12 +470,11 @@ mod tests {
#[test] #[test]
fn test_errors_missing_scheme() { fn test_errors_missing_scheme() {
let runtime = Runtime::new().unwrap();
let uri = "example.domain".parse().unwrap(); let uri = "example.domain".parse().unwrap();
let dst = Destination { let dst = Destination {
uri, uri,
}; };
let connector = HttpConnector::new(1, runtime.handle()); let connector = HttpConnector::new(1);
assert_eq!(connector.connect(dst).wait().unwrap_err().kind(), io::ErrorKind::InvalidInput); assert_eq!(connector.connect(dst).wait().unwrap_err().kind(), io::ErrorKind::InvalidInput);
} }

234
src/client/mod.rs
View File

@@ -2,7 +2,6 @@
use std::fmt; use std::fmt;
use std::io; use std::io;
use std::marker::PhantomData;
use std::sync::Arc; use std::sync::Arc;
use std::time::Duration; use std::time::Duration;
@@ -12,7 +11,6 @@ use futures::sync::oneshot;
use http::{Method, Request, Response, Uri, Version}; use http::{Method, Request, Response, Uri, Version};
use http::header::{Entry, HeaderValue, HOST}; use http::header::{Entry, HeaderValue, HOST};
use http::uri::Scheme; use http::uri::Scheme;
use tokio::reactor::Handle;
use tokio_executor::spawn; use tokio_executor::spawn;
pub use tokio_service::Service; pub use tokio_service::Service;
@@ -21,7 +19,6 @@ use self::pool::Pool;
pub use self::connect::{Connect, HttpConnector}; pub use self::connect::{Connect, HttpConnector};
use self::background::{bg, Background};
use self::connect::Destination; use self::connect::Destination;
pub mod conn; pub mod conn;
@@ -45,14 +42,14 @@ pub struct Client<C, B = Body> {
impl Client<HttpConnector, Body> { impl Client<HttpConnector, Body> {
/// Create a new Client with the default config. /// Create a new Client with the default config.
#[inline] #[inline]
pub fn new(handle: &Handle) -> Client<HttpConnector, Body> { pub fn new() -> Client<HttpConnector, Body> {
Config::default().build(handle) Builder::default().build_http()
} }
} }
impl Default for Client<HttpConnector, Body> { impl Default for Client<HttpConnector, Body> {
fn default() -> Client<HttpConnector, Body> { fn default() -> Client<HttpConnector, Body> {
Client::new(&Handle::current()) Client::new()
} }
} }
@@ -61,36 +58,18 @@ impl Client<HttpConnector, Body> {
/// ///
/// # Example /// # Example
/// ///
/// ```no_run /// ```
/// # extern crate hyper; /// use hyper::Client;
/// # extern crate tokio;
/// ///
/// # fn main() { /// let client = Client::builder()
/// # let runtime = tokio::runtime::Runtime::new().unwrap();
/// # let handle = runtime.handle();
/// let client = hyper::Client::configure()
/// .keep_alive(true) /// .keep_alive(true)
/// .build(&handle); /// .build_http();
/// # drop(client); /// # let infer: Client<_, hyper::Body> = client;
/// # } /// # drop(infer);
/// ``` /// ```
#[inline] #[inline]
pub fn configure() -> Config<UseDefaultConnector, Body> { pub fn builder() -> Builder {
Config::default() Builder::default()
}
}
impl<C, B> Client<C, B> {
#[inline]
fn configured(config: Config<C, B>, exec: Exec) -> Client<C, B> {
Client {
connector: Arc::new(config.connector),
executor: exec,
h1_writev: config.h1_writev,
pool: Pool::new(config.keep_alive, config.keep_alive_timeout),
retry_canceled_requests: config.retry_canceled_requests,
set_host: config.set_host,
}
} }
} }
@@ -425,11 +404,11 @@ fn set_relative_uri(uri: &mut Uri, is_proxied: bool) {
*uri = path; *uri = path;
} }
/// Configuration for a Client /// Builder for a Client
pub struct Config<C, B> { #[derive(Clone)]
_body_type: PhantomData<B>, pub struct Builder {
//connect_timeout: Duration, //connect_timeout: Duration,
connector: C, exec: Exec,
keep_alive: bool, keep_alive: bool,
keep_alive_timeout: Option<Duration>, keep_alive_timeout: Option<Duration>,
h1_writev: bool, h1_writev: bool,
@@ -439,15 +418,10 @@ pub struct Config<C, B> {
set_host: bool, set_host: bool,
} }
/// Phantom type used to signal that `Config` should create a `HttpConnector`. impl Default for Builder {
#[derive(Debug, Clone, Copy)] fn default() -> Self {
pub struct UseDefaultConnector(()); Self {
exec: Exec::Default,
impl Default for Config<UseDefaultConnector, Body> {
fn default() -> Config<UseDefaultConnector, Body> {
Config {
_body_type: PhantomData::<Body>,
connector: UseDefaultConnector(()),
keep_alive: true, keep_alive: true,
keep_alive_timeout: Some(Duration::from_secs(90)), keep_alive_timeout: Some(Duration::from_secs(90)),
h1_writev: true, h1_writev: true,
@@ -458,50 +432,12 @@ impl Default for Config<UseDefaultConnector, Body> {
} }
} }
impl<C, B> Config<C, B> { impl Builder {
/// Set the body stream to be used by the `Client`.
///
/// # Example
///
/// ```rust
/// # use hyper::client::Config;
/// let cfg = Config::default()
/// .body::<hyper::Body>();
/// # drop(cfg);
#[inline]
pub fn body<BB>(self) -> Config<C, BB> {
Config {
_body_type: PhantomData::<BB>,
connector: self.connector,
keep_alive: self.keep_alive,
keep_alive_timeout: self.keep_alive_timeout,
h1_writev: self.h1_writev,
max_idle: self.max_idle,
retry_canceled_requests: self.retry_canceled_requests,
set_host: self.set_host,
}
}
/// Set the `Connect` type to be used.
#[inline]
pub fn connector<CC>(self, val: CC) -> Config<CC, B> {
Config {
_body_type: self._body_type,
connector: val,
keep_alive: self.keep_alive,
keep_alive_timeout: self.keep_alive_timeout,
h1_writev: self.h1_writev,
max_idle: self.max_idle,
retry_canceled_requests: self.retry_canceled_requests,
set_host: self.set_host,
}
}
/// Enable or disable keep-alive mechanics. /// Enable or disable keep-alive mechanics.
/// ///
/// Default is enabled. /// Default is enabled.
#[inline] #[inline]
pub fn keep_alive(mut self, val: bool) -> Config<C, B> { pub fn keep_alive(&mut self, val: bool) -> &mut Self {
self.keep_alive = val; self.keep_alive = val;
self self
} }
@@ -512,7 +448,7 @@ impl<C, B> Config<C, B> {
/// ///
/// Default is 90 seconds. /// Default is 90 seconds.
#[inline] #[inline]
pub fn keep_alive_timeout(mut self, val: Option<Duration>) -> Config<C, B> { pub fn keep_alive_timeout(&mut self, val: Option<Duration>) -> &mut Self {
self.keep_alive_timeout = val; self.keep_alive_timeout = val;
self self
} }
@@ -526,7 +462,7 @@ impl<C, B> Config<C, B> {
/// ///
/// Default is `true`. /// Default is `true`.
#[inline] #[inline]
pub fn http1_writev(mut self, val: bool) -> Config<C, B> { pub fn http1_writev(&mut self, val: bool) -> &mut Self {
self.h1_writev = val; self.h1_writev = val;
self self
} }
@@ -543,7 +479,7 @@ impl<C, B> Config<C, B> {
/// ///
/// Default is `true`. /// Default is `true`.
#[inline] #[inline]
pub fn retry_canceled_requests(mut self, val: bool) -> Config<C, B> { pub fn retry_canceled_requests(&mut self, val: bool) -> &mut Self {
self.retry_canceled_requests = val; self.retry_canceled_requests = val;
self self
} }
@@ -555,71 +491,56 @@ impl<C, B> Config<C, B> {
/// ///
/// Default is `true`. /// Default is `true`.
#[inline] #[inline]
pub fn set_host(mut self, val: bool) -> Config<C, B> { pub fn set_host(&mut self, val: bool) -> &mut Self {
self.set_host = val; self.set_host = val;
self self
} }
}
impl<C, B> Config<C, B> /// Provide an executor to execute background `Connection` tasks.
where C: Connect, pub fn executor<E>(&mut self, exec: E) -> &mut Self
C::Transport: 'static,
C::Future: 'static,
B: Payload + Send,
B::Data: Send,
{
/// Construct the Client with this configuration.
#[inline]
pub fn build(self) -> Client<C, B> {
Client::configured(self, Exec::Default)
}
/// Construct a Client with this configuration and an executor.
///
/// The executor will be used to spawn "background" connection tasks
/// to drive requests and responses.
pub fn executor<E>(self, executor: E) -> Client<C, B>
where where
E: Executor<Background> + Send + Sync + 'static, E: Executor<Box<Future<Item=(), Error=()> + Send>> + Send + Sync + 'static,
{ {
Client::configured(self, Exec::new(executor)) self.exec = Exec::Executor(Arc::new(exec));
self
} }
}
impl<B> Config<UseDefaultConnector, B> /// Builder a client with this configuration and the default `HttpConnector`.
where pub fn build_http<B>(&self) -> Client<HttpConnector, B>
B: Payload + Send, where
B::Data: Send, B: Payload + Send,
{ B::Data: Send,
/// Construct the Client with this configuration. {
#[inline] let mut connector = HttpConnector::new(4);
pub fn build(self, handle: &Handle) -> Client<HttpConnector, B> {
let mut connector = HttpConnector::new(4, handle);
if self.keep_alive { if self.keep_alive {
connector.set_keepalive(self.keep_alive_timeout); connector.set_keepalive(self.keep_alive_timeout);
} }
self.connector(connector).build() self.build(connector)
} }
/// Construct a Client with this configuration and an executor. /// Combine the configuration of this builder with a connector to create a `Client`.
/// pub fn build<C, B>(&self, connector: C) -> Client<C, B>
/// The executor will be used to spawn "background" connection tasks
/// to drive requests and responses.
pub fn build_with_executor<E>(self, handle: &Handle, executor: E) -> Client<HttpConnector, B>
where where
E: Executor<Background> + Send + Sync + 'static, C: Connect,
C::Transport: 'static,
C::Future: 'static,
B: Payload + Send,
B::Data: Send,
{ {
let mut connector = HttpConnector::new(4, handle); Client {
if self.keep_alive { connector: Arc::new(connector),
connector.set_keepalive(self.keep_alive_timeout); executor: self.exec.clone(),
h1_writev: self.h1_writev,
pool: Pool::new(self.keep_alive, self.keep_alive_timeout),
retry_canceled_requests: self.retry_canceled_requests,
set_host: self.set_host,
} }
self.connector(connector).executor(executor)
} }
} }
impl<C, B> fmt::Debug for Config<C, B> { 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("Config") f.debug_struct("Builder")
.field("keep_alive", &self.keep_alive) .field("keep_alive", &self.keep_alive)
.field("keep_alive_timeout", &self.keep_alive_timeout) .field("keep_alive_timeout", &self.keep_alive_timeout)
.field("http1_writev", &self.h1_writev) .field("http1_writev", &self.h1_writev)
@@ -629,29 +550,16 @@ impl<C, B> fmt::Debug for Config<C, B> {
} }
} }
impl<C: Clone, B> Clone for Config<C, B> {
fn clone(&self) -> Config<C, B> {
Config {
connector: self.connector.clone(),
.. *self
}
}
}
// ===== impl Exec ===== // ===== impl Exec =====
#[derive(Clone)] #[derive(Clone)]
enum Exec { enum Exec {
Default, Default,
Executor(Arc<Executor<Background> + Send + Sync>), Executor(Arc<Executor<Box<Future<Item=(), Error=()> + Send>> + Send + Sync>),
} }
impl Exec { impl Exec {
pub(crate) fn new<E: Executor<Background> + Send + Sync + 'static>(executor: E) -> Exec {
Exec::Executor(Arc::new(executor))
}
fn execute<F>(&self, fut: F) fn execute<F>(&self, fut: F)
where where
F: Future<Item=(), Error=()> + Send + 'static, F: Future<Item=(), Error=()> + Send + 'static,
@@ -659,7 +567,7 @@ impl Exec {
match *self { match *self {
Exec::Default => spawn(fut), Exec::Default => spawn(fut),
Exec::Executor(ref e) => { Exec::Executor(ref e) => {
let _ = e.execute(bg(Box::new(fut))) let _ = e.execute(Box::new(fut))
.map_err(|err| { .map_err(|err| {
panic!("executor error: {:?}", err.kind()); panic!("executor error: {:?}", err.kind());
}); });
@@ -668,33 +576,3 @@ impl Exec {
} }
} }
// ===== impl Background =====
// The types inside this module are not exported out of the crate,
// so they are in essence un-nameable.
mod background {
use futures::{Future, Poll};
// This is basically `impl Future`, since the type is un-nameable,
// and only implementeds `Future`.
#[allow(missing_debug_implementations)]
pub struct Background {
inner: Box<Future<Item=(), Error=()> + Send>,
}
pub fn bg(fut: Box<Future<Item=(), Error=()> + Send>) -> Background {
Background {
inner: fut,
}
}
impl Future for Background {
type Item = ();
type Error = ();
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
self.inner.poll()
}
}
}

3
src/client/pool.rs
View File

@@ -503,11 +503,12 @@ mod tests {
#[test] #[test]
fn test_pool_timer_removes_expired() { fn test_pool_timer_removes_expired() {
use std::sync::Arc;
let runtime = ::tokio::runtime::Runtime::new().unwrap(); let runtime = ::tokio::runtime::Runtime::new().unwrap();
let pool = Pool::new(true, Some(Duration::from_millis(100))); let pool = Pool::new(true, Some(Duration::from_millis(100)));
let executor = runtime.executor(); let executor = runtime.executor();
pool.spawn_expired_interval(&Exec::new(executor)); pool.spawn_expired_interval(&Exec::Executor(Arc::new(executor)));
let key = Arc::new("foo".to_string()); let key = Arc::new("foo".to_string());
pool.pooled(key.clone(), 41); pool.pooled(key.clone(), 41);

12
src/client/tests.rs
View File

@@ -20,9 +20,9 @@ fn retryable_request() {
let sock1 = connector.mock("http://mock.local"); let sock1 = connector.mock("http://mock.local");
let sock2 = connector.mock("http://mock.local"); let sock2 = connector.mock("http://mock.local");
let client = Client::configure() let client = Client::builder()
.connector(connector) .executor(executor.sender().clone())
.executor(executor.sender().clone()); .build::<_, ::Body>(connector);
{ {
@@ -66,9 +66,9 @@ fn conn_reset_after_write() {
let sock1 = connector.mock("http://mock.local"); let sock1 = connector.mock("http://mock.local");
let client = Client::configure() let client = Client::builder()
.connector(connector) .executor(executor.sender().clone())
.executor(executor.sender().clone()); .build::<_, ::Body>(connector);
{ {
let req = Request::builder() let req = Request::builder()

86
tests/client.rs
View File

@@ -186,11 +186,15 @@ macro_rules! test {
let addr = server.local_addr().expect("local_addr"); let addr = server.local_addr().expect("local_addr");
let runtime = $runtime; let runtime = $runtime;
let mut config = Client::configure(); let mut config = Client::builder();
if !$set_host { if !$set_host {
config = config.set_host(false); config.set_host(false);
} }
let client = config.build_with_executor(&runtime.reactor(), runtime.executor()); let connector = ::hyper::client::HttpConnector::new_with_handle(1, runtime.reactor().clone());
let client = Client::builder()
.set_host($set_host)
.executor(runtime.executor())
.build(connector);
let body = if let Some(body) = $request_body { let body = if let Some(body) = $request_body {
let body: &'static str = body; let body: &'static str = body;
@@ -657,9 +661,9 @@ mod dispatch_impl {
let addr = server.local_addr().unwrap(); let addr = server.local_addr().unwrap();
let runtime = Runtime::new().unwrap(); let runtime = Runtime::new().unwrap();
let (closes_tx, closes) = mpsc::channel(10); let (closes_tx, closes) = mpsc::channel(10);
let client = Client::configure() let client = Client::builder()
.connector(DebugConnector::with_http_and_closes(HttpConnector::new(1, &runtime.reactor()), closes_tx)) .executor(runtime.executor())
.executor(runtime.executor()); .build(DebugConnector::with_http_and_closes(HttpConnector::new_with_handle(1, runtime.reactor().clone()), closes_tx));
let (tx1, rx1) = oneshot::channel(); let (tx1, rx1) = oneshot::channel();
@@ -716,9 +720,9 @@ mod dispatch_impl {
}); });
let res = { let res = {
let client = Client::configure() let client = Client::builder()
.connector(DebugConnector::with_http_and_closes(HttpConnector::new(1, &handle), closes_tx)) .executor(runtime.executor())
.executor(runtime.executor()); .build(DebugConnector::with_http_and_closes(HttpConnector::new_with_handle(1, handle.clone()), closes_tx));
let req = Request::builder() let req = Request::builder()
.uri(&*format!("http://{}/a", addr)) .uri(&*format!("http://{}/a", addr))
@@ -769,9 +773,9 @@ mod dispatch_impl {
let _ = client_drop_rx.wait(); let _ = client_drop_rx.wait();
}); });
let client = Client::configure() let client = Client::builder()
.connector(DebugConnector::with_http_and_closes(HttpConnector::new(1, &handle), closes_tx)) .executor(runtime.executor())
.executor(runtime.executor()); .build(DebugConnector::with_http_and_closes(HttpConnector::new_with_handle(1, handle.clone()), closes_tx));
let req = Request::builder() let req = Request::builder()
.uri(&*format!("http://{}/a", addr)) .uri(&*format!("http://{}/a", addr))
@@ -833,9 +837,9 @@ mod dispatch_impl {
}); });
let res = { let res = {
let client = Client::configure() let client = Client::builder()
.connector(DebugConnector::with_http_and_closes(HttpConnector::new(1, &handle), closes_tx)) .executor(runtime.executor())
.executor(runtime.executor()); .build(DebugConnector::with_http_and_closes(HttpConnector::new_with_handle(1, handle.clone()), closes_tx));
let req = Request::builder() let req = Request::builder()
.uri(&*format!("http://{}/a", addr)) .uri(&*format!("http://{}/a", addr))
@@ -885,9 +889,9 @@ mod dispatch_impl {
}); });
let res = { let res = {
let client = Client::configure() let client = Client::builder()
.connector(DebugConnector::with_http_and_closes(HttpConnector::new(1, &handle), closes_tx)) .executor(runtime.executor())
.executor(runtime.executor()); .build(DebugConnector::with_http_and_closes(HttpConnector::new_with_handle(1, handle.clone()), closes_tx));
let req = Request::builder() let req = Request::builder()
.uri(&*format!("http://{}/a", addr)) .uri(&*format!("http://{}/a", addr))
@@ -935,10 +939,10 @@ mod dispatch_impl {
let _ = rx2.wait(); let _ = rx2.wait();
}); });
let client = Client::configure() let client = Client::builder()
.connector(DebugConnector::with_http_and_closes(HttpConnector::new(1, &handle), closes_tx))
.keep_alive(false) .keep_alive(false)
.executor(runtime.executor()); .executor(runtime.executor())
.build(DebugConnector::with_http_and_closes(HttpConnector::new_with_handle(1, handle.clone()), closes_tx));
let req = Request::builder() let req = Request::builder()
.uri(&*format!("http://{}/a", addr)) .uri(&*format!("http://{}/a", addr))
@@ -984,9 +988,9 @@ mod dispatch_impl {
let _ = tx1.send(()); let _ = tx1.send(());
}); });
let client = Client::configure() let client = Client::builder()
.connector(DebugConnector::with_http_and_closes(HttpConnector::new(1, &handle), closes_tx)) .executor(runtime.executor())
.executor(runtime.executor()); .build(DebugConnector::with_http_and_closes(HttpConnector::new_with_handle(1, handle.clone()), closes_tx));
let req = Request::builder() let req = Request::builder()
.uri(&*format!("http://{}/a", addr)) .uri(&*format!("http://{}/a", addr))
@@ -1029,9 +1033,9 @@ mod dispatch_impl {
let _ = tx1.send(()); let _ = tx1.send(());
}); });
let client = Client::configure() let client = Client::builder()
.connector(DebugConnector::with_http_and_closes(HttpConnector::new(1, &handle), closes_tx)) .executor(runtime.executor())
.executor(runtime.executor()); .build(DebugConnector::with_http_and_closes(HttpConnector::new_with_handle(1, handle.clone()), closes_tx));
let req = Request::builder() let req = Request::builder()
.uri(&*format!("http://{}/a", addr)) .uri(&*format!("http://{}/a", addr))
@@ -1068,14 +1072,14 @@ mod dispatch_impl {
let connector = DebugConnector::new(&handle); let connector = DebugConnector::new(&handle);
let connects = connector.connects.clone(); let connects = connector.connects.clone();
let client = Client::configure() let client = Client::builder()
.connector(connector) .executor(runtime.executor())
.executor(runtime.executor()); .build(connector);
assert_eq!(connects.load(Ordering::Relaxed), 0); assert_eq!(connects.load(Ordering::Relaxed), 0);
let req = Request::builder() let req = Request::builder()
.uri("http://hyper.local/a") .uri("http://hyper.local/a")
.body(Default::default()) .body(Body::empty())
.unwrap(); .unwrap();
let _fut = client.request(req); let _fut = client.request(req);
// internal Connect::connect should have been lazy, and not // internal Connect::connect should have been lazy, and not
@@ -1093,9 +1097,9 @@ mod dispatch_impl {
let connector = DebugConnector::new(&handle); let connector = DebugConnector::new(&handle);
let connects = connector.connects.clone(); let connects = connector.connects.clone();
let client = Client::configure() let client = Client::builder()
.connector(connector) .executor(runtime.executor())
.executor(runtime.executor()); .build(connector);
let (tx1, rx1) = oneshot::channel(); let (tx1, rx1) = oneshot::channel();
let (tx2, rx2) = oneshot::channel(); let (tx2, rx2) = oneshot::channel();
@@ -1156,9 +1160,9 @@ mod dispatch_impl {
let connector = DebugConnector::new(&handle); let connector = DebugConnector::new(&handle);
let connects = connector.connects.clone(); let connects = connector.connects.clone();
let client = Client::configure() let client = Client::builder()
.connector(connector) .executor(runtime.executor())
.executor(runtime.executor()); .build(connector);
let (tx1, rx1) = oneshot::channel(); let (tx1, rx1) = oneshot::channel();
let (tx2, rx2) = oneshot::channel(); let (tx2, rx2) = oneshot::channel();
@@ -1217,9 +1221,9 @@ mod dispatch_impl {
let connector = DebugConnector::new(&handle) let connector = DebugConnector::new(&handle)
.proxy(); .proxy();
let client = Client::configure() let client = Client::builder()
.connector(connector) .executor(runtime.executor())
.executor(runtime.executor()); .build(connector);
let (tx1, rx1) = oneshot::channel(); let (tx1, rx1) = oneshot::channel();
thread::spawn(move || { thread::spawn(move || {
@@ -1256,7 +1260,7 @@ mod dispatch_impl {
impl DebugConnector { impl DebugConnector {
fn new(handle: &Handle) -> DebugConnector { fn new(handle: &Handle) -> DebugConnector {
let http = HttpConnector::new(1, handle); let http = HttpConnector::new_with_handle(1, handle.clone());
let (tx, _) = mpsc::channel(10); let (tx, _) = mpsc::channel(10);
DebugConnector::with_http_and_closes(http, tx) DebugConnector::with_http_and_closes(http, tx)
} }