hpx 2.4.10 - Docs.rs

use std::{
    net::SocketAddr,
    pin::Pin,
    task::{Context, Poll},
};

use bytes::Bytes;
#[cfg(feature = "charset")]
use encoding_rs::{Encoding, UTF_8};
#[cfg(feature = "stream")]
use futures_util::{Stream, TryStreamExt};
use http::{HeaderMap, StatusCode, Uri, Version};
use http_body::{Body as HttpBody, Frame};
use http_body_util::{BodyExt, Collected};
#[cfg(feature = "charset")]
use mime::Mime;
#[cfg(any(feature = "json", feature = "form"))]
use serde::de::DeserializeOwned;

use super::{
    ClientResponseBody,
    conn::HttpInfo,
    core::{ext::ReasonPhrase, upgrade},
};
#[cfg(feature = "cookies")]
use crate::cookie;
use crate::{Body, Error, Upgraded, error::BoxError, ext::RequestUri};

#[cfg(feature = "simd-json")]
fn simd_json_input(bytes: Bytes) -> bytes::BytesMut {
    match bytes.try_into_mut() {
        Ok(bytes_mut) => bytes_mut,
        Err(bytes) => bytes::BytesMut::from(bytes.as_ref()),
    }
}

/// A Response to a submitted [`crate::Request`].
#[derive(Debug)]
pub struct Response {
    uri: Uri,
    res: http::Response<Body>,
}

impl Response {
    pub(super) fn new<B>(res: http::Response<B>, uri: Uri) -> Response
    where
        B: HttpBody + Send + Sync + 'static,
        B::Data: Into<Bytes>,
        B::Error: Into<BoxError>,
    {
        Response {
            uri,
            res: res.map(Body::wrap),
        }
    }

    /// Construct a [`Response`] from a standard `http::Response` and the final request URI.
    ///
    /// This is the reverse bridge for the tower-native integration: when using
    /// [`TowerServiceExt::into_tower_service()`](crate::tower_compat::TowerServiceExt), the
    /// returned service yields `http::Response<ClientResponseBody>`. Use this method to
    /// convert it back to `hpx::Response` for the convenience API.
    pub fn from_http<B>(uri: Uri, res: http::Response<B>) -> Response
    where
        B: HttpBody + Send + Sync + 'static,
        B::Data: Into<Bytes>,
        B::Error: Into<BoxError>,
    {
        Response::new(res, uri)
    }

    pub(crate) fn from_client_response(uri: Uri, res: http::Response<ClientResponseBody>) -> Self {
        Response {
            uri,
            res: res.map(Body::from),
        }
    }

    /// Get the final [`Uri`] of this [`Response`].
    #[inline]
    pub fn uri(&self) -> &Uri {
        &self.uri
    }

    /// Get the [`StatusCode`] of this [`Response`].
    #[inline]
    pub fn status(&self) -> StatusCode {
        self.res.status()
    }

    /// Get the HTTP [`Version`] of this [`Response`].
    #[inline]
    pub fn version(&self) -> Version {
        self.res.version()
    }

    /// Get the [`HeaderMap`] of this [`Response`].
    #[inline]
    pub fn headers(&self) -> &HeaderMap {
        self.res.headers()
    }

    /// Get a mutable reference to the [`HeaderMap`] of this [`Response`].
    #[inline]
    pub fn headers_mut(&mut self) -> &mut HeaderMap {
        self.res.headers_mut()
    }

    /// Get the content length of the [`Response`], if it is known.
    ///
    /// This value does not directly represents the value of the `Content-Length`
    /// header, but rather the size of the response's body. To read the header's
    /// value, please use the [`Response::headers`] method instead.
    ///
    /// Reasons it may not be known:
    ///
    /// - The response does not include a body (e.g. it responds to a `HEAD` request).
    /// - The response is gzipped and automatically decoded (thus changing the actual decoded
    ///   length).
    #[inline]
    pub fn content_length(&self) -> Option<u64> {
        HttpBody::size_hint(self.res.body()).exact()
    }

    /// Retrieve the cookies contained in the [`Response`].
    ///
    /// Note that invalid 'Set-Cookie' headers will be ignored.
    ///
    /// # Optional
    ///
    /// This requires the optional `cookies` feature to be enabled.
    #[cfg(feature = "cookies")]
    pub fn cookies(&self) -> impl Iterator<Item = cookie::Cookie<'_>> {
        self.res
            .headers()
            .get_all(crate::header::SET_COOKIE)
            .iter()
            .map(cookie::Cookie::parse)
            .filter_map(Result::ok)
    }

    /// Get the local address used to get this [`Response`].
    pub fn local_addr(&self) -> Option<SocketAddr> {
        self.res
            .extensions()
            .get::<HttpInfo>()
            .map(HttpInfo::local_addr)
    }

    /// Get the remote address used to get this [`Response`].
    pub fn remote_addr(&self) -> Option<SocketAddr> {
        self.res
            .extensions()
            .get::<HttpInfo>()
            .map(HttpInfo::remote_addr)
    }

    // body methods

    /// Get the full response text.
    ///
    /// This method decodes the response body with BOM sniffing
    /// and with malformed sequences replaced with the [`char::REPLACEMENT_CHARACTER`].
    /// Encoding is determined from the `charset` parameter of `Content-Type` header,
    /// and defaults to `utf-8` if not presented.
    ///
    /// Note that the BOM is stripped from the returned String.
    ///
    /// # Note
    ///
    /// If the `charset` feature is disabled the method will only attempt to decode the
    /// response as UTF-8, regardless of the given `Content-Type`
    ///
    /// # Example
    ///
    /// ```
    /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
    /// let content = hpx::Client::new()
    ///     .get("http://httpbin.org/range/26")
    ///     .send()
    ///     .await?
    ///     .text()
    ///     .await?;
    ///
    /// println!("text: {content:?}");
    /// # Ok(())
    /// # }
    /// ```
    pub async fn text(self) -> crate::Result<String> {
        #[cfg(feature = "charset")]
        {
            self.text_with_charset("utf-8").await
        }

        #[cfg(not(feature = "charset"))]
        {
            let full = self.bytes().await?;
            let text = String::from_utf8_lossy(&full);
            Ok(text.into_owned())
        }
    }

    /// Get the full response text given a specific encoding.
    ///
    /// This method decodes the response body with BOM sniffing
    /// and with malformed sequences replaced with the
    /// [`char::REPLACEMENT_CHARACTER`].
    /// You can provide a default encoding for decoding the raw message, while the
    /// `charset` parameter of `Content-Type` header is still prioritized. For more information
    /// about the possible encoding name, please go to [`encoding_rs`] docs.
    ///
    /// Note that the BOM is stripped from the returned String.
    ///
    /// [`encoding_rs`]: https://docs.rs/encoding_rs/0.8/encoding_rs/#relationship-with-windows-code-pages
    ///
    /// # Optional
    ///
    /// This requires the optional `encoding_rs` feature enabled.
    ///
    /// # Example
    ///
    /// ```
    /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
    /// let content = hpx::Client::new()
    ///     .get("http://httpbin.org/range/26")
    ///     .send()
    ///     .await?
    ///     .text_with_charset("utf-8")
    ///     .await?;
    ///
    /// println!("text: {content:?}");
    /// # Ok(())
    /// # }
    /// ```
    #[cfg(feature = "charset")]
    #[cfg_attr(docsrs, doc(cfg(feature = "charset")))]
    pub async fn text_with_charset(
        self,
        default_encoding: impl AsRef<str>,
    ) -> crate::Result<String> {
        let content_type = self
            .headers()
            .get(crate::header::CONTENT_TYPE)
            .and_then(|value| value.to_str().ok())
            .and_then(|value| value.parse::<Mime>().ok());
        let encoding_name = content_type
            .as_ref()
            .and_then(|mime| mime.get_param("charset").map(|charset| charset.as_str()))
            .unwrap_or(default_encoding.as_ref());
        let encoding = Encoding::for_label(encoding_name.as_bytes()).unwrap_or(UTF_8);

        let full = self.bytes().await?;
        let (text, _, _) = encoding.decode(&full);
        Ok(text.into_owned())
    }

    /// Try to deserialize the response body as `application/x-www-form-urlencoded`.
    ///
    /// # Optional
    ///
    /// This requires the optional `form` feature enabled.
    ///
    /// # Examples
    ///
    /// ```
    /// # extern crate hpx;
    /// # extern crate serde;
    /// #
    /// # use hpx::Error;
    /// # use serde::Deserialize;
    /// #
    /// #[derive(Deserialize)]
    /// struct TokenResponse {
    ///     access_token: String,
    ///     expires_in: u64,
    /// }
    ///
    /// # async fn run() -> Result<(), Error> {
    /// let token = hpx::Client::new()
    ///     .get("https://example.com/oauth/token")
    ///     .send()
    ///     .await?
    ///     .form::<TokenResponse>()
    ///     .await?;
    ///
    /// println!(
    ///     "token: {} expires in {}",
    ///     token.access_token, token.expires_in
    /// );
    /// # Ok(())
    /// # }
    /// #
    /// # fn main() { }
    /// ```
    #[cfg(feature = "form")]
    #[cfg_attr(docsrs, doc(cfg(feature = "form")))]
    pub async fn form<T: DeserializeOwned>(self) -> crate::Result<T> {
        let full = self.bytes().await?;
        serde_urlencoded::from_bytes(&full).map_err(Error::decode)
    }

    /// Try to deserialize the response body as JSON.
    ///
    /// # Optional
    ///
    /// This requires the optional `json` feature enabled.
    ///
    /// # Examples
    ///
    /// ```
    /// # extern crate hpx;
    /// # extern crate serde;
    /// #
    /// # use hpx::Error;
    /// # use serde::Deserialize;
    /// #
    /// // This `derive` requires the `serde` dependency.
    /// #[derive(Deserialize)]
    /// struct Ip {
    ///     origin: String,
    /// }
    ///
    /// # async fn run() -> Result<(), Error> {
    /// let ip = hpx::Client::new()
    ///     .get("http://httpbin.org/ip")
    ///     .send()
    ///     .await?
    ///     .json::<Ip>()
    ///     .await?;
    ///
    /// println!("ip: {}", ip.origin);
    /// # Ok(())
    /// # }
    /// #
    /// # fn main() { }
    /// ```
    ///
    /// # Errors
    ///
    /// This method fails whenever the response body is not in JSON format
    /// or it cannot be properly deserialized to target type `T`. For more
    /// details please see [`serde_json::from_reader`].
    ///
    /// [`serde_json::from_reader`]: https://docs.serde.rs/serde_json/fn.from_reader.html
    #[cfg(all(feature = "json", not(feature = "simd-json")))]
    #[cfg_attr(docsrs, doc(cfg(feature = "json")))]
    pub async fn json<T: DeserializeOwned>(self) -> crate::Result<T> {
        let full = self.bytes().await?;
        serde_json::from_slice(&full).map_err(Error::decode)
    }

    /// Try to deserialize the response body as JSON using SIMD-accelerated parsing.
    ///
    /// This method uses `simd-json` for faster JSON parsing on supported platforms.
    /// It requires a mutable buffer, so the response bytes are copied to a Vec.
    ///
    /// # Optional
    ///
    /// This requires the optional `simd-json` feature enabled.
    ///
    /// # Examples
    ///
    /// ```
    /// # extern crate hpx;
    /// # extern crate serde;
    /// #
    /// # use hpx::Error;
    /// # use serde::Deserialize;
    /// #
    /// #[derive(Deserialize)]
    /// struct Ip {
    ///     origin: String,
    /// }
    ///
    /// # async fn run() -> Result<(), Error> {
    /// let ip = hpx::Client::new()
    ///     .get("http://httpbin.org/ip")
    ///     .send()
    ///     .await?
    ///     .json::<Ip>()
    ///     .await?;
    ///
    /// println!("ip: {}", ip.origin);
    /// # Ok(())
    /// # }
    /// ```
    #[cfg(feature = "simd-json")]
    #[cfg_attr(docsrs, doc(cfg(feature = "simd-json")))]
    pub async fn json<T: DeserializeOwned>(self) -> crate::Result<T> {
        let full = self.bytes().await?;
        let mut bytes_mut = simd_json_input(full);
        simd_json::from_slice(&mut bytes_mut).map_err(Error::decode)
    }

    /// Get the full response body as [`Bytes`].
    ///
    /// # Example
    ///
    /// ```
    /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
    /// let bytes = hpx::Client::new()
    ///     .get("http://httpbin.org/ip")
    ///     .send()
    ///     .await?
    ///     .bytes()
    ///     .await?;
    ///
    /// println!("bytes: {bytes:?}");
    /// # Ok(())
    /// # }
    /// ```
    pub async fn bytes(self) -> crate::Result<Bytes> {
        BodyExt::collect(self.res.into_body())
            .await
            .map(Collected::<Bytes>::to_bytes)
    }

    /// Stream a chunk of the response body.
    ///
    /// When the response body has been exhausted, this will return `None`.
    ///
    /// # Example
    ///
    /// ```
    /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
    /// let mut res = hpx::get("https://hyper.rs").send().await?;
    ///
    /// while let Some(chunk) = res.chunk().await? {
    ///     println!("Chunk: {chunk:?}");
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub async fn chunk(&mut self) -> crate::Result<Option<Bytes>> {
        loop {
            if let Some(res) = self.res.body_mut().frame().await {
                if let Ok(buf) = res?.into_data() {
                    return Ok(Some(buf));
                }
            } else {
                return Ok(None);
            }
        }
    }

    /// Convert the response into a [`Stream`] of [`Bytes`] from the body.
    ///
    /// # Example
    ///
    /// ```
    /// use futures_util::StreamExt;
    ///
    /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
    /// let mut stream = hpx::Client::new()
    ///     .get("http://httpbin.org/ip")
    ///     .send()
    ///     .await?
    ///     .bytes_stream();
    ///
    /// while let Some(item) = stream.next().await {
    ///     println!("Chunk: {:?}", item?);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Optional
    ///
    /// This requires the optional `stream` feature to be enabled.
    #[cfg(feature = "stream")]
    #[cfg_attr(docsrs, doc(cfg(feature = "stream")))]
    pub fn bytes_stream(self) -> impl Stream<Item = crate::Result<Bytes>> {
        http_body_util::BodyDataStream::new(self.res.into_body())
    }

    /// Consume the response and return the raw [`Body`].
    ///
    /// This is useful for zero-copy streaming scenarios where you want
    /// direct access to the body without additional processing.
    ///
    /// # Example
    ///
    /// ```
    /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
    /// let response = hpx::get("https://httpbin.org/bytes/1024").send().await?;
    /// let body = response.into_body();
    /// // Use body directly for zero-copy streaming
    /// # Ok(())
    /// # }
    /// ```
    pub fn into_body(self) -> Body {
        self.res.into_body()
    }

    /// Get an [`tokio::io::AsyncRead`] adapter for the response body.
    ///
    /// This enables efficient streaming of the response body using
    /// standard async I/O traits.
    ///
    /// # Example
    ///
    /// ```
    /// use tokio::io::AsyncReadExt;
    ///
    /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
    /// let response = hpx::get("https://httpbin.org/bytes/1024").send().await?;
    /// let mut reader = response.reader();
    ///
    /// let mut buffer = Vec::new();
    /// reader.read_to_end(&mut buffer).await?;
    /// println!("Read {} bytes", buffer.len());
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Optional
    ///
    /// This requires the optional `stream` feature to be enabled.
    #[cfg(feature = "stream")]
    #[cfg_attr(docsrs, doc(cfg(feature = "stream")))]
    pub fn reader(self) -> impl tokio::io::AsyncRead {
        tokio_util::io::StreamReader::new(
            http_body_util::BodyDataStream::new(self.res.into_body())
                .map_err(std::io::Error::other),
        )
    }

    // extension methods

    /// Returns a reference to the associated extensions.
    ///
    /// # Example
    ///
    /// ```
    /// # use hpx::{Client, tls::TlsInfo};
    /// # async fn run() -> hpx::Result<()> {
    /// // Build a client that records TLS information.
    /// let client = Client::builder().tls_info(true).build()?;
    ///
    /// // Make a request.
    /// let resp = client.get("https://www.google.com").send().await?;
    ///
    /// // Take the TlsInfo extension to inspect it.
    /// if let Some(tls_info) = resp.extensions().get::<TlsInfo>() {
    ///     // Now you own the TlsInfo and can process it.
    ///     println!("Peer certificate: {:?}", tls_info.peer_certificate());
    /// }
    ///
    /// # Ok(())
    /// # }
    /// ```
    #[inline]
    pub fn extensions(&self) -> &http::Extensions {
        self.res.extensions()
    }

    /// Returns a mutable reference to the associated extensions.
    ///
    /// # Example
    ///
    /// ```
    /// # use hpx::{Client, tls::TlsInfo};
    /// # async fn run() -> hpx::Result<()> {
    /// // Build a client that records TLS information.
    /// let client = Client::builder().tls_info(true).build()?;
    ///
    /// // Make a request.
    /// let mut resp = client.get("https://www.google.com").send().await?;
    ///
    /// // Take the TlsInfo extension to inspect it.
    /// if let Some(tls_info) = resp.extensions_mut().remove::<TlsInfo>() {
    ///     // Now you own the TlsInfo and can process it.
    ///     println!("Peer certificate: {:?}", tls_info.peer_certificate());
    /// }
    ///
    /// # Ok(())
    /// # }
    /// ```
    #[inline]
    pub fn extensions_mut(&mut self) -> &mut http::Extensions {
        self.res.extensions_mut()
    }

    // util methods

    /// Turn a response into an error if the server returned an error.
    ///
    /// # Example
    ///
    /// ```
    /// # use hpx::Response;
    /// fn on_response(res: Response) {
    ///     match res.error_for_status() {
    ///         Ok(_res) => (),
    ///         Err(err) => {
    ///             // asserting a 400 as an example
    ///             // it could be any status between 400...599
    ///             assert_eq!(err.status(), Some(hpx::StatusCode::BAD_REQUEST));
    ///         }
    ///     }
    /// }
    /// # fn main() {}
    /// ```
    pub fn error_for_status(mut self) -> crate::Result<Self> {
        let status = self.status();
        if status.is_client_error() || status.is_server_error() {
            let reason = self.res.extensions_mut().remove::<ReasonPhrase>();
            Err(Error::status_code(self.uri, status, reason))
        } else {
            Ok(self)
        }
    }

    /// Turn a reference to a response into an error if the server returned an error.
    ///
    /// # Example
    ///
    /// ```
    /// # use hpx::Response;
    /// fn on_response(res: &Response) {
    ///     match res.error_for_status_ref() {
    ///         Ok(_res) => (),
    ///         Err(err) => {
    ///             // asserting a 400 as an example
    ///             // it could be any status between 400...599
    ///             assert_eq!(err.status(), Some(hpx::StatusCode::BAD_REQUEST));
    ///         }
    ///     }
    /// }
    /// # fn main() {}
    /// ```
    pub fn error_for_status_ref(&self) -> crate::Result<&Self> {
        let status = self.status();
        if status.is_client_error() || status.is_server_error() {
            let reason = self.res.extensions().get::<ReasonPhrase>().cloned();
            Err(Error::status_code(self.uri.clone(), status, reason))
        } else {
            Ok(self)
        }
    }

    /// Consumes the [`Response`] and returns a future for a possible HTTP upgrade.
    pub async fn upgrade(self) -> crate::Result<Upgraded> {
        upgrade::on(self.res).await.map_err(Error::upgrade)
    }
}

/// I'm not sure this conversion is that useful... People should be encouraged
/// to use [`http::Response`], not `hpx::Response`.
impl<T: Into<Body>> From<http::Response<T>> for Response {
    fn from(r: http::Response<T>) -> Response {
        let mut res = r.map(Into::into);
        let uri = res
            .extensions_mut()
            .remove::<RequestUri>()
            .unwrap_or_else(|| RequestUri(Uri::from_static("http://no.url.provided.local")));
        Response { res, uri: uri.0 }
    }
}

/// A [`Response`] can be converted into a [`http::Response`].
// It's supposed to be the inverse of the conversion above.
impl From<Response> for http::Response<Body> {
    fn from(r: Response) -> http::Response<Body> {
        let mut res = r.res;
        res.extensions_mut().insert(RequestUri(r.uri));
        res
    }
}

/// A [`Response`] can be piped as the [`Body`] of another request.
impl From<Response> for Body {
    fn from(r: Response) -> Body {
        r.res.into_body()
    }
}

/// A [`Response`] implements [`HttpBody`] to allow streaming the body.
impl HttpBody for Response {
    type Data = Bytes;

    type Error = Error;

    #[inline]
    fn poll_frame(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
        Pin::new(self.res.body_mut()).poll_frame(cx)
    }

    #[inline]
    fn is_end_stream(&self) -> bool {
        self.res.body().is_end_stream()
    }

    #[inline]
    fn size_hint(&self) -> http_body::SizeHint {
        self.res.body().size_hint()
    }
}

#[cfg(test)]
mod tests {
    use bytes::Bytes;
    use http::Uri;
    #[cfg(feature = "form")]
    use serde::Deserialize;

    use super::Response;
    #[cfg(feature = "simd-json")]
    use super::simd_json_input;
    use crate::{Body, ClientResponseBody, ext::RequestUri};

    fn response_with_body<T: Into<Body>>(body: T) -> Response {
        let mut res = http::Response::new(body);
        res.extensions_mut()
            .insert(RequestUri(Uri::from_static("https://example.com/resource")));
        res.into()
    }

    #[cfg(feature = "form")]
    #[tokio::test]
    async fn response_form_decodes_urlencoded_body() {
        #[derive(Debug, Deserialize, PartialEq, Eq)]
        struct TokenResponse {
            access_token: String,
            expires_in: u64,
        }

        let response = response_with_body("access_token=abc123&expires_in=60");
        let token = response.form::<TokenResponse>().await.unwrap();

        assert_eq!(
            token,
            TokenResponse {
                access_token: "abc123".to_owned(),
                expires_in: 60,
            }
        );
    }

    #[test]
    fn response_into_body_preserves_reusable_bytes() {
        let response = response_with_body(Bytes::from_static(b"hello"));
        let body: Body = response.into();

        assert_eq!(body.as_bytes(), Some(&b"hello"[..]));
    }

    #[test]
    fn response_into_http_preserves_reusable_bytes_and_uri() {
        let response = response_with_body(Bytes::from_static(b"hello"));
        let http_response: http::Response<Body> = response.into();

        assert_eq!(http_response.body().as_bytes(), Some(&b"hello"[..]));
        assert_eq!(
            http_response
                .extensions()
                .get::<RequestUri>()
                .map(|uri| &uri.0),
            Some(&Uri::from_static("https://example.com/resource"))
        );
    }

    #[test]
    fn response_from_client_response_preserves_direct_body_access() {
        let response = Response::from_client_response(
            Uri::from_static("https://example.com/resource"),
            http::Response::new(ClientResponseBody::from(Bytes::from_static(b"hello"))),
        );

        let body: Body = response.into();
        assert_eq!(body.as_bytes(), Some(&b"hello"[..]));
    }

    #[cfg(feature = "simd-json")]
    #[test]
    fn simd_json_input_reuses_unique_bytes_storage() {
        let bytes = Bytes::from(Vec::from(&b"{\"ok\":true}"[..]));
        let ptr = bytes.as_ptr();

        let buffer = simd_json_input(bytes);

        assert_eq!(buffer.as_ptr(), ptr);
    }

    #[cfg(feature = "simd-json")]
    #[test]
    fn simd_json_input_copies_when_bytes_are_shared() {
        let bytes = Bytes::from(Vec::from(&b"{\"ok\":true}"[..]));
        let shared = bytes.clone();
        let ptr = shared.as_ptr();

        let buffer = simd_json_input(shared);

        assert_eq!(buffer.as_ref(), bytes.as_ref());
        assert_ne!(buffer.as_ptr(), ptr);
    }
}