apollo-http-client 0.2.0

//! HTTP/1 and HTTP/2 handshake adapters layered over a raw IO connector.
//!
//! Both adapters take a `Service` that returns an established IO stream and
//! produce an instrumented [`H1Connection`] or [`H2Connection`].
//!
//! Each adapter carries a base set of metric attributes shared by the
//! `http.client.open_connections` and `http.client.connection.duration`
//! instruments.

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;

use hyper::client::conn::{http1, http2};
use hyper_util::rt::{TokioExecutor, TokioTimer};
use opentelemetry::KeyValue;
use tower::BoxError;
use tower::Service;

use super::Origin;
use super::connect_with_timeout;
use super::h1::H1Connection;
use super::h2::H2Connection;
use crate::error::HttpClientError;
use crate::metrics::ConnectionMetrics;

/// Adapts a `Service<http::Uri>` (e.g. [`hyper_rustls::HttpsConnector`]) into
/// a `Service<Origin>` that synthesises a URI from the pool key and dispatches
/// it under a connect timeout. The shared TCP/TLS entry point feeding the
/// HTTP/1 and HTTP/2 handshake adapters on the direct (non-proxy) path.
#[derive(Clone)]
pub(crate) struct TcpConnect<S> {
    inner: S,
    timeout: Duration,
}

impl<S> TcpConnect<S> {
    pub(crate) fn new(inner: S, timeout: Duration) -> Self {
        Self { inner, timeout }
    }
}

impl<S, IO> Service<Origin> for TcpConnect<S>
where
    S: Service<http::Uri, Response = IO> + Clone + Send + 'static,
    S::Future: Send + 'static,
    S::Error: Into<BoxError>,
{
    type Response = IO;
    type Error = BoxError;
    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.inner.poll_ready(cx).map_err(Into::into)
    }

    fn call(&mut self, origin: Origin) -> Self::Future {
        let clone = self.inner.clone();
        let inner = std::mem::replace(&mut self.inner, clone);
        let timeout = self.timeout;
        Box::pin(async move {
            connect_with_timeout(inner, origin, timeout)
                .await
                .map_err(BoxError::from)
        })
    }
}

/// Adapts a `Service<Origin>` into `Service<()>` by pre-binding a fixed
/// pool key at construction.
///
/// Used by [`Http2Handshake`]: HTTP/2 maintains a single multiplexed
/// connection per host, so its inner connector is keyed once and reused.
#[derive(Clone)]
pub(crate) struct FixedOrigin<S> {
    pub(crate) inner: S,
    pub(crate) origin: Origin,
}

impl<S> Service<()> for FixedOrigin<S>
where
    S: Service<Origin> + Clone,
{
    type Response = S::Response;
    type Error = S::Error;
    type Future = S::Future;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.inner.poll_ready(cx)
    }

    fn call(&mut self, (): ()) -> Self::Future {
        self.inner.call(self.origin.clone())
    }
}

/// Wraps a raw IO service and performs the HTTP/1.1 handshake, returning an
/// instrumented [`H1Connection`].
#[derive(Clone)]
pub(crate) struct Http1Handshake<S> {
    pub(crate) inner: S,
    pub(crate) metrics: ConnectionMetrics,
    pub(crate) attrs: Vec<KeyValue>,
}

impl<S> Http1Handshake<S> {
    pub(crate) fn new(inner: S, metrics: ConnectionMetrics, attrs: Vec<KeyValue>) -> Self {
        Self {
            inner,
            metrics,
            attrs,
        }
    }
}

impl<S, IO> Service<Origin> for Http1Handshake<S>
where
    S: Service<Origin, Response = IO, Error = BoxError> + Clone + Send + 'static,
    S::Future: Send + 'static,
    IO: hyper::rt::Read + hyper::rt::Write + Unpin + Send + 'static,
{
    type Response = H1Connection;
    type Error = BoxError;
    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.inner.poll_ready(cx)
    }

    fn call(&mut self, origin: Origin) -> Self::Future {
        let io_fut = self.inner.call(origin);
        let metrics = self.metrics.clone();
        let attrs = self.attrs.clone();

        Box::pin(async move {
            let io = io_fut.await?;
            let (sender, conn) = http1::handshake(io)
                .await
                .map_err(|source| BoxError::from(HttpClientError::Handshake { source }))?;
            // `conn` drives the connection's I/O loop. It must be polled concurrently
            // with the sender; without it, no reads or writes happen. When `conn`
            // completes the sender becomes broken, and the next request will fail.
            // The connection result itself is discarded — there's nothing useful to do
            // with it here.
            tokio::spawn(async move {
                if let Err(e) = conn.await {
                    log::debug!(error:% = e; "HTTP/1 connection error");
                }
            });
            Ok(H1Connection::new(sender, &metrics, attrs))
        })
    }
}

/// Performs the HTTP/2 handshake on an established connection, returning an
/// instrumented [`H2Connection`].
#[derive(Clone)]
pub(crate) struct Http2Handshake<S> {
    pub(crate) inner: S,
    pub(crate) metrics: ConnectionMetrics,
    pub(crate) attrs: Vec<KeyValue>,
    pub(crate) keep_alive_interval: Duration,
    pub(crate) keep_alive_timeout: Duration,
    pub(crate) keep_alive_while_idle: bool,
}

impl<S, IO> Service<()> for Http2Handshake<S>
where
    S: Service<(), Response = IO, Error = BoxError> + Clone + Send + 'static,
    S::Future: Send + 'static,
    IO: hyper::rt::Read + hyper::rt::Write + Unpin + Send + 'static,
{
    type Response = H2Connection;
    type Error = BoxError;
    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.inner.poll_ready(cx)
    }

    fn call(&mut self, (): ()) -> Self::Future {
        let io_fut = self.inner.call(());
        let metrics = self.metrics.clone();
        let attrs = self.attrs.clone();
        let keep_alive_interval = self.keep_alive_interval;
        let keep_alive_timeout = self.keep_alive_timeout;
        let keep_alive_while_idle = self.keep_alive_while_idle;

        Box::pin(async move {
            let io = io_fut.await?;
            let (sender, conn) = http2::Builder::new(TokioExecutor::new())
                .timer(TokioTimer::new())
                .keep_alive_interval(keep_alive_interval)
                .keep_alive_timeout(keep_alive_timeout)
                .keep_alive_while_idle(keep_alive_while_idle)
                .handshake(io)
                .await
                .map_err(|source| BoxError::from(HttpClientError::Handshake { source }))?;
            // Same as H1: `conn` must be driven concurrently to keep the H2 connection
            // alive. When `conn` completes the sender becomes broken, and the next
            // request will fail. The connection result itself is discarded.
            tokio::spawn(async move {
                if let Err(e) = conn.await {
                    log::debug!(error:% = e; "HTTP/2 connection error");
                }
            });
            Ok(H2Connection::new(sender, &metrics, attrs))
        })
    }
}