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}