hyper_util/client/legacy/
client.rs

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