1//! The legacy HTTP Client from 0.14.x
2//!
3//! This `Client` will eventually be deconstructed into more composable parts.
4//! For now, to enable people to use hyper 1.0 quicker, this `Client` exists
5//! in much the same way it did in hyper 0.14.
6
7use std::error::Error as StdError;
8use std::fmt;
9use std::future::Future;
10use std::pin::Pin;
11use std::task::{self, Poll};
12use std::time::Duration;
13
14use futures_util::future::{self, Either, FutureExt, TryFutureExt};
15use http::uri::Scheme;
16use hyper::client::conn::TrySendError as ConnTrySendError;
17use hyper::header::{HeaderValue, HOST};
18use hyper::rt::Timer;
19use hyper::{body::Body, Method, Request, Response, Uri, Version};
20use tracing::{debug, trace, warn};
21
22use super::connect::capture::CaptureConnectionExtension;
23#[cfg(feature = "tokio")]
24use super::connect::HttpConnector;
25use super::connect::{Alpn, Connect, Connected, Connection};
26use super::pool::{self, Ver};
27
28use crate::common::{lazy as hyper_lazy, timer, Exec, Lazy, SyncWrapper};
29
30type BoxSendFuture = Pin<Box<dyn Future<Output = ()> + Send>>;
31
32/// A Client to make outgoing HTTP requests.
33///
34/// `Client` is cheap to clone and cloning is the recommended way to share a `Client`. The
35/// underlying connection pool will be reused.
36#[cfg_attr(docsrs, doc(cfg(any(feature = "http1", feature = "http2"))))]
37pub struct Client<C, B> {
38 config: Config,
39 connector: C,
40 exec: Exec,
41 #[cfg(feature = "http1")]
42 h1_builder: hyper::client::conn::http1::Builder,
43 #[cfg(feature = "http2")]
44 h2_builder: hyper::client::conn::http2::Builder<Exec>,
45 pool: pool::Pool<PoolClient<B>, PoolKey>,
46}
47
48#[derive(Clone, Copy, Debug)]
49struct Config {
50 retry_canceled_requests: bool,
51 set_host: bool,
52 ver: Ver,
53}
54
55/// Client errors
56pub struct Error {
57 kind: ErrorKind,
58 source: Option<Box<dyn StdError + Send + Sync>>,
59 #[cfg(any(feature = "http1", feature = "http2"))]
60 connect_info: Option<Connected>,
61}
62
63#[derive(Debug)]
64enum ErrorKind {
65 Canceled,
66 ChannelClosed,
67 Connect,
68 UserUnsupportedRequestMethod,
69 UserUnsupportedVersion,
70 UserAbsoluteUriRequired,
71 SendRequest,
72}
73
74macro_rules! e {
75 ($kind:ident) => {
76 Error {
77 kind: ErrorKind::$kind,
78 source: None,
79 connect_info: None,
80 }
81 };
82 ($kind:ident, $src:expr) => {
83 Error {
84 kind: ErrorKind::$kind,
85 source: Some($src.into()),
86 connect_info: None,
87 }
88 };
89}
90
91// We might change this... :shrug:
92type PoolKey = (http::uri::Scheme, http::uri::Authority);
93
94enum TrySendError<B> {
95 Retryable {
96 error: Error,
97 req: Request<B>,
98 connection_reused: bool,
99 },
100 Nope(Error),
101}
102
103/// A `Future` that will resolve to an HTTP Response.
104///
105/// This is returned by `Client::request` (and `Client::get`).
106#[must_use = "futures do nothing unless polled"]
107pub struct ResponseFuture {
108 inner: SyncWrapper<
109 Pin<Box<dyn Future<Output = Result<Response<hyper::body::Incoming>, Error>> + Send>>,
110 >,
111}
112
113// ===== impl Client =====
114
115impl Client<(), ()> {
116 /// Create a builder to configure a new `Client`.
117 ///
118 /// # Example
119 ///
120 /// ```
121 /// # #[cfg(feature = "tokio")]
122 /// # fn run () {
123 /// use std::time::Duration;
124 /// use hyper_util::client::legacy::Client;
125 /// use hyper_util::rt::TokioExecutor;
126 ///
127 /// let client = Client::builder(TokioExecutor::new())
128 /// .pool_idle_timeout(Duration::from_secs(30))
129 /// .http2_only(true)
130 /// .build_http();
131 /// # let infer: Client<_, http_body_util::Full<bytes::Bytes>> = client;
132 /// # drop(infer);
133 /// # }
134 /// # fn main() {}
135 /// ```
136 pub fn builder<E>(executor: E) -> Builder
137 where
138 E: hyper::rt::Executor<BoxSendFuture> + Send + Sync + Clone + 'static,
139 {
140 Builder::new(executor)
141 }
142}
143
144impl<C, B> Client<C, B>
145where
146 C: Connect + Clone + Send + Sync + 'static,
147 B: Body + Send + 'static + Unpin,
148 B::Data: Send,
149 B::Error: Into<Box<dyn StdError + Send + Sync>>,
150{
151 /// Send a `GET` request to the supplied `Uri`.
152 ///
153 /// # Note
154 ///
155 /// This requires that the `Body` type have a `Default` implementation.
156 /// It *should* return an "empty" version of itself, such that
157 /// `Body::is_end_stream` is `true`.
158 ///
159 /// # Example
160 ///
161 /// ```
162 /// # #[cfg(feature = "tokio")]
163 /// # fn run () {
164 /// use hyper::Uri;
165 /// use hyper_util::client::legacy::Client;
166 /// use hyper_util::rt::TokioExecutor;
167 /// use bytes::Bytes;
168 /// use http_body_util::Full;
169 ///
170 /// let client: Client<_, Full<Bytes>> = Client::builder(TokioExecutor::new()).build_http();
171 ///
172 /// let future = client.get(Uri::from_static("http://httpbin.org/ip"));
173 /// # }
174 /// # fn main() {}
175 /// ```
176 pub fn get(&self, uri: Uri) -> ResponseFuture
177 where
178 B: Default,
179 {
180 let body = B::default();
181 if !body.is_end_stream() {
182 warn!("default Body used for get() does not return true for is_end_stream");
183 }
184
185 let mut req = Request::new(body);
186 *req.uri_mut() = uri;
187 self.request(req)
188 }
189
190 /// Send a constructed `Request` using this `Client`.
191 ///
192 /// # Example
193 ///
194 /// ```
195 /// # #[cfg(feature = "tokio")]
196 /// # fn run () {
197 /// use hyper::{Method, Request};
198 /// use hyper_util::client::legacy::Client;
199 /// use http_body_util::Full;
200 /// use hyper_util::rt::TokioExecutor;
201 /// use bytes::Bytes;
202 ///
203 /// let client: Client<_, Full<Bytes>> = Client::builder(TokioExecutor::new()).build_http();
204 ///
205 /// let req: Request<Full<Bytes>> = Request::builder()
206 /// .method(Method::POST)
207 /// .uri("http://httpbin.org/post")
208 /// .body(Full::from("Hallo!"))
209 /// .expect("request builder");
210 ///
211 /// let future = client.request(req);
212 /// # }
213 /// # fn main() {}
214 /// ```
215 pub fn request(&self, mut req: Request<B>) -> ResponseFuture {
216 let is_http_connect = req.method() == Method::CONNECT;
217 match req.version() {
218 Version::HTTP_11 => (),
219 Version::HTTP_10 => {
220 if is_http_connect {
221 warn!("CONNECT is not allowed for HTTP/1.0");
222 return ResponseFuture::new(future::err(e!(UserUnsupportedRequestMethod)));
223 }
224 }
225 Version::HTTP_2 => (),
226 // completely unsupported HTTP version (like HTTP/0.9)!
227 other => return ResponseFuture::error_version(other),
228 };
229
230 let pool_key = match extract_domain(req.uri_mut(), is_http_connect) {
231 Ok(s) => s,
232 Err(err) => {
233 return ResponseFuture::new(future::err(err));
234 }
235 };
236
237 ResponseFuture::new(self.clone().send_request(req, pool_key))
238 }
239
240 async fn send_request(
241 self,
242 mut req: Request<B>,
243 pool_key: PoolKey,
244 ) -> Result<Response<hyper::body::Incoming>, Error> {
245 let uri = req.uri().clone();
246
247 loop {
248 req = match self.try_send_request(req, pool_key.clone()).await {
249 Ok(resp) => return Ok(resp),
250 Err(TrySendError::Nope(err)) => return Err(err),
251 Err(TrySendError::Retryable {
252 mut req,
253 error,
254 connection_reused,
255 }) => {
256 if !self.config.retry_canceled_requests || !connection_reused {
257 // if client disabled, don't retry
258 // a fresh connection means we definitely can't retry
259 return Err(error);
260 }
261
262 trace!(
263 "unstarted request canceled, trying again (reason={:?})",
264 error
265 );
266 *req.uri_mut() = uri.clone();
267 req
268 }
269 }
270 }
271 }
272
273 async fn try_send_request(
274 &self,
275 mut req: Request<B>,
276 pool_key: PoolKey,
277 ) -> Result<Response<hyper::body::Incoming>, TrySendError<B>> {
278 let mut pooled = self
279 .connection_for(pool_key)
280 .await
281 // `connection_for` already retries checkout errors, so if
282 // it returns an error, there's not much else to retry
283 .map_err(TrySendError::Nope)?;
284
285 req.extensions_mut()
286 .get_mut::<CaptureConnectionExtension>()
287 .map(|conn| conn.set(&pooled.conn_info));
288
289 if pooled.is_http1() {
290 if req.version() == Version::HTTP_2 {
291 warn!("Connection is HTTP/1, but request requires HTTP/2");
292 return Err(TrySendError::Nope(
293 e!(UserUnsupportedVersion).with_connect_info(pooled.conn_info.clone()),
294 ));
295 }
296
297 if self.config.set_host {
298 let uri = req.uri().clone();
299 req.headers_mut().entry(HOST).or_insert_with(|| {
300 let hostname = uri.host().expect("authority implies host");
301 if let Some(port) = get_non_default_port(&uri) {
302 let s = format!("{}:{}", hostname, port);
303 HeaderValue::from_str(&s)
304 } else {
305 HeaderValue::from_str(hostname)
306 }
307 .expect("uri host is valid header value")
308 });
309 }
310
311 // CONNECT always sends authority-form, so check it first...
312 if req.method() == Method::CONNECT {
313 authority_form(req.uri_mut());
314 } else if pooled.conn_info.is_proxied {
315 absolute_form(req.uri_mut());
316 } else {
317 origin_form(req.uri_mut());
318 }
319 } else if req.method() == Method::CONNECT && !pooled.is_http2() {
320 authority_form(req.uri_mut());
321 }
322
323 let mut res = match pooled.try_send_request(req).await {
324 Ok(res) => res,
325 Err(mut err) => {
326 return if let Some(req) = err.take_message() {
327 Err(TrySendError::Retryable {
328 connection_reused: pooled.is_reused(),
329 error: e!(Canceled, err.into_error())
330 .with_connect_info(pooled.conn_info.clone()),
331 req,
332 })
333 } else {
334 Err(TrySendError::Nope(
335 e!(SendRequest, err.into_error())
336 .with_connect_info(pooled.conn_info.clone()),
337 ))
338 }
339 }
340 };
341
342 // If the Connector included 'extra' info, add to Response...
343 if let Some(extra) = &pooled.conn_info.extra {
344 extra.set(res.extensions_mut());
345 }
346
347 // If pooled is HTTP/2, we can toss this reference immediately.
348 //
349 // when pooled is dropped, it will try to insert back into the
350 // pool. To delay that, spawn a future that completes once the
351 // sender is ready again.
352 //
353 // This *should* only be once the related `Connection` has polled
354 // for a new request to start.
355 //
356 // It won't be ready if there is a body to stream.
357 if pooled.is_http2() || !pooled.is_pool_enabled() || pooled.is_ready() {
358 drop(pooled);
359 } else if !res.body().is_end_stream() {
360 //let (delayed_tx, delayed_rx) = oneshot::channel::<()>();
361 //res.body_mut().delayed_eof(delayed_rx);
362 let on_idle = future::poll_fn(move |cx| pooled.poll_ready(cx)).map(move |_| {
363 // At this point, `pooled` is dropped, and had a chance
364 // to insert into the pool (if conn was idle)
365 //drop(delayed_tx);
366 });
367
368 self.exec.execute(on_idle);
369 } else {
370 // There's no body to delay, but the connection isn't
371 // ready yet. Only re-insert when it's ready
372 let on_idle = future::poll_fn(move |cx| pooled.poll_ready(cx)).map(|_| ());
373
374 self.exec.execute(on_idle);
375 }
376
377 Ok(res)
378 }
379
380 async fn connection_for(
381 &self,
382 pool_key: PoolKey,
383 ) -> Result<pool::Pooled<PoolClient<B>, PoolKey>, Error> {
384 loop {
385 match self.one_connection_for(pool_key.clone()).await {
386 Ok(pooled) => return Ok(pooled),
387 Err(ClientConnectError::Normal(err)) => return Err(err),
388 Err(ClientConnectError::CheckoutIsClosed(reason)) => {
389 if !self.config.retry_canceled_requests {
390 return Err(e!(Connect, reason));
391 }
392
393 trace!(
394 "unstarted request canceled, trying again (reason={:?})",
395 reason,
396 );
397 continue;
398 }
399 };
400 }
401 }
402
403 async fn one_connection_for(
404 &self,
405 pool_key: PoolKey,
406 ) -> Result<pool::Pooled<PoolClient<B>, PoolKey>, ClientConnectError> {
407 // Return a single connection if pooling is not enabled
408 if !self.pool.is_enabled() {
409 return self
410 .connect_to(pool_key)
411 .await
412 .map_err(ClientConnectError::Normal);
413 }
414
415 // This actually races 2 different futures to try to get a ready
416 // connection the fastest, and to reduce connection churn.
417 //
418 // - If the pool has an idle connection waiting, that's used
419 // immediately.
420 // - Otherwise, the Connector is asked to start connecting to
421 // the destination Uri.
422 // - Meanwhile, the pool Checkout is watching to see if any other
423 // request finishes and tries to insert an idle connection.
424 // - If a new connection is started, but the Checkout wins after
425 // (an idle connection became available first), the started
426 // connection future is spawned into the runtime to complete,
427 // and then be inserted into the pool as an idle connection.
428 let checkout = self.pool.checkout(pool_key.clone());
429 let connect = self.connect_to(pool_key);
430 let is_ver_h2 = self.config.ver == Ver::Http2;
431
432 // The order of the `select` is depended on below...
433
434 match future::select(checkout, connect).await {
435 // Checkout won, connect future may have been started or not.
436 //
437 // If it has, let it finish and insert back into the pool,
438 // so as to not waste the socket...
439 Either::Left((Ok(checked_out), connecting)) => {
440 // This depends on the `select` above having the correct
441 // order, such that if the checkout future were ready
442 // immediately, the connect future will never have been
443 // started.
444 //
445 // If it *wasn't* ready yet, then the connect future will
446 // have been started...
447 if connecting.started() {
448 let bg = connecting
449 .map_err(|err| {
450 trace!("background connect error: {}", err);
451 })
452 .map(|_pooled| {
453 // dropping here should just place it in
454 // the Pool for us...
455 });
456 // An execute error here isn't important, we're just trying
457 // to prevent a waste of a socket...
458 self.exec.execute(bg);
459 }
460 Ok(checked_out)
461 }
462 // Connect won, checkout can just be dropped.
463 Either::Right((Ok(connected), _checkout)) => Ok(connected),
464 // Either checkout or connect could get canceled:
465 //
466 // 1. Connect is canceled if this is HTTP/2 and there is
467 // an outstanding HTTP/2 connecting task.
468 // 2. Checkout is canceled if the pool cannot deliver an
469 // idle connection reliably.
470 //
471 // In both cases, we should just wait for the other future.
472 Either::Left((Err(err), connecting)) => {
473 if err.is_canceled() {
474 connecting.await.map_err(ClientConnectError::Normal)
475 } else {
476 Err(ClientConnectError::Normal(e!(Connect, err)))
477 }
478 }
479 Either::Right((Err(err), checkout)) => {
480 if err.is_canceled() {
481 checkout.await.map_err(move |err| {
482 if is_ver_h2 && err.is_canceled() {
483 ClientConnectError::CheckoutIsClosed(err)
484 } else {
485 ClientConnectError::Normal(e!(Connect, err))
486 }
487 })
488 } else {
489 Err(ClientConnectError::Normal(err))
490 }
491 }
492 }
493 }
494
495 #[cfg(any(feature = "http1", feature = "http2"))]
496 fn connect_to(
497 &self,
498 pool_key: PoolKey,
499 ) -> impl Lazy<Output = Result<pool::Pooled<PoolClient<B>, PoolKey>, Error>> + Send + Unpin
500 {
501 let executor = self.exec.clone();
502 let pool = self.pool.clone();
503 #[cfg(feature = "http1")]
504 let h1_builder = self.h1_builder.clone();
505 #[cfg(feature = "http2")]
506 let h2_builder = self.h2_builder.clone();
507 let ver = self.config.ver;
508 let is_ver_h2 = ver == Ver::Http2;
509 let connector = self.connector.clone();
510 let dst = domain_as_uri(pool_key.clone());
511 hyper_lazy(move || {
512 // Try to take a "connecting lock".
513 //
514 // If the pool_key is for HTTP/2, and there is already a
515 // connection being established, then this can't take a
516 // second lock. The "connect_to" future is Canceled.
517 let connecting = match pool.connecting(&pool_key, ver) {
518 Some(lock) => lock,
519 None => {
520 let canceled = e!(Canceled);
521 // TODO
522 //crate::Error::new_canceled().with("HTTP/2 connection in progress");
523 return Either::Right(future::err(canceled));
524 }
525 };
526 Either::Left(
527 connector
528 .connect(super::connect::sealed::Internal, dst)
529 .map_err(|src| e!(Connect, src))
530 .and_then(move |io| {
531 let connected = io.connected();
532 // If ALPN is h2 and we aren't http2_only already,
533 // then we need to convert our pool checkout into
534 // a single HTTP2 one.
535 let connecting = if connected.alpn == Alpn::H2 && !is_ver_h2 {
536 match connecting.alpn_h2(&pool) {
537 Some(lock) => {
538 trace!("ALPN negotiated h2, updating pool");
539 lock
540 }
541 None => {
542 // Another connection has already upgraded,
543 // the pool checkout should finish up for us.
544 let canceled = e!(Canceled, "ALPN upgraded to HTTP/2");
545 return Either::Right(future::err(canceled));
546 }
547 }
548 } else {
549 connecting
550 };
551
552 #[cfg_attr(not(feature = "http2"), allow(unused))]
553 let is_h2 = is_ver_h2 || connected.alpn == Alpn::H2;
554
555 Either::Left(Box::pin(async move {
556 let tx = if is_h2 {
557 #[cfg(feature = "http2")] {
558 let (mut tx, conn) =
559 h2_builder.handshake(io).await.map_err(Error::tx)?;
560
561 trace!(
562 "http2 handshake complete, spawning background dispatcher task"
563 );
564 executor.execute(
565 conn.map_err(|e| debug!("client connection error: {}", e))
566 .map(|_| ()),
567 );
568
569 // Wait for 'conn' to ready up before we
570 // declare this tx as usable
571 tx.ready().await.map_err(Error::tx)?;
572 PoolTx::Http2(tx)
573 }
574 #[cfg(not(feature = "http2"))]
575 panic!("http2 feature is not enabled");
576 } else {
577 #[cfg(feature = "http1")] {
578 let (mut tx, conn) =
579 h1_builder.handshake(io).await.map_err(Error::tx)?;
580
581 trace!(
582 "http1 handshake complete, spawning background dispatcher task"
583 );
584 executor.execute(
585 conn.with_upgrades()
586 .map_err(|e| debug!("client connection error: {}", e))
587 .map(|_| ()),
588 );
589
590 // Wait for 'conn' to ready up before we
591 // declare this tx as usable
592 tx.ready().await.map_err(Error::tx)?;
593 PoolTx::Http1(tx)
594 }
595 #[cfg(not(feature = "http1"))] {
596 panic!("http1 feature is not enabled");
597 }
598 };
599
600 Ok(pool.pooled(
601 connecting,
602 PoolClient {
603 conn_info: connected,
604 tx,
605 },
606 ))
607 }))
608 }),
609 )
610 })
611 }
612}
613
614impl<C, B> tower_service::Service<Request<B>> for Client<C, B>
615where
616 C: Connect + Clone + Send + Sync + 'static,
617 B: Body + Send + 'static + Unpin,
618 B::Data: Send,
619 B::Error: Into<Box<dyn StdError + Send + Sync>>,
620{
621 type Response = Response<hyper::body::Incoming>;
622 type Error = Error;
623 type Future = ResponseFuture;
624
625 fn poll_ready(&mut self, _: &mut task::Context<'_>) -> Poll<Result<(), Self::Error>> {
626 Poll::Ready(Ok(()))
627 }
628
629 fn call(&mut self, req: Request<B>) -> Self::Future {
630 self.request(req)
631 }
632}
633
634impl<C, B> tower_service::Service<Request<B>> for &'_ Client<C, B>
635where
636 C: Connect + Clone + Send + Sync + 'static,
637 B: Body + Send + 'static + Unpin,
638 B::Data: Send,
639 B::Error: Into<Box<dyn StdError + Send + Sync>>,
640{
641 type Response = Response<hyper::body::Incoming>;
642 type Error = Error;
643 type Future = ResponseFuture;
644
645 fn poll_ready(&mut self, _: &mut task::Context<'_>) -> Poll<Result<(), Self::Error>> {
646 Poll::Ready(Ok(()))
647 }
648
649 fn call(&mut self, req: Request<B>) -> Self::Future {
650 self.request(req)
651 }
652}
653
654impl<C: Clone, B> Clone for Client<C, B> {
655 fn clone(&self) -> Client<C, B> {
656 Client {
657 config: self.config,
658 exec: self.exec.clone(),
659 #[cfg(feature = "http1")]
660 h1_builder: self.h1_builder.clone(),
661 #[cfg(feature = "http2")]
662 h2_builder: self.h2_builder.clone(),
663 connector: self.connector.clone(),
664 pool: self.pool.clone(),
665 }
666 }
667}
668
669impl<C, B> fmt::Debug for Client<C, B> {
670 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
671 f.debug_struct(name:"Client").finish()
672 }
673}
674
675// ===== impl ResponseFuture =====
676
677impl ResponseFuture {
678 fn new<F>(value: F) -> Self
679 where
680 F: Future<Output = Result<Response<hyper::body::Incoming>, Error>> + Send + 'static,
681 {
682 Self {
683 inner: SyncWrapper::new(Box::pin(value)),
684 }
685 }
686
687 fn error_version(ver: Version) -> Self {
688 warn!("Request has unsupported version \"{:?}\"", ver);
689 ResponseFuture::new(Box::pin(future::err(e!(UserUnsupportedVersion))))
690 }
691}
692
693impl fmt::Debug for ResponseFuture {
694 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
695 f.pad("Future<Response>")
696 }
697}
698
699impl Future for ResponseFuture {
700 type Output = Result<Response<hyper::body::Incoming>, Error>;
701
702 fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
703 self.inner.get_mut().as_mut().poll(cx)
704 }
705}
706
707// ===== impl PoolClient =====
708
709// FIXME: allow() required due to `impl Trait` leaking types to this lint
710#[allow(missing_debug_implementations)]
711struct PoolClient<B> {
712 conn_info: Connected,
713 tx: PoolTx<B>,
714}
715
716enum PoolTx<B> {
717 #[cfg(feature = "http1")]
718 Http1(hyper::client::conn::http1::SendRequest<B>),
719 #[cfg(feature = "http2")]
720 Http2(hyper::client::conn::http2::SendRequest<B>),
721}
722
723impl<B> PoolClient<B> {
724 fn poll_ready(
725 &mut self,
726 #[allow(unused_variables)] cx: &mut task::Context<'_>,
727 ) -> Poll<Result<(), Error>> {
728 match self.tx {
729 #[cfg(feature = "http1")]
730 PoolTx::Http1(ref mut tx) => tx.poll_ready(cx).map_err(Error::closed),
731 #[cfg(feature = "http2")]
732 PoolTx::Http2(_) => Poll::Ready(Ok(())),
733 }
734 }
735
736 fn is_http1(&self) -> bool {
737 !self.is_http2()
738 }
739
740 fn is_http2(&self) -> bool {
741 match self.tx {
742 #[cfg(feature = "http1")]
743 PoolTx::Http1(_) => false,
744 #[cfg(feature = "http2")]
745 PoolTx::Http2(_) => true,
746 }
747 }
748
749 fn is_poisoned(&self) -> bool {
750 self.conn_info.poisoned.poisoned()
751 }
752
753 fn is_ready(&self) -> bool {
754 match self.tx {
755 #[cfg(feature = "http1")]
756 PoolTx::Http1(ref tx) => tx.is_ready(),
757 #[cfg(feature = "http2")]
758 PoolTx::Http2(ref tx) => tx.is_ready(),
759 }
760 }
761}
762
763impl<B: Body + 'static> PoolClient<B> {
764 fn try_send_request(
765 &mut self,
766 req: Request<B>,
767 ) -> impl Future<Output = Result<Response<hyper::body::Incoming>, ConnTrySendError<Request<B>>>>
768 where
769 B: Send,
770 {
771 #[cfg(all(feature = "http1", feature = "http2"))]
772 return match self.tx {
773 #[cfg(feature = "http1")]
774 PoolTx::Http1(ref mut tx) => Either::Left(tx.try_send_request(req)),
775 #[cfg(feature = "http2")]
776 PoolTx::Http2(ref mut tx) => Either::Right(tx.try_send_request(req)),
777 };
778
779 #[cfg(feature = "http1")]
780 #[cfg(not(feature = "http2"))]
781 return match self.tx {
782 #[cfg(feature = "http1")]
783 PoolTx::Http1(ref mut tx) => tx.try_send_request(req),
784 };
785
786 #[cfg(not(feature = "http1"))]
787 #[cfg(feature = "http2")]
788 return match self.tx {
789 #[cfg(feature = "http2")]
790 PoolTx::Http2(ref mut tx) => tx.try_send_request(req),
791 };
792 }
793}
794
795impl<B> pool::Poolable for PoolClient<B>
796where
797 B: Send + 'static,
798{
799 fn is_open(&self) -> bool {
800 !self.is_poisoned() && self.is_ready()
801 }
802
803 fn reserve(self) -> pool::Reservation<Self> {
804 match self.tx {
805 #[cfg(feature = "http1")]
806 PoolTx::Http1(tx) => pool::Reservation::Unique(PoolClient {
807 conn_info: self.conn_info,
808 tx: PoolTx::Http1(tx),
809 }),
810 #[cfg(feature = "http2")]
811 PoolTx::Http2(tx) => {
812 let b = PoolClient {
813 conn_info: self.conn_info.clone(),
814 tx: PoolTx::Http2(tx.clone()),
815 };
816 let a = PoolClient {
817 conn_info: self.conn_info,
818 tx: PoolTx::Http2(tx),
819 };
820 pool::Reservation::Shared(a, b)
821 }
822 }
823 }
824
825 fn can_share(&self) -> bool {
826 self.is_http2()
827 }
828}
829
830enum ClientConnectError {
831 Normal(Error),
832 CheckoutIsClosed(pool::Error),
833}
834
835fn origin_form(uri: &mut Uri) {
836 let path: Uri = match uri.path_and_query() {
837 Some(path: &PathAndQuery) if path.as_str() != "/" => {
838 let mut parts: Parts = ::http::uri::Parts::default();
839 parts.path_and_query = Some(path.clone());
840 Uri::from_parts(parts).expect(msg:"path is valid uri")
841 }
842 _none_or_just_slash: Option<&PathAndQuery> => {
843 debug_assert!(Uri::default() == "/");
844 Uri::default()
845 }
846 };
847 *uri = path
848}
849
850fn absolute_form(uri: &mut Uri) {
851 debug_assert!(uri.scheme().is_some(), "absolute_form needs a scheme");
852 debug_assert!(
853 uri.authority().is_some(),
854 "absolute_form needs an authority"
855 );
856 // If the URI is to HTTPS, and the connector claimed to be a proxy,
857 // then it *should* have tunneled, and so we don't want to send
858 // absolute-form in that case.
859 if uri.scheme() == Some(&Scheme::HTTPS) {
860 origin_form(uri);
861 }
862}
863
864fn authority_form(uri: &mut Uri) {
865 if let Some(path: &PathAndQuery) = uri.path_and_query() {
866 // `https://hyper.rs` would parse with `/` path, don't
867 // annoy people about that...
868 if path != "/" {
869 warn!("HTTP/1.1 CONNECT request stripping path: {:?}", path);
870 }
871 }
872 *uri = match uri.authority() {
873 Some(auth: &Authority) => {
874 let mut parts: Parts = ::http::uri::Parts::default();
875 parts.authority = Some(auth.clone());
876 Uri::from_parts(parts).expect(msg:"authority is valid")
877 }
878 None => {
879 unreachable!("authority_form with relative uri");
880 }
881 };
882}
883
884fn extract_domain(uri: &mut Uri, is_http_connect: bool) -> Result<PoolKey, Error> {
885 let uri_clone: Uri = uri.clone();
886 match (uri_clone.scheme(), uri_clone.authority()) {
887 (Some(scheme: &Scheme), Some(auth: &Authority)) => Ok((scheme.clone(), auth.clone())),
888 (None, Some(auth: &Authority)) if is_http_connect => {
889 let scheme: Scheme = match auth.port_u16() {
890 Some(443) => {
891 set_scheme(uri, scheme:Scheme::HTTPS);
892 Scheme::HTTPS
893 }
894 _ => {
895 set_scheme(uri, scheme:Scheme::HTTP);
896 Scheme::HTTP
897 }
898 };
899 Ok((scheme, auth.clone()))
900 }
901 _ => {
902 debug!("Client requires absolute-form URIs, received: {:?}", uri);
903 Err(e!(UserAbsoluteUriRequired))
904 }
905 }
906}
907
908fn domain_as_uri((scheme: Scheme, auth: Authority): PoolKey) -> Uri {
909 http::uri::Builder::new()
910 .scheme(scheme)
911 .authority(auth)
912 .path_and_query("/")
913 .build()
914 .expect(msg:"domain is valid Uri")
915}
916
917fn set_scheme(uri: &mut Uri, scheme: Scheme) {
918 debug_assert!(
919 uri.scheme().is_none(),
920 "set_scheme expects no existing scheme"
921 );
922 let old: Uri = std::mem::take(dest:uri);
923 let mut parts: ::http::uri::Parts = old.into();
924 parts.scheme = Some(scheme);
925 parts.path_and_query = Some("/".parse().expect(msg:"slash is a valid path"));
926 *uri = Uri::from_parts(parts).expect(msg:"scheme is valid");
927}
928
929fn get_non_default_port(uri: &Uri) -> Option<http::uri::Port<&str>> {
930 match (uri.port().map(|p: Port<&str>| p.as_u16()), is_schema_secure(uri)) {
931 (Some(443), true) => None,
932 (Some(80), false) => None,
933 _ => uri.port(),
934 }
935}
936
937fn is_schema_secure(uri: &Uri) -> bool {
938 uriOption.scheme_str()
939 .map(|scheme_str: &str| matches!(scheme_str, "wss" | "https"))
940 .unwrap_or_default()
941}
942
943/// A builder to configure a new [`Client`](Client).
944///
945/// # Example
946///
947/// ```
948/// # #[cfg(feature = "tokio")]
949/// # fn run () {
950/// use std::time::Duration;
951/// use hyper_util::client::legacy::Client;
952/// use hyper_util::rt::TokioExecutor;
953///
954/// let client = Client::builder(TokioExecutor::new())
955/// .pool_idle_timeout(Duration::from_secs(30))
956/// .http2_only(true)
957/// .build_http();
958/// # let infer: Client<_, http_body_util::Full<bytes::Bytes>> = client;
959/// # drop(infer);
960/// # }
961/// # fn main() {}
962/// ```
963#[cfg_attr(docsrs, doc(cfg(any(feature = "http1", feature = "http2"))))]
964#[derive(Clone)]
965pub struct Builder {
966 client_config: Config,
967 exec: Exec,
968 #[cfg(feature = "http1")]
969 h1_builder: hyper::client::conn::http1::Builder,
970 #[cfg(feature = "http2")]
971 h2_builder: hyper::client::conn::http2::Builder<Exec>,
972 pool_config: pool::Config,
973 pool_timer: Option<timer::Timer>,
974}
975
976impl Builder {
977 /// Construct a new Builder.
978 pub fn new<E>(executor: E) -> Self
979 where
980 E: hyper::rt::Executor<BoxSendFuture> + Send + Sync + Clone + 'static,
981 {
982 let exec = Exec::new(executor);
983 Self {
984 client_config: Config {
985 retry_canceled_requests: true,
986 set_host: true,
987 ver: Ver::Auto,
988 },
989 exec: exec.clone(),
990 #[cfg(feature = "http1")]
991 h1_builder: hyper::client::conn::http1::Builder::new(),
992 #[cfg(feature = "http2")]
993 h2_builder: hyper::client::conn::http2::Builder::new(exec),
994 pool_config: pool::Config {
995 idle_timeout: Some(Duration::from_secs(90)),
996 max_idle_per_host: usize::MAX,
997 },
998 pool_timer: None,
999 }
1000 }
1001 /// Set an optional timeout for idle sockets being kept-alive.
1002 /// A `Timer` is required for this to take effect. See `Builder::pool_timer`
1003 ///
1004 /// Pass `None` to disable timeout.
1005 ///
1006 /// Default is 90 seconds.
1007 ///
1008 /// # Example
1009 ///
1010 /// ```
1011 /// # #[cfg(feature = "tokio")]
1012 /// # fn run () {
1013 /// use std::time::Duration;
1014 /// use hyper_util::client::legacy::Client;
1015 /// use hyper_util::rt::{TokioExecutor, TokioTimer};
1016 ///
1017 /// let client = Client::builder(TokioExecutor::new())
1018 /// .pool_idle_timeout(Duration::from_secs(30))
1019 /// .pool_timer(TokioTimer::new())
1020 /// .build_http();
1021 ///
1022 /// # let infer: Client<_, http_body_util::Full<bytes::Bytes>> = client;
1023 /// # }
1024 /// # fn main() {}
1025 /// ```
1026 pub fn pool_idle_timeout<D>(&mut self, val: D) -> &mut Self
1027 where
1028 D: Into<Option<Duration>>,
1029 {
1030 self.pool_config.idle_timeout = val.into();
1031 self
1032 }
1033
1034 #[doc(hidden)]
1035 #[deprecated(note = "renamed to `pool_max_idle_per_host`")]
1036 pub fn max_idle_per_host(&mut self, max_idle: usize) -> &mut Self {
1037 self.pool_config.max_idle_per_host = max_idle;
1038 self
1039 }
1040
1041 /// Sets the maximum idle connection per host allowed in the pool.
1042 ///
1043 /// Default is `usize::MAX` (no limit).
1044 pub fn pool_max_idle_per_host(&mut self, max_idle: usize) -> &mut Self {
1045 self.pool_config.max_idle_per_host = max_idle;
1046 self
1047 }
1048
1049 // HTTP/1 options
1050
1051 /// Sets the exact size of the read buffer to *always* use.
1052 ///
1053 /// Note that setting this option unsets the `http1_max_buf_size` option.
1054 ///
1055 /// Default is an adaptive read buffer.
1056 #[cfg(feature = "http1")]
1057 #[cfg_attr(docsrs, doc(cfg(feature = "http1")))]
1058 pub fn http1_read_buf_exact_size(&mut self, sz: usize) -> &mut Self {
1059 self.h1_builder.read_buf_exact_size(Some(sz));
1060 self
1061 }
1062
1063 /// Set the maximum buffer size for the connection.
1064 ///
1065 /// Default is ~400kb.
1066 ///
1067 /// Note that setting this option unsets the `http1_read_exact_buf_size` option.
1068 ///
1069 /// # Panics
1070 ///
1071 /// The minimum value allowed is 8192. This method panics if the passed `max` is less than the minimum.
1072 #[cfg(feature = "http1")]
1073 #[cfg_attr(docsrs, doc(cfg(feature = "http1")))]
1074 pub fn http1_max_buf_size(&mut self, max: usize) -> &mut Self {
1075 self.h1_builder.max_buf_size(max);
1076 self
1077 }
1078
1079 /// Set whether HTTP/1 connections will accept spaces between header names
1080 /// and the colon that follow them in responses.
1081 ///
1082 /// Newline codepoints (`\r` and `\n`) will be transformed to spaces when
1083 /// parsing.
1084 ///
1085 /// You probably don't need this, here is what [RFC 7230 Section 3.2.4.] has
1086 /// to say about it:
1087 ///
1088 /// > No whitespace is allowed between the header field-name and colon. In
1089 /// > the past, differences in the handling of such whitespace have led to
1090 /// > security vulnerabilities in request routing and response handling. A
1091 /// > server MUST reject any received request message that contains
1092 /// > whitespace between a header field-name and colon with a response code
1093 /// > of 400 (Bad Request). A proxy MUST remove any such whitespace from a
1094 /// > response message before forwarding the message downstream.
1095 ///
1096 /// Note that this setting does not affect HTTP/2.
1097 ///
1098 /// Default is false.
1099 ///
1100 /// [RFC 7230 Section 3.2.4.]: https://tools.ietf.org/html/rfc7230#section-3.2.4
1101 #[cfg(feature = "http1")]
1102 #[cfg_attr(docsrs, doc(cfg(feature = "http1")))]
1103 pub fn http1_allow_spaces_after_header_name_in_responses(&mut self, val: bool) -> &mut Self {
1104 self.h1_builder
1105 .allow_spaces_after_header_name_in_responses(val);
1106 self
1107 }
1108
1109 /// Set whether HTTP/1 connections will accept obsolete line folding for
1110 /// header values.
1111 ///
1112 /// You probably don't need this, here is what [RFC 7230 Section 3.2.4.] has
1113 /// to say about it:
1114 ///
1115 /// > A server that receives an obs-fold in a request message that is not
1116 /// > within a message/http container MUST either reject the message by
1117 /// > sending a 400 (Bad Request), preferably with a representation
1118 /// > explaining that obsolete line folding is unacceptable, or replace
1119 /// > each received obs-fold with one or more SP octets prior to
1120 /// > interpreting the field value or forwarding the message downstream.
1121 ///
1122 /// > A proxy or gateway that receives an obs-fold in a response message
1123 /// > that is not within a message/http container MUST either discard the
1124 /// > message and replace it with a 502 (Bad Gateway) response, preferably
1125 /// > with a representation explaining that unacceptable line folding was
1126 /// > received, or replace each received obs-fold with one or more SP
1127 /// > octets prior to interpreting the field value or forwarding the
1128 /// > message downstream.
1129 ///
1130 /// > A user agent that receives an obs-fold in a response message that is
1131 /// > not within a message/http container MUST replace each received
1132 /// > obs-fold with one or more SP octets prior to interpreting the field
1133 /// > value.
1134 ///
1135 /// Note that this setting does not affect HTTP/2.
1136 ///
1137 /// Default is false.
1138 ///
1139 /// [RFC 7230 Section 3.2.4.]: https://tools.ietf.org/html/rfc7230#section-3.2.4
1140 #[cfg(feature = "http1")]
1141 #[cfg_attr(docsrs, doc(cfg(feature = "http1")))]
1142 pub fn http1_allow_obsolete_multiline_headers_in_responses(&mut self, val: bool) -> &mut Self {
1143 self.h1_builder
1144 .allow_obsolete_multiline_headers_in_responses(val);
1145 self
1146 }
1147
1148 /// Sets whether invalid header lines should be silently ignored in HTTP/1 responses.
1149 ///
1150 /// This mimics the behaviour of major browsers. You probably don't want this.
1151 /// You should only want this if you are implementing a proxy whose main
1152 /// purpose is to sit in front of browsers whose users access arbitrary content
1153 /// which may be malformed, and they expect everything that works without
1154 /// the proxy to keep working with the proxy.
1155 ///
1156 /// This option will prevent Hyper's client from returning an error encountered
1157 /// when parsing a header, except if the error was caused by the character NUL
1158 /// (ASCII code 0), as Chrome specifically always reject those.
1159 ///
1160 /// The ignorable errors are:
1161 /// * empty header names;
1162 /// * characters that are not allowed in header names, except for `\0` and `\r`;
1163 /// * when `allow_spaces_after_header_name_in_responses` is not enabled,
1164 /// spaces and tabs between the header name and the colon;
1165 /// * missing colon between header name and colon;
1166 /// * characters that are not allowed in header values except for `\0` and `\r`.
1167 ///
1168 /// If an ignorable error is encountered, the parser tries to find the next
1169 /// line in the input to resume parsing the rest of the headers. An error
1170 /// will be emitted nonetheless if it finds `\0` or a lone `\r` while
1171 /// looking for the next line.
1172 #[cfg(feature = "http1")]
1173 #[cfg_attr(docsrs, doc(cfg(feature = "http1")))]
1174 pub fn http1_ignore_invalid_headers_in_responses(&mut self, val: bool) -> &mut Builder {
1175 self.h1_builder.ignore_invalid_headers_in_responses(val);
1176 self
1177 }
1178
1179 /// Set whether HTTP/1 connections should try to use vectored writes,
1180 /// or always flatten into a single buffer.
1181 ///
1182 /// Note that setting this to false may mean more copies of body data,
1183 /// but may also improve performance when an IO transport doesn't
1184 /// support vectored writes well, such as most TLS implementations.
1185 ///
1186 /// Setting this to true will force hyper to use queued strategy
1187 /// which may eliminate unnecessary cloning on some TLS backends
1188 ///
1189 /// Default is `auto`. In this mode hyper will try to guess which
1190 /// mode to use
1191 #[cfg(feature = "http1")]
1192 #[cfg_attr(docsrs, doc(cfg(feature = "http1")))]
1193 pub fn http1_writev(&mut self, enabled: bool) -> &mut Builder {
1194 self.h1_builder.writev(enabled);
1195 self
1196 }
1197
1198 /// Set whether HTTP/1 connections will write header names as title case at
1199 /// the socket level.
1200 ///
1201 /// Note that this setting does not affect HTTP/2.
1202 ///
1203 /// Default is false.
1204 #[cfg(feature = "http1")]
1205 #[cfg_attr(docsrs, doc(cfg(feature = "http1")))]
1206 pub fn http1_title_case_headers(&mut self, val: bool) -> &mut Self {
1207 self.h1_builder.title_case_headers(val);
1208 self
1209 }
1210
1211 /// Set whether to support preserving original header cases.
1212 ///
1213 /// Currently, this will record the original cases received, and store them
1214 /// in a private extension on the `Response`. It will also look for and use
1215 /// such an extension in any provided `Request`.
1216 ///
1217 /// Since the relevant extension is still private, there is no way to
1218 /// interact with the original cases. The only effect this can have now is
1219 /// to forward the cases in a proxy-like fashion.
1220 ///
1221 /// Note that this setting does not affect HTTP/2.
1222 ///
1223 /// Default is false.
1224 #[cfg(feature = "http1")]
1225 #[cfg_attr(docsrs, doc(cfg(feature = "http1")))]
1226 pub fn http1_preserve_header_case(&mut self, val: bool) -> &mut Self {
1227 self.h1_builder.preserve_header_case(val);
1228 self
1229 }
1230
1231 /// Set the maximum number of headers.
1232 ///
1233 /// When a response is received, the parser will reserve a buffer to store headers for optimal
1234 /// performance.
1235 ///
1236 /// If client receives more headers than the buffer size, the error "message header too large"
1237 /// is returned.
1238 ///
1239 /// The headers is allocated on the stack by default, which has higher performance. After
1240 /// setting this value, headers will be allocated in heap memory, that is, heap memory
1241 /// allocation will occur for each response, and there will be a performance drop of about 5%.
1242 ///
1243 /// Note that this setting does not affect HTTP/2.
1244 ///
1245 /// Default is 100.
1246 #[cfg(feature = "http1")]
1247 #[cfg_attr(docsrs, doc(cfg(feature = "http1")))]
1248 pub fn http1_max_headers(&mut self, val: usize) -> &mut Self {
1249 self.h1_builder.max_headers(val);
1250 self
1251 }
1252
1253 /// Set whether HTTP/0.9 responses should be tolerated.
1254 ///
1255 /// Default is false.
1256 #[cfg(feature = "http1")]
1257 #[cfg_attr(docsrs, doc(cfg(feature = "http1")))]
1258 pub fn http09_responses(&mut self, val: bool) -> &mut Self {
1259 self.h1_builder.http09_responses(val);
1260 self
1261 }
1262
1263 /// Set whether the connection **must** use HTTP/2.
1264 ///
1265 /// The destination must either allow HTTP2 Prior Knowledge, or the
1266 /// `Connect` should be configured to do use ALPN to upgrade to `h2`
1267 /// as part of the connection process. This will not make the `Client`
1268 /// utilize ALPN by itself.
1269 ///
1270 /// Note that setting this to true prevents HTTP/1 from being allowed.
1271 ///
1272 /// Default is false.
1273 #[cfg(feature = "http2")]
1274 #[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
1275 pub fn http2_only(&mut self, val: bool) -> &mut Self {
1276 self.client_config.ver = if val { Ver::Http2 } else { Ver::Auto };
1277 self
1278 }
1279
1280 /// Configures the maximum number of pending reset streams allowed before a GOAWAY will be sent.
1281 ///
1282 /// This will default to the default value set by the [`h2` crate](https://crates.io/crates/h2).
1283 /// As of v0.4.0, it is 20.
1284 ///
1285 /// See <https://github.com/hyperium/hyper/issues/2877> for more information.
1286 #[cfg(feature = "http2")]
1287 #[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
1288 pub fn http2_max_pending_accept_reset_streams(
1289 &mut self,
1290 max: impl Into<Option<usize>>,
1291 ) -> &mut Self {
1292 self.h2_builder.max_pending_accept_reset_streams(max.into());
1293 self
1294 }
1295
1296 /// Sets the [`SETTINGS_INITIAL_WINDOW_SIZE`][spec] option for HTTP2
1297 /// stream-level flow control.
1298 ///
1299 /// Passing `None` will do nothing.
1300 ///
1301 /// If not set, hyper will use a default.
1302 ///
1303 /// [spec]: https://http2.github.io/http2-spec/#SETTINGS_INITIAL_WINDOW_SIZE
1304 #[cfg(feature = "http2")]
1305 #[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
1306 pub fn http2_initial_stream_window_size(&mut self, sz: impl Into<Option<u32>>) -> &mut Self {
1307 self.h2_builder.initial_stream_window_size(sz.into());
1308 self
1309 }
1310
1311 /// Sets the max connection-level flow control for HTTP2
1312 ///
1313 /// Passing `None` will do nothing.
1314 ///
1315 /// If not set, hyper will use a default.
1316 #[cfg(feature = "http2")]
1317 #[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
1318 pub fn http2_initial_connection_window_size(
1319 &mut self,
1320 sz: impl Into<Option<u32>>,
1321 ) -> &mut Self {
1322 self.h2_builder.initial_connection_window_size(sz.into());
1323 self
1324 }
1325
1326 /// Sets the initial maximum of locally initiated (send) streams.
1327 ///
1328 /// This value will be overwritten by the value included in the initial
1329 /// SETTINGS frame received from the peer as part of a [connection preface].
1330 ///
1331 /// Passing `None` will do nothing.
1332 ///
1333 /// If not set, hyper will use a default.
1334 ///
1335 /// [connection preface]: https://httpwg.org/specs/rfc9113.html#preface
1336 #[cfg(feature = "http2")]
1337 #[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
1338 pub fn http2_initial_max_send_streams(
1339 &mut self,
1340 initial: impl Into<Option<usize>>,
1341 ) -> &mut Self {
1342 self.h2_builder.initial_max_send_streams(initial);
1343 self
1344 }
1345
1346 /// Sets whether to use an adaptive flow control.
1347 ///
1348 /// Enabling this will override the limits set in
1349 /// `http2_initial_stream_window_size` and
1350 /// `http2_initial_connection_window_size`.
1351 #[cfg(feature = "http2")]
1352 #[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
1353 pub fn http2_adaptive_window(&mut self, enabled: bool) -> &mut Self {
1354 self.h2_builder.adaptive_window(enabled);
1355 self
1356 }
1357
1358 /// Sets the maximum frame size to use for HTTP2.
1359 ///
1360 /// Passing `None` will do nothing.
1361 ///
1362 /// If not set, hyper will use a default.
1363 #[cfg(feature = "http2")]
1364 #[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
1365 pub fn http2_max_frame_size(&mut self, sz: impl Into<Option<u32>>) -> &mut Self {
1366 self.h2_builder.max_frame_size(sz);
1367 self
1368 }
1369
1370 /// Sets the max size of received header frames for HTTP2.
1371 ///
1372 /// Default is currently 16KB, but can change.
1373 #[cfg(feature = "http2")]
1374 #[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
1375 pub fn http2_max_header_list_size(&mut self, max: u32) -> &mut Self {
1376 self.h2_builder.max_header_list_size(max);
1377 self
1378 }
1379
1380 /// Sets an interval for HTTP2 Ping frames should be sent to keep a
1381 /// connection alive.
1382 ///
1383 /// Pass `None` to disable HTTP2 keep-alive.
1384 ///
1385 /// Default is currently disabled.
1386 ///
1387 /// # Cargo Feature
1388 ///
1389 /// Requires the `tokio` cargo feature to be enabled.
1390 #[cfg(feature = "tokio")]
1391 #[cfg(feature = "http2")]
1392 #[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
1393 pub fn http2_keep_alive_interval(
1394 &mut self,
1395 interval: impl Into<Option<Duration>>,
1396 ) -> &mut Self {
1397 self.h2_builder.keep_alive_interval(interval);
1398 self
1399 }
1400
1401 /// Sets a timeout for receiving an acknowledgement of the keep-alive ping.
1402 ///
1403 /// If the ping is not acknowledged within the timeout, the connection will
1404 /// be closed. Does nothing if `http2_keep_alive_interval` is disabled.
1405 ///
1406 /// Default is 20 seconds.
1407 ///
1408 /// # Cargo Feature
1409 ///
1410 /// Requires the `tokio` cargo feature to be enabled.
1411 #[cfg(feature = "tokio")]
1412 #[cfg(feature = "http2")]
1413 #[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
1414 pub fn http2_keep_alive_timeout(&mut self, timeout: Duration) -> &mut Self {
1415 self.h2_builder.keep_alive_timeout(timeout);
1416 self
1417 }
1418
1419 /// Sets whether HTTP2 keep-alive should apply while the connection is idle.
1420 ///
1421 /// If disabled, keep-alive pings are only sent while there are open
1422 /// request/responses streams. If enabled, pings are also sent when no
1423 /// streams are active. Does nothing if `http2_keep_alive_interval` is
1424 /// disabled.
1425 ///
1426 /// Default is `false`.
1427 ///
1428 /// # Cargo Feature
1429 ///
1430 /// Requires the `tokio` cargo feature to be enabled.
1431 #[cfg(feature = "tokio")]
1432 #[cfg(feature = "http2")]
1433 #[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
1434 pub fn http2_keep_alive_while_idle(&mut self, enabled: bool) -> &mut Self {
1435 self.h2_builder.keep_alive_while_idle(enabled);
1436 self
1437 }
1438
1439 /// Sets the maximum number of HTTP2 concurrent locally reset streams.
1440 ///
1441 /// See the documentation of [`h2::client::Builder::max_concurrent_reset_streams`] for more
1442 /// details.
1443 ///
1444 /// The default value is determined by the `h2` crate.
1445 ///
1446 /// [`h2::client::Builder::max_concurrent_reset_streams`]: https://docs.rs/h2/client/struct.Builder.html#method.max_concurrent_reset_streams
1447 #[cfg(feature = "http2")]
1448 #[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
1449 pub fn http2_max_concurrent_reset_streams(&mut self, max: usize) -> &mut Self {
1450 self.h2_builder.max_concurrent_reset_streams(max);
1451 self
1452 }
1453
1454 /// Provide a timer to be used for h2
1455 ///
1456 /// See the documentation of [`h2::client::Builder::timer`] for more
1457 /// details.
1458 ///
1459 /// [`h2::client::Builder::timer`]: https://docs.rs/h2/client/struct.Builder.html#method.timer
1460 pub fn timer<M>(&mut self, timer: M) -> &mut Self
1461 where
1462 M: Timer + Send + Sync + 'static,
1463 {
1464 #[cfg(feature = "http2")]
1465 self.h2_builder.timer(timer);
1466 self
1467 }
1468
1469 /// Provide a timer to be used for timeouts and intervals in connection pools.
1470 pub fn pool_timer<M>(&mut self, timer: M) -> &mut Self
1471 where
1472 M: Timer + Clone + Send + Sync + 'static,
1473 {
1474 self.pool_timer = Some(timer::Timer::new(timer.clone()));
1475 self
1476 }
1477
1478 /// Set the maximum write buffer size for each HTTP/2 stream.
1479 ///
1480 /// Default is currently 1MB, but may change.
1481 ///
1482 /// # Panics
1483 ///
1484 /// The value must be no larger than `u32::MAX`.
1485 #[cfg(feature = "http2")]
1486 #[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
1487 pub fn http2_max_send_buf_size(&mut self, max: usize) -> &mut Self {
1488 self.h2_builder.max_send_buf_size(max);
1489 self
1490 }
1491
1492 /// Set whether to retry requests that get disrupted before ever starting
1493 /// to write.
1494 ///
1495 /// This means a request that is queued, and gets given an idle, reused
1496 /// connection, and then encounters an error immediately as the idle
1497 /// connection was found to be unusable.
1498 ///
1499 /// When this is set to `false`, the related `ResponseFuture` would instead
1500 /// resolve to an `Error::Cancel`.
1501 ///
1502 /// Default is `true`.
1503 #[inline]
1504 pub fn retry_canceled_requests(&mut self, val: bool) -> &mut Self {
1505 self.client_config.retry_canceled_requests = val;
1506 self
1507 }
1508
1509 /// Set whether to automatically add the `Host` header to requests.
1510 ///
1511 /// If true, and a request does not include a `Host` header, one will be
1512 /// added automatically, derived from the authority of the `Uri`.
1513 ///
1514 /// Default is `true`.
1515 #[inline]
1516 pub fn set_host(&mut self, val: bool) -> &mut Self {
1517 self.client_config.set_host = val;
1518 self
1519 }
1520
1521 /// Build a client with this configuration and the default `HttpConnector`.
1522 #[cfg(feature = "tokio")]
1523 pub fn build_http<B>(&self) -> Client<HttpConnector, B>
1524 where
1525 B: Body + Send,
1526 B::Data: Send,
1527 {
1528 let mut connector = HttpConnector::new();
1529 if self.pool_config.is_enabled() {
1530 connector.set_keepalive(self.pool_config.idle_timeout);
1531 }
1532 self.build(connector)
1533 }
1534
1535 /// Combine the configuration of this builder with a connector to create a `Client`.
1536 pub fn build<C, B>(&self, connector: C) -> Client<C, B>
1537 where
1538 C: Connect + Clone,
1539 B: Body + Send,
1540 B::Data: Send,
1541 {
1542 let exec = self.exec.clone();
1543 let timer = self.pool_timer.clone();
1544 Client {
1545 config: self.client_config,
1546 exec: exec.clone(),
1547 #[cfg(feature = "http1")]
1548 h1_builder: self.h1_builder.clone(),
1549 #[cfg(feature = "http2")]
1550 h2_builder: self.h2_builder.clone(),
1551 connector,
1552 pool: pool::Pool::new(self.pool_config, exec, timer),
1553 }
1554 }
1555}
1556
1557impl fmt::Debug for Builder {
1558 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1559 f&mut DebugStruct<'_, '_>.debug_struct("Builder")
1560 .field("client_config", &self.client_config)
1561 .field(name:"pool_config", &self.pool_config)
1562 .finish()
1563 }
1564}
1565
1566// ==== impl Error ====
1567
1568impl fmt::Debug for Error {
1569 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1570 let mut f: DebugTuple<'_, '_> = f.debug_tuple(name:"hyper_util::client::legacy::Error");
1571 f.field(&self.kind);
1572 if let Some(ref cause: &Box) = self.source {
1573 f.field(cause);
1574 }
1575 f.finish()
1576 }
1577}
1578
1579impl fmt::Display for Error {
1580 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1581 write!(f, "client error ({:?})", self.kind)
1582 }
1583}
1584
1585impl StdError for Error {
1586 fn source(&self) -> Option<&(dyn StdError + 'static)> {
1587 self.source.as_ref().map(|e: &Box| &**e as _)
1588 }
1589}
1590
1591impl Error {
1592 /// Returns true if this was an error from `Connect`.
1593 pub fn is_connect(&self) -> bool {
1594 matches!(self.kind, ErrorKind::Connect)
1595 }
1596
1597 /// Returns the info of the client connection on which this error occurred.
1598 #[cfg(any(feature = "http1", feature = "http2"))]
1599 pub fn connect_info(&self) -> Option<&Connected> {
1600 self.connect_info.as_ref()
1601 }
1602
1603 #[cfg(any(feature = "http1", feature = "http2"))]
1604 fn with_connect_info(self, connect_info: Connected) -> Self {
1605 Self {
1606 connect_info: Some(connect_info),
1607 ..self
1608 }
1609 }
1610 fn is_canceled(&self) -> bool {
1611 matches!(self.kind, ErrorKind::Canceled)
1612 }
1613
1614 fn tx(src: hyper::Error) -> Self {
1615 e!(SendRequest, src)
1616 }
1617
1618 fn closed(src: hyper::Error) -> Self {
1619 e!(ChannelClosed, src)
1620 }
1621}
1622