5fb6d033-8551-4bbf-82b0-80b1bfd2cd99

Committed 28 Apr 2026 07:15AM UTC coverage: 99.694% (+5.2%) from 94.53%

Build # 5fb6d033-8551-4bbf-82b0-80b1bfd2cd99

Build Type

push

circleci

Committed by

Haixing-Hu

Commit Message

chore(ci): sync shared Rust CI scripts

Coverage Stats

3259 of 3269 relevant lines covered (99.69%)

64.59 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.69

/src/client/http_client.rs

/*******************************************************************************
 *
 *    Copyright (c) 2025 - 2026.
 *    Haixing Hu, Qubit Co. Ltd.
 *
 *    All rights reserved.
 *
 ******************************************************************************/
//! HTTP client: builds requests, applies defaults and interceptors, executes
//! them with optional retry, and exposes SSE helpers with reconnect.
//!
//! Single-shot execution is [`HttpClient::execute`] / [`HttpClient::execute_once`];
//! retry policy comes from [`crate::HttpClientOptions::retry`] unless overridden
//! per request.
//!
//! # Author
//!
//! Haixing Hu

use std::time::{Duration, Instant};

use qubit_retry::{
    AttemptFailure, AttemptFailureDecision, Retry, RetryContext, RetryError, RetryErrorReason,
};

use crate::sse::SseReconnectRunner;
use crate::{
    response::HttpResponseOptions,
    sse::{SseEventStream, SseReconnectOptions},
    AsyncHttpHeaderInjector, HttpClientOptions, HttpError, HttpHeaderInjector, HttpLogger,
    HttpRequest, HttpRequestBuilder, HttpRequestInterceptor, HttpRequestInterceptors, HttpResponse,
    HttpResponseInterceptor, HttpResponseInterceptors, HttpResponseMeta, HttpResult,
    HttpRetryOptions,
};

/// High-level HTTP client: default headers, injectors, interceptors, logging,
/// timeouts, and optional per-request retry.
///
/// [`Clone`] is shallow and cheap enough for typical use (including passing into
/// retry closures); cloning does not duplicate the underlying connection pool
/// beyond what [`reqwest::Client`] already shares.
#[derive(Clone)]
pub struct HttpClient {
    /// Pluggable low-level HTTP stack used to send requests (currently reqwest).
    pub(super) backend: reqwest::Client,
    /// Timeouts, proxy, logging, default headers, and related settings.
    pub(super) options: HttpClientOptions,
    /// Header injectors applied to every outgoing request after default
    /// headers.
    pub(super) injectors: Vec<HttpHeaderInjector>,
    /// Async header injectors applied after sync injectors and before request-level headers.
    pub(super) async_injectors: Vec<AsyncHttpHeaderInjector>,
    /// Request interceptors applied before request send for each attempt.
    request_interceptors: HttpRequestInterceptors,
    /// Response interceptors applied on successful responses before return.
    response_interceptors: HttpResponseInterceptors,
}

impl HttpClient {
    /// Wraps a built [`reqwest::Client`] with the given options and an empty
    /// injector list.
    ///
    /// # Parameters
    /// - `backend`: Configured low-level HTTP client used for I/O.
    /// - `options`: Client-wide timeouts, headers, proxy, logging, etc.
    ///
    /// # Returns
    /// A new [`HttpClient`] with no injectors until
    /// [`HttpClient::add_header_injector`] is called.
    pub(crate) fn new(backend: reqwest::Client, options: HttpClientOptions) -> Self {
        Self {
            backend,
            options,
            injectors: Vec::new(),
            async_injectors: Vec::new(),
            request_interceptors: HttpRequestInterceptors::new(),
            response_interceptors: HttpResponseInterceptors::new(),
        }
    }

    /// Returns a reference to the client-wide options (timeouts, proxy, logging,
    /// default headers, retry defaults, etc.).
    ///
    /// # Returns
    /// Immutable borrow of [`HttpClientOptions`]. Never `None`; always the
    /// options installed on this client.
    pub fn options(&self) -> &HttpClientOptions {
        &self.options
    }

    /// Appends a [`HttpHeaderInjector`] so its mutation function runs on every
    /// request. Mutates `self` in place.
    ///
    /// # Parameters
    /// - `injector`: Injector to append (order is preserved).
    pub fn add_header_injector(&mut self, injector: HttpHeaderInjector) {
        self.injectors.push(injector);
    }

    /// Appends an async header injector whose mutation runs after sync injectors.
    /// Mutates `self` in place.
    ///
    /// # Parameters
    /// - `injector`: Async injector to append (order is preserved).
    pub fn add_async_header_injector(&mut self, injector: AsyncHttpHeaderInjector) {
        self.async_injectors.push(injector);
    }

    /// Appends a request interceptor run before each send attempt (including
    /// each retry attempt). Mutates `self` in place.
    ///
    /// # Parameters
    /// - `interceptor`: Request interceptor to append (order is preserved).
    pub fn add_request_interceptor(&mut self, interceptor: HttpRequestInterceptor) {
        self.request_interceptors.push(interceptor);
    }

    /// Appends a response interceptor run only after a successful HTTP status
    /// (after the internal `execute_once` step) and before response body logging.
    /// Mutates `self` in place.
    ///
    /// # Parameters
    /// - `interceptor`: Response interceptor to append (order is preserved).
    pub fn add_response_interceptor(&mut self, interceptor: HttpResponseInterceptor) {
        self.response_interceptors.push(interceptor);
    }

    /// Validates and adds one client-level default header.
    ///
    /// The header is applied to every request before header injectors and
    /// request-level headers.
    ///
    /// # Parameters
    /// - `name`: Header name.
    /// - `value`: Header value.
    ///
    /// # Returns
    /// `Ok(self)` after the header is stored.
    ///
    /// # Errors
    /// Returns [`HttpError`] when the header name or value is invalid.
    pub fn add_header(&mut self, name: &str, value: &str) -> HttpResult<&mut Self> {
        self.options.add_header(name, value)?;
        Ok(self)
    }

    /// Validates and adds many client-level default headers atomically.
    ///
    /// If any input pair is invalid, no header from this batch is applied.
    ///
    /// # Parameters
    /// - `headers`: Iterator of `(name, value)` pairs.
    ///
    /// # Returns
    /// `Ok(self)` after all headers are stored.
    ///
    /// # Errors
    /// Returns [`HttpError`] when any name/value pair is invalid (nothing from
    /// this call is applied).
    pub fn add_headers(&mut self, headers: &[(&str, &str)]) -> HttpResult<&mut Self> {
        self.options.add_headers(headers)?;
        Ok(self)
    }

    /// Clears all synchronous header injectors. Mutates `self` in place.
    pub fn clear_header_injectors(&mut self) {
        self.injectors.clear();
    }

    /// Clears all async header injectors. Mutates `self` in place.
    pub fn clear_async_header_injectors(&mut self) {
        self.async_injectors.clear();
    }

    /// Clears all request interceptors. Mutates `self` in place.
    pub fn clear_request_interceptors(&mut self) {
        self.request_interceptors.clear();
    }

    /// Clears all response interceptors. Mutates `self` in place.
    pub fn clear_response_interceptors(&mut self) {
        self.response_interceptors.clear();
    }

    /// Starts building an [`HttpRequest`] with the given method and path
    /// (relative or absolute URL string).
    ///
    /// # Parameters
    /// - `method`: HTTP verb (GET, POST, …).
    /// - `path`: Path relative to [`HttpClientOptions::base_url`] or a full URL
    ///   string.
    ///
    /// # Returns
    /// A new [`HttpRequestBuilder`] borrowing this client for defaults; it is
    /// not sent until built and passed to [`HttpClient::execute`] (or related
    /// APIs).
    pub fn request(&self, method: http::Method, path: &str) -> HttpRequestBuilder {
        HttpRequestBuilder::new(method, path, self)
    }

    /// Returns a clone of the client-level default header map.
    ///
    /// Used when constructing a built [`HttpRequest`] so the snapshot reflects
    /// headers at build time.
    ///
    /// # Returns
    /// Owned [`http::HeaderMap`] copy of [`HttpClientOptions`] default headers.
    pub(crate) fn headers_snapshot(&self) -> http::HeaderMap {
        self.options.default_headers.clone()
    }

    /// Returns a clone of the registered synchronous header injectors list.
    ///
    /// # Returns
    /// New [`Vec`] with the same injectors and order as on this client.
    pub(crate) fn injectors_snapshot(&self) -> Vec<HttpHeaderInjector> {
        self.injectors.clone()
    }

    /// Returns a clone of the registered async header injectors list.
    ///
    /// # Returns
    /// New [`Vec`] with the same injectors and order as on this client.
    pub(crate) fn async_injectors_snapshot(&self) -> Vec<AsyncHttpHeaderInjector> {
        self.async_injectors.clone()
    }

    /// Sends the request and returns a unified [`HttpResponse`].
    ///
    /// Chooses retry vs single attempt from resolved [`HttpRetryOptions`] for
    /// this request. Performs network I/O and may await the internal
    /// `execute_once` path
    /// multiple times with backoff between attempts when retry is enabled.
    ///
    /// # Parameters
    /// - `request`: Built request (URL resolved against `base_url` if path is
    ///   not absolute).
    ///
    /// # Returns
    /// - `Ok(HttpResponse)` when the HTTP status is success
    ///   ([`http::StatusCode::is_success`]).
    /// - `Err(HttpError)` when any attempt fails for URL/header validation,
    ///   cancellation, interceptor failure, transport/timeout, non-success
    ///   status, or when the retry executor aborts or exceeds limits.
    pub async fn execute(&self, request: HttpRequest) -> HttpResult<HttpResponse> {
        let retry_options = self.options.retry.resolve(&request);
        if retry_options.should_retry(&request) {
            self.execute_with_retry(request, retry_options).await
        } else {
            self.execute_once(request).await
        }
    }

    /// Performs one non-retrying execution: pre-send cancellation check,
    /// request interceptors, resolve URL, merge headers, log the request, send
    /// with configured timeouts, map non-success status to an error, then
    /// response interceptors and response logging. The returned body is read
    /// lazily according to [`HttpResponse`].
    ///
    /// # Parameters
    /// - `request`: Built request to send (same fields as for
    ///   [`HttpClient::execute`]).
    ///
    /// # Returns
    /// - `Ok(HttpResponse)` on success status and after interceptors/logging
    ///   steps succeed.
    /// - `Err(HttpError)` from request/response interceptors, cancellation,
    ///   send/transport errors, status mapping, URL resolution for the response
    ///   wrapper, or response logging failures.
    ///
    /// # Side effects
    /// Network I/O, optional logging, and user-provided interceptor callbacks.
    pub(crate) async fn execute_once(&self, request: HttpRequest) -> HttpResult<HttpResponse> {
        let mut request = request;
        if let Some(error) = request.cancelled_error_if_needed("Request cancelled before sending") {
            return Err(error);
        }
        self.request_interceptors.apply(&mut request)?;
        let response = self
            .prepare_and_send_once(request, "Request cancelled before sending")
            .await?;
        let mut response = response
            .into_success_or_status_error("HTTP request failed")
            .await?;
        self.response_interceptors.apply(&mut response.meta)?;
        let logger = HttpLogger::new(&self.options);
        logger.log_response(&mut response).await?;
        Ok(response)
    }

    /// Single low-level send: cancellation check, request logging, one backend
    /// round-trip, then wraps the backend response as [`HttpResponse`].
    ///
    /// Does not run response interceptors or success-status enforcement; those
    /// happen in [`HttpClient::execute_once`] after this returns.
    ///
    /// # Parameters
    /// - `request`: Request to send (may be mutated for logging/send path).
    /// - `cancellation_message`: Message embedded if the request is already
    ///   cancelled when this runs.
    ///
    /// # Returns
    /// - `Ok(HttpResponse)` with lazy body and metadata.
    /// - `Err(HttpError)` if cancelled before send, send fails, or
    ///   [`HttpRequest::resolved_url`] fails when building the wrapper.
    ///
    /// # Side effects
    /// Async network I/O and request logging via [`HttpLogger`].
    async fn prepare_and_send_once(
        &self,
        request: HttpRequest,
        cancellation_message: &str,
    ) -> HttpResult<HttpResponse> {
        let mut request = request;
        if let Some(error) = request.cancelled_error_if_needed(cancellation_message) {
            return Err(error);
        }
        let logger = HttpLogger::new(&self.options);
        let backend_response = request.send_impl(&self.backend, &logger).await?;
        let meta = HttpResponseMeta::new(
            backend_response.status(),
            backend_response.headers().clone(),
            backend_response.url().clone(),
            request.method().clone(),
        );
        let response_options = HttpResponseOptions::new(
            self.options.error_response_preview_limit,
            self.options.sse_json_mode,
            self.options.sse_max_line_bytes,
            self.options.sse_max_frame_bytes,
            self.options.sse_done_marker_policy.clone(),
        );
        Ok(HttpResponse::from_backend(
            meta,
            backend_response,
            request.read_timeout(),
            request.cancellation_token().cloned(),
            request.resolved_url_with_query()?,
            response_options,
        ))
    }

    /// Runs [`HttpClient::execute_once`] under the given retry policy.
    ///
    /// Between attempts waits according to the resolved retry delay, optionally
    /// honoring `Retry-After` by extending the next sleep. Each attempt clones
    /// the request so request bodies can be rebuilt when supported.
    ///
    /// # Parameters
    /// - `request`: Built request passed to each [`HttpClient::execute_once`]
    ///   attempt (cloned per retry closure).
    /// - `options`: Effective retry options for this request (from resolution
    ///   in [`HttpClient::execute`]).
    ///
    /// # Returns
    /// - `Ok(HttpResponse)` when an attempt completes with success status.
    /// - `Err(HttpError)` from retry option validation, from any
    ///   [`HttpClient::execute_once`] failure that is non-retryable, or from
    ///   retry exhaustion/max-duration enforcement.
    ///
    /// # Side effects
    /// Multiple async HTTP attempts and optional sleeps.
    async fn execute_with_retry(
        &self,
        request: HttpRequest,
        options: HttpRetryOptions,
    ) -> HttpResult<HttpResponse> {
        let honor_retry_after = request.retry_override().should_honor_retry_after();
        let retry_options = options.to_executor_options()?;
        let started_at = Instant::now();

        let retry_policy_options = options.clone();
        let retry_delay_options = retry_options.clone();
        let retry_policy = match Retry::<HttpError>::builder()
            .options(retry_options)
            .retry_after_from_error(move |error| {
                honor_retry_after.then_some(error.retry_after).flatten()
            })
            .on_failure(
                move |failure: &AttemptFailure<HttpError>, context: &RetryContext| {
                    Self::retry_failure_decision(
                        failure,
                        context,
                        &retry_policy_options,
                        &retry_delay_options,
                    )
                },
            )
            .build()
        {
            Ok(retry_policy) => retry_policy,
            Err(error) => {
                return Err(HttpError::other(format!(
                    "Invalid HTTP retry executor: {error}"
                )))
            }
        };

        let cancellation_token = request.cancellation_token().cloned();
        let request_method = request.method().clone();
        let request_url = request.resolved_url_with_query().ok();
        let retry_request = request.clone();
        let retry_future = retry_policy.run_async(|| {
            let attempt_request = retry_request.clone();
            async move { self.execute_once(attempt_request).await }
        });

        let retry_result = if let Some(token) = cancellation_token.as_ref() {
            tokio::select! {
                _ = token.cancelled() => {
                    return Err(Self::retry_cancelled_error(
                        "HTTP retry cancelled while waiting before next attempt",
                        &request_method,
                        request_url.as_ref(),
                    ));
                }
                result = retry_future => result,
            }
        } else {
            retry_future.await
        };

        match retry_result {
            Ok(response) => Ok(response),
            Err(error) => Err(Self::map_retry_error(
                error,
                started_at,
                options.max_duration,
                options.max_attempts,
            )),
        }
    }

    /// Returns whether `error` is retryable under `options`.
    ///
    /// # Parameters
    /// - `error`: Error produced by a single HTTP attempt.
    /// - `options`: Effective retry options for the request.
    ///
    /// # Returns
    /// `true` if another attempt may be scheduled.
    fn is_retryable_error(error: &HttpError, options: &HttpRetryOptions) -> bool {
        if error.kind == crate::HttpErrorKind::Status {
            error
                .status
                .is_some_and(|status| options.is_retryable_status(status))
        } else {
            options.is_retryable_error_kind(error.kind)
        }
    }

    /// Computes the sleep before the next retry attempt.
    ///
    /// # Parameters
    /// - `base_delay`: Delay selected from retry policy and jitter.
    /// - `retry_after_hint`: Retry-After delay extracted by the retry policy.
    ///
    /// # Returns
    /// `base_delay`, or the larger `Retry-After` value when present.
    fn retry_sleep_delay(base_delay: Duration, retry_after_hint: Option<Duration>) -> Duration {
        retry_after_hint
            .map(|retry_after| retry_after.max(base_delay))
            .unwrap_or(base_delay)
    }

    /// Decides how HTTP retry should handle one failed attempt.
    ///
    /// # Parameters
    /// - `failure`: Failed attempt reported by `qubit-retry`.
    /// - `context`: Retry context for the failed attempt.
    /// - `policy_options`: HTTP retry allowlists and method policy.
    /// - `delay_options`: Retry executor options used to calculate base delay.
    ///
    /// # Returns
    /// Decision for `qubit-retry`: abort non-retryable HTTP errors, otherwise
    /// retry after the larger base delay / Retry-After hint. Non-HTTP runtime
    /// failures fall back to the retry executor default.
    fn retry_failure_decision(
        failure: &AttemptFailure<HttpError>,
        context: &RetryContext,
        policy_options: &HttpRetryOptions,
        delay_options: &qubit_retry::RetryOptions,
    ) -> AttemptFailureDecision {
        let Some(error) = failure.as_error() else {
            return AttemptFailureDecision::UseDefault;
        };
        if !Self::is_retryable_error(error, policy_options) {
            return AttemptFailureDecision::Abort;
        }

        let base_delay = delay_options.delay_for_attempt(context.attempt());
        let sleep_delay = Self::retry_sleep_delay(base_delay, context.retry_after_hint());
        AttemptFailureDecision::RetryAfter(sleep_delay)
    }

    /// Exposes retry failure decisions to coverage-only integration tests.
    ///
    /// # Parameters
    /// - `failure`: Failed attempt to inspect.
    /// - `context`: Retry context for the failed attempt.
    /// - `policy_options`: HTTP retry policy options.
    /// - `delay_options`: Retry executor options.
    ///
    /// # Returns
    /// The retry decision selected by [`Self::retry_failure_decision`].
    #[cfg(coverage)]
    #[doc(hidden)]
    pub(crate) fn coverage_retry_failure_decision(
        failure: &AttemptFailure<HttpError>,
        context: &RetryContext,
        policy_options: &HttpRetryOptions,
        delay_options: &qubit_retry::RetryOptions,
    ) -> AttemptFailureDecision {
        Self::retry_failure_decision(failure, context, policy_options, delay_options)
    }

    /// Adds retry-attempt exhaustion context to the last attempt error.
    ///
    /// # Parameters
    /// - `error`: Last retryable attempt error.
    /// - `attempts`: Number of attempts already made.
    /// - `max_attempts`: Configured maximum attempts.
    ///
    /// # Returns
    /// The same error with retry exhaustion details appended to its message.
    fn map_retry_attempts_exhausted(
        mut error: HttpError,
        attempts: u32,
        max_attempts: u32,
    ) -> HttpError {
        error.message = format!(
            "{} (retry attempts exhausted: {attempts}/{max_attempts})",
            error.message
        );
        error
    }

    /// Builds the error returned when retry policy stops early.
    ///
    /// # Parameters
    /// - `error`: Attempt error that the retry policy chose not to retry.
    /// - `attempts`: Number of attempts already made.
    /// - `started_at`: Start instant of the retry flow.
    ///
    /// # Returns
    /// [`HttpError::retry_aborted`] with the original [`HttpError`] chained as
    /// source for callers that need the underlying status or transport error.
    fn map_retry_aborted(error: HttpError, attempts: u32, started_at: Instant) -> HttpError {
        let elapsed = started_at.elapsed();
        let summary = error.message.clone();
        HttpError::retry_aborted(format!(
            "HTTP retry aborted after {attempts} attempt(s) in {elapsed:?}: {summary}"
        ))
        .with_source(error)
    }

    /// Builds the error when retry max-duration is exhausted.
    ///
    /// # Parameters
    /// - `started_at`: Start instant of the retry flow.
    /// - `max_duration`: Configured max-duration budget.
    /// - `last_error`: Last captured retryable attempt error, if any.
    ///
    /// # Returns
    /// Augments the last failure when present, otherwise a dedicated
    /// max-duration error with no underlying attempt error.
    fn map_retry_max_duration_exceeded(
        started_at: Instant,
        max_duration: Option<Duration>,
        last_error: Option<HttpError>,
    ) -> HttpError {
        let elapsed = started_at.elapsed();
        let max_duration_text = max_duration
            .map(|duration| format!("{duration:?}"))
            .unwrap_or_else(|| "unbounded".to_string());
        match last_error {
            Some(mut error) => {
                error.message = format!(
                    "{} (retry max duration exceeded: {elapsed:?}/{max_duration_text})",
                    error.message
                );
                error
            }
            None => HttpError::retry_max_elapsed_exceeded(format!(
                "HTTP retry max duration exceeded before a retryable error was captured: {elapsed:?}/{max_duration_text}"
            )),
        }
    }

    /// Maps a [`qubit_retry::RetryError`] into this crate's HTTP error model.
    ///
    /// # Parameters
    /// - `error`: Terminal retry error from `qubit-retry`.
    /// - `started_at`: Monotonic start instant of the HTTP retry flow.
    /// - `max_duration`: Optional HTTP total retry budget.
    /// - `max_attempts`: Configured maximum HTTP attempts.
    ///
    /// # Returns
    /// A rich [`HttpError`] preserving the last attempt error when available.
    fn map_retry_error(
        error: RetryError<HttpError>,
        started_at: Instant,
        max_duration: Option<Duration>,
        max_attempts: u32,
    ) -> HttpError {
        let attempts = error.attempts();
        let reason = error.reason();
        let (_, last_failure, _) = error.into_parts();
        let last_error = last_failure.and_then(AttemptFailure::into_error);

        match reason {
            RetryErrorReason::AttemptsExceeded => last_error
                .map(|error| Self::map_retry_attempts_exhausted(error, attempts, max_attempts))
                .unwrap_or_else(|| {
                    HttpError::retry_aborted(format!(
                        "HTTP retry attempts exhausted without a captured HTTP error: {attempts}/{max_attempts}"
                    ))
                }),
            RetryErrorReason::MaxOperationElapsedExceeded
            | RetryErrorReason::MaxTotalElapsedExceeded => {
                Self::map_retry_max_duration_exceeded(started_at, max_duration, last_error)
            }
            RetryErrorReason::Aborted => match last_error {
                Some(error) if error.kind == crate::HttpErrorKind::Cancelled => error,
                Some(error) => Self::map_retry_aborted(error, attempts, started_at),
                None => HttpError::retry_aborted(format!(
                    "HTTP retry aborted after {attempts} attempt(s) without a captured HTTP error"
                )),
            },
            RetryErrorReason::UnsupportedOperation | RetryErrorReason::WorkerStillRunning => {
                HttpError::other(format!(
                    "HTTP retry executor failed after {attempts} attempt(s): {reason:?}"
                ))
            }
        }
    }

    /// Exposes retry error mapping to coverage-only integration tests.
    ///
    /// # Parameters
    /// - `error`: Synthetic terminal retry error.
    /// - `started_at`: Monotonic retry-flow start instant.
    /// - `max_duration`: Optional HTTP max-duration budget.
    /// - `max_attempts`: Configured maximum attempts.
    ///
    /// # Returns
    /// HTTP error mapped by [`Self::map_retry_error`].
    #[cfg(coverage)]
    #[doc(hidden)]
    pub(crate) fn coverage_map_retry_error(
        error: RetryError<HttpError>,
        started_at: Instant,
        max_duration: Option<Duration>,
        max_attempts: u32,
    ) -> HttpError {
        Self::map_retry_error(error, started_at, max_duration, max_attempts)
    }

    /// Exercises the low-level pre-send cancellation check for coverage tests.
    ///
    /// # Returns
    /// Error kind returned before any network I/O starts.
    #[cfg(coverage)]
    #[doc(hidden)]
    pub(crate) async fn coverage_prepare_cancelled_error() -> crate::HttpErrorKind {
        let client = crate::HttpClientFactory::new()
            .create_default()
            .expect("coverage HTTP client should build");
        let token = tokio_util::sync::CancellationToken::new();
        token.cancel();
        let mut request = client
            .request(
                http::Method::GET,
                "https://example.com/cancelled-before-send",
            )
            .build();
        request.set_cancellation_token(token);

        client
            .prepare_and_send_once(request, "coverage request cancelled before send")
            .await
            .expect_err("pre-cancelled request should fail before send")
            .kind
    }

    /// Builds a cancellation error for retry wait cancellation.
    ///
    /// # Parameters
    /// - `message`: Human-readable cancellation reason.
    /// - `method`: Request method to attach.
    /// - `url`: Optional resolved request URL to attach.
    ///
    /// # Returns
    /// [`HttpErrorKind::Cancelled`](crate::HttpErrorKind::Cancelled) with request
    /// context.
    fn retry_cancelled_error(
        message: &str,
        method: &http::Method,
        url: Option<&url::Url>,
    ) -> HttpError {
        let mut error = HttpError::cancelled(message).with_method(method);
        if let Some(url) = url {
            error = error.with_url(url);
        }
        error
    }

    /// Opens an SSE stream and reconnects automatically on retryable stream
    /// failures.
    ///
    /// Reconnect behavior:
    /// - retryable transport/read failures trigger reconnects;
    /// - optional reconnect on clean EOF (`reconnect_on_eof`);
    /// - `Last-Event-ID` is set from the latest parsed SSE `id:` field;
    /// - optional use of SSE `retry:` as next reconnect delay.
    ///
    /// # Parameters
    /// - `request`: SSE request template reused on reconnect.
    /// - `options`: Reconnect limits and delay policy.
    ///
    /// # Returns
    /// SSE event stream yielding events from one or more reconnect sessions.
    ///
    /// # Errors
    /// Stream items are `Result`; `Err` covers per-item failures such as:
    /// - initial stream-open failures when not reconnectable or retries exhausted;
    /// - SSE protocol errors (non-reconnectable by default);
    /// - transport/read errors after reconnect budget is exhausted.
    ///
    /// # Side effects
    /// Performs repeated HTTP requests and reads on reconnect; may sleep between
    /// attempts according to reconnect options.
    pub fn execute_sse_with_reconnect(
        &self,
        request: HttpRequest,
        options: SseReconnectOptions,
    ) -> SseEventStream {
        SseReconnectRunner::new(self.clone(), request, options).run()
    }
}

impl std::fmt::Debug for HttpClient {
    /// Formats the client for debugging (exposes options and injectors; omits
    /// the backend client).
    ///
    /// # Parameters
    /// - `f`: Destination formatter.
    ///
    /// # Returns
    /// `fmt::Result` from writing the debug struct.
    ///
    /// # Errors
    /// Returns an error if formatting to `f` fails.
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("HttpClient")
            .field("options", &self.options)
            .field("injectors", &self.injectors)
            .field("async_injectors", &self.async_injectors)
            .field("request_interceptors", &self.request_interceptors)
            .field("response_interceptors", &self.response_interceptors)
            .finish_non_exhaustive()
    }
}

qubit-ltd / rs-http / 5fb6d033-8551-4bbf-82b0-80b1bfd2cd99

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous