qubit_http/response/
http_response.rs

1/*******************************************************************************
2 *
3 *    Copyright (c) 2025 - 2026 Haixing Hu.
4 *
5 *    SPDX-License-Identifier: Apache-2.0
6 *
7 *    Licensed under the Apache License, Version 2.0.
8 *
9 ******************************************************************************/
10//! Unified HTTP response type and helpers.
11
12use std::fmt;
13use std::sync::{
14    Arc,
15    Mutex,
16};
17use std::time::Duration;
18
19use async_stream::stream;
20use bytes::Bytes;
21use futures_util::{
22    stream as futures_stream,
23    StreamExt,
24};
25use http::header::{
26    CONTENT_LENGTH,
27    CONTENT_TYPE,
28};
29use http::{
30    HeaderMap,
31    Method,
32    StatusCode,
33};
34use serde::de::DeserializeOwned;
35use tokio_util::sync::CancellationToken;
36use url::Url;
37
38use crate::content_type;
39use crate::error::{
40    backend_error_mapper::map_reqwest_error,
41    ReqwestErrorPhase,
42};
43use crate::sanitize::{
44    BodyLogContext,
45    BodyPreview,
46    LogSanitizer,
47    SanitizedDebugger,
48};
49use crate::sse::{
50    DoneMarkerPolicy,
51    SseChunkStream,
52    SseJsonMode,
53    SseMessageStream,
54};
55use crate::{
56    HttpByteStream,
57    HttpError,
58    HttpErrorKind,
59    HttpResult,
60};
61
62use super::{
63    HttpResponseMeta,
64    HttpResponseOptions,
65};
66
67/// Snapshot of a body read failure retained after the backend body is consumed.
68#[derive(Debug, Clone)]
69struct BodyReadFailure {
70    /// Error category from the original failure.
71    kind: HttpErrorKind,
72    /// Original failure message.
73    message: String,
74    /// Optional request method context.
75    method: Option<Method>,
76    /// Optional request URL context.
77    url: Option<Url>,
78    /// Optional HTTP status context.
79    status: Option<StatusCode>,
80}
81
82impl BodyReadFailure {
83    /// Captures cloneable diagnostic fields from a read error.
84    ///
85    /// # Parameters
86    /// - `error`: Original body read failure.
87    ///
88    /// # Returns
89    /// Cloneable failure snapshot for subsequent reads.
90    fn from_error(error: &HttpError) -> Self {
91        Self {
92            kind: error.kind,
93            message: error.message.clone(),
94            method: error.method.clone(),
95            url: error.url.clone(),
96            status: error.status,
97        }
98    }
99
100    /// Rebuilds an [`HttpError`] for a later read attempt.
101    ///
102    /// # Returns
103    /// Error preserving the original kind and cloneable context.
104    fn to_error(&self) -> HttpError {
105        let mut error = HttpError::new(
106            self.kind,
107            format!("previous response body read failed: {}", self.message),
108        );
109        if let Some(method) = self.method.as_ref() {
110            error = error.with_method(method);
111        }
112        if let Some(url) = self.url.as_ref() {
113            error = error.with_url(url);
114        }
115        if let Some(status) = self.status {
116            error = error.with_status(status);
117        }
118        error
119    }
120}
121
122/// Shared response body failure state.
123type BodyReadFailureState = Arc<Mutex<Option<BodyReadFailure>>>;
124
125/// Maps one backend response body read error into this crate's error model.
126///
127/// # Parameters
128/// - `error`: Backend error returned while reading response body bytes.
129/// - `method`: Request method used for diagnostics.
130/// - `url`: Request URL used for diagnostics.
131///
132/// # Returns
133/// [`HttpErrorKind::ReadTimeout`] for timeout errors, otherwise
134/// [`HttpErrorKind::Transport`] for backend body read failures.
135fn map_response_read_error(error: reqwest::Error, method: Method, url: Url, status: StatusCode) -> HttpError {
136    if error.is_timeout() {
137        return map_reqwest_error(error, HttpErrorKind::Transport, ReqwestErrorPhase::Read, method, url)
138            .with_status(status);
139    }
140
141    let error = error.without_url();
142    HttpError::transport(format!("Failed to read response body: {}", error))
143        .with_method(&method)
144        .with_url(&url)
145        .with_status(status)
146        .with_source(error)
147}
148
149/// Runtime state bound to one response instance.
150#[derive(Debug, Clone)]
151struct HttpResponseRuntime {
152    /// Per-response read timeout inherited from request/client.
153    read_timeout: Duration,
154    /// Optional cancellation token inherited from request.
155    cancellation_token: Option<CancellationToken>,
156    /// Request URL used in read/cancellation error context.
157    request_url: Url,
158    /// First response body read failure, if the backend stream failed after
159    /// being taken from this response.
160    body_read_failure: BodyReadFailureState,
161}
162
163impl HttpResponseRuntime {
164    fn new(read_timeout: Duration, cancellation_token: Option<CancellationToken>, request_url: Url) -> Self {
165        Self {
166            read_timeout,
167            cancellation_token,
168            request_url,
169            body_read_failure: Arc::new(Mutex::new(None)),
170        }
171    }
172}
173
174/// Unified HTTP response with lazily consumed body.
175pub struct HttpResponse {
176    /// Response metadata (status, headers, final URL, request method).
177    pub(crate) meta: HttpResponseMeta,
178    /// Raw backend response until consumed.
179    backend: Option<reqwest::Response>,
180    /// Cached full body bytes after eager or lazy read.
181    buffered_body: Option<Bytes>,
182    /// Runtime state inherited from request/client.
183    runtime: HttpResponseRuntime,
184    /// Decode and error-preview options inherited from client options.
185    options: HttpResponseOptions,
186}
187
188impl fmt::Debug for HttpResponse {
189    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
190        let debugger = SanitizedDebugger::new(self.options.log_sanitizer.policy());
191        let url = debugger.url(self.meta.url());
192        let request_url = debugger.url(&self.runtime.request_url);
193        formatter
194            .debug_struct("HttpResponse")
195            .field("status", &self.meta.status())
196            .field("headers", &debugger.headers(self.meta.headers()))
197            .field("url", &url)
198            .field("request_url", &request_url)
199            .field("method", self.meta.method())
200            .field("backend_present", &self.backend.is_some())
201            .field("buffered_body_len", &self.buffered_body.as_ref().map(Bytes::len))
202            .field("read_timeout", &self.runtime.read_timeout)
203            .field("cancellation_token_present", &self.runtime.cancellation_token.is_some())
204            .field("options", &self.options)
205            .finish()
206    }
207}
208
209impl HttpResponse {
210    /// Creates a buffered response.
211    pub fn new(status: StatusCode, headers: HeaderMap, body: Bytes, url: Url, method: Method) -> Self {
212        Self {
213            meta: HttpResponseMeta::new(status, headers, url.clone(), method),
214            backend: None,
215            buffered_body: Some(body),
216            runtime: HttpResponseRuntime::new(Duration::from_secs(30), None, url),
217            options: HttpResponseOptions::default(),
218        }
219    }
220
221    /// Creates a response from backend response and request-scoped options.
222    pub(crate) fn from_backend(
223        meta: HttpResponseMeta,
224        backend: reqwest::Response,
225        read_timeout: Duration,
226        cancellation_token: Option<CancellationToken>,
227        request_url: Url,
228        options: HttpResponseOptions,
229    ) -> Self {
230        Self {
231            meta,
232            backend: Some(backend),
233            buffered_body: None,
234            runtime: HttpResponseRuntime::new(read_timeout, cancellation_token, request_url),
235            options,
236        }
237    }
238
239    /// Returns shared response metadata.
240    #[inline]
241    pub fn meta(&self) -> &HttpResponseMeta {
242        &self.meta
243    }
244
245    /// Returns response status code.
246    #[inline]
247    pub fn status(&self) -> StatusCode {
248        self.meta.status()
249    }
250
251    /// Returns response headers.
252    #[inline]
253    pub fn headers(&self) -> &HeaderMap {
254        self.meta.headers()
255    }
256
257    /// Returns final response URL.
258    #[inline]
259    pub fn url(&self) -> &Url {
260        self.meta.url()
261    }
262
263    /// Returns request URL used in response read context.
264    #[inline]
265    pub fn request_url(&self) -> &Url {
266        &self.runtime.request_url
267    }
268
269    /// Returns whether status is success.
270    #[inline]
271    pub fn is_success(&self) -> bool {
272        self.status().is_success()
273    }
274
275    /// Returns parsed `Retry-After` hint when status and headers provide one.
276    #[inline]
277    pub fn retry_after_hint(&self) -> Option<Duration> {
278        self.meta.retry_after_hint()
279    }
280
281    /// Returns a previous body read failure, if any.
282    ///
283    /// # Returns
284    /// `Some(HttpError)` when an earlier body read failed; otherwise `None`.
285    fn previous_body_read_error(&self) -> Option<HttpError> {
286        let guard = self.runtime.body_read_failure.lock().ok()?;
287        guard.as_ref().map(BodyReadFailure::to_error)
288    }
289
290    /// Stores the first body read failure for later read attempts.
291    ///
292    /// # Parameters
293    /// - `error`: Error produced while reading the response body.
294    fn remember_body_read_failure(&self, error: &HttpError) {
295        Self::remember_body_read_failure_state(&self.runtime.body_read_failure, error);
296    }
297
298    /// Stores the first body read failure in a shared state holder.
299    ///
300    /// # Parameters
301    /// - `state`: Shared failure state captured by response streams.
302    /// - `error`: Error produced while reading the response body.
303    fn remember_body_read_failure_state(state: &BodyReadFailureState, error: &HttpError) {
304        if let Ok(mut guard) = state.lock() {
305            guard.get_or_insert_with(|| BodyReadFailure::from_error(error));
306        }
307    }
308
309    /// Returns `Ok(self)` for success statuses, otherwise maps a status error
310    /// with `Retry-After` and response-body preview context.
311    pub(crate) async fn into_success_or_status_error(self, message_prefix: &str) -> HttpResult<Self> {
312        let status = self.status();
313        if status.is_success() {
314            return Ok(self);
315        }
316        let retry_after = self.retry_after_hint();
317        let method = self.meta.method().clone();
318        let url = self.request_url().clone();
319        let error_preview_limit = self.options.error_response_preview_limit;
320        let diagnostic_sanitizer = LogSanitizer::for_debug(self.options.log_sanitizer.policy());
321        let body_preview = self.into_error_body_preview(error_preview_limit).await?;
322        let log_sanitize_policy = diagnostic_sanitizer.policy().clone();
323        let message = format!(
324            "{} with status {} for {} {}; response body preview: {}",
325            message_prefix,
326            status,
327            method,
328            diagnostic_sanitizer.sanitize_url(&url),
329            body_preview
330        );
331        let mut mapped = HttpError::status(status, message)
332            .with_method(&method)
333            .with_url(&url)
334            .with_response_body_preview(body_preview)
335            .with_log_sanitize_policy(log_sanitize_policy);
336        if let Some(retry_after) = retry_after {
337            mapped = mapped.with_retry_after(retry_after);
338        }
339        Err(mapped)
340    }
341
342    /// Consumes this response and returns a bounded body preview for status errors.
343    ///
344    /// # Errors
345    /// Returns [`HttpErrorKind::Cancelled`](crate::HttpErrorKind::Cancelled)
346    /// when the request cancellation token fires while preview bytes are being
347    /// read.
348    pub(crate) async fn into_error_body_preview(mut self, max_bytes: usize) -> HttpResult<String> {
349        let limit = max_bytes.max(1);
350        let Some(backend) = self.backend.take() else {
351            return Ok("<empty>".to_string());
352        };
353        let content_type = Self::content_type_value(self.meta.headers());
354        self.read_error_body_preview(backend, limit, content_type).await
355    }
356
357    /// Returns full body bytes, consuming backend stream lazily on first call.
358    pub async fn bytes(&mut self) -> HttpResult<Bytes> {
359        if let Some(body) = &self.buffered_body {
360            return Ok(body.clone());
361        }
362        if let Some(error) = self.previous_body_read_error() {
363            return Err(error);
364        }
365        let Some(mut backend) = self.backend.take() else {
366            self.buffered_body = Some(Bytes::new());
367            return Ok(Bytes::new());
368        };
369
370        let method = self.meta.method().clone();
371        let url = self.runtime.request_url.clone();
372        let status = self.meta.status();
373        let read_timeout = self.runtime.read_timeout;
374        let cancellation_token = self.runtime.cancellation_token.clone();
375        let mut body = bytes::BytesMut::new();
376
377        loop {
378            let next = if let Some(token) = &cancellation_token {
379                tokio::select! {
380                    _ = token.cancelled() => {
381                        let error = HttpError::cancelled("Request cancelled while reading response body")
382                            .with_method(&method)
383                            .with_url(&url)
384                            .with_status(status);
385                        self.remember_body_read_failure(&error);
386                        return Err(error);
387                    }
388                    item = tokio::time::timeout(read_timeout, backend.chunk()) => item,
389                }
390            } else {
391                tokio::time::timeout(read_timeout, backend.chunk()).await
392            };
393
394            match next {
395                Ok(Ok(Some(chunk))) => body.extend_from_slice(&chunk),
396                Ok(Ok(None)) => {
397                    let body = body.freeze();
398                    self.buffered_body = Some(body.clone());
399                    return Ok(body);
400                }
401                Ok(Err(error)) => {
402                    let error = map_response_read_error(error, method, url, status);
403                    self.remember_body_read_failure(&error);
404                    return Err(error);
405                }
406                Err(_) => {
407                    let error = HttpError::read_timeout(format!(
408                        "Read timeout after {:?} while reading response body",
409                        read_timeout
410                    ))
411                    .with_method(self.meta.method())
412                    .with_url(&self.runtime.request_url)
413                    .with_status(status);
414                    self.remember_body_read_failure(&error);
415                    return Err(error);
416                }
417            }
418        }
419    }
420
421    /// Returns body as stream; if already buffered, returns stream backed by cached bytes.
422    pub fn stream(&mut self) -> HttpResult<HttpByteStream> {
423        if let Some(body) = self.buffered_body.as_ref() {
424            let bytes = body.clone();
425            return Ok(Box::pin(futures_stream::once(async move { Ok(bytes) })));
426        }
427        if let Some(error) = self.previous_body_read_error() {
428            return Err(error);
429        }
430        if let Some(error) = self.cancelled_error_if_needed("Streaming response cancelled before reading response body")
431        {
432            return Err(error);
433        }
434        let Some(backend) = self.backend.take() else {
435            return Ok(Box::pin(futures_stream::empty()));
436        };
437
438        let method = self.meta.method().clone();
439        let url = self.runtime.request_url.clone();
440        let status = self.meta.status();
441        let read_timeout = self.runtime.read_timeout;
442        let cancellation_token = self.runtime.cancellation_token.clone();
443        let body_read_failure = self.runtime.body_read_failure.clone();
444        let mut stream = backend.bytes_stream();
445        let wrapped = stream! {
446            loop {
447                let next = if let Some(token) = &cancellation_token {
448                    tokio::select! {
449                        _ = token.cancelled() => {
450                            let error = HttpError::cancelled("Streaming response cancelled while reading body")
451                                .with_method(&method)
452                                .with_url(&url)
453                                .with_status(status);
454                            Self::remember_body_read_failure_state(&body_read_failure, &error);
455                            yield Err(error);
456                            break;
457                        }
458                        item = tokio::time::timeout(read_timeout, stream.next()) => item,
459                    }
460                } else {
461                    tokio::time::timeout(read_timeout, stream.next()).await
462                };
463                match next {
464                    Ok(Some(Ok(bytes))) => yield Ok(bytes),
465                    Ok(Some(Err(error))) => {
466                        let mapped = map_response_read_error(error, method.clone(), url.clone(), status);
467                        Self::remember_body_read_failure_state(&body_read_failure, &mapped);
468                        yield Err(mapped);
469                        break;
470                    }
471                    Ok(None) => break,
472                    Err(_) => {
473                        let error = HttpError::read_timeout(format!(
474                            "Read timeout after {:?} while streaming response",
475                            read_timeout
476                        ))
477                        .with_method(&method)
478                        .with_url(&url)
479                        .with_status(status);
480                        Self::remember_body_read_failure_state(&body_read_failure, &error);
481                        yield Err(error);
482                        break;
483                    }
484                }
485            }
486        };
487        Ok(Box::pin(wrapped))
488    }
489
490    /// Interprets response body as UTF-8 text.
491    pub async fn text(&mut self) -> HttpResult<String> {
492        let body = self.bytes().await?;
493        String::from_utf8(body.to_vec()).map_err(|error| {
494            HttpError::decode(format!("Failed to decode response body as UTF-8: {}", error))
495                .with_status(self.meta.status())
496                .with_url(self.meta.url())
497        })
498    }
499
500    /// Deserializes response body as JSON.
501    pub async fn json<T>(&mut self) -> HttpResult<T>
502    where
503        T: DeserializeOwned,
504    {
505        let body = self.bytes().await?;
506        serde_json::from_slice(&body).map_err(|error| {
507            HttpError::decode(format!("Failed to decode response JSON: {}", error))
508                .with_status(self.meta.status())
509                .with_url(self.meta.url())
510        })
511    }
512
513    /// Overrides the maximum allowed size (in bytes) for one SSE line on this response.
514    ///
515    /// Values below 1 are clamped to 1. Returns `self` so callers can chain configuration
516    /// before consuming the body with [`Self::sse_messages`] or [`Self::sse_chunks`]
517    /// (together with [`Self::sse_json_mode`], [`Self::sse_done_marker_policy`], etc.).
518    #[inline]
519    pub fn sse_max_line_bytes(mut self, max_line_bytes: usize) -> Self {
520        self.options.sse_max_line_bytes = max_line_bytes.max(1);
521        self
522    }
523
524    /// Overrides the maximum allowed size (in bytes) for one SSE frame on this response.
525    ///
526    /// Values below 1 are clamped to 1. Returns `self` for chained configuration.
527    #[inline]
528    pub fn sse_max_frame_bytes(mut self, max_frame_bytes: usize) -> Self {
529        self.options.sse_max_frame_bytes = max_frame_bytes.max(1);
530        self
531    }
532
533    /// Overrides the JSON decoding mode used by [`Self::sse_chunks`] on this response.
534    #[inline]
535    pub fn sse_json_mode(mut self, mode: SseJsonMode) -> Self {
536        self.options.sse_json_mode = mode;
537        self
538    }
539
540    /// Overrides how [`Self::sse_chunks`] detects end-of-stream from trimmed `data:` payloads.
541    #[inline]
542    pub fn sse_done_marker_policy(mut self, policy: DoneMarkerPolicy) -> Self {
543        self.options.sse_done_marker_policy = policy;
544        self
545    }
546
547    /// Decodes body stream as SSE messages using this response's SSE line/frame byte limits (from
548    /// client defaults unless overridden via [`Self::sse_max_line_bytes`] /
549    /// [`Self::sse_max_frame_bytes`]).
550    pub fn sse_messages(mut self) -> SseMessageStream {
551        let max_line_bytes = self.options.sse_max_line_bytes;
552        let max_frame_bytes = self.options.sse_max_frame_bytes;
553        match self.stream() {
554            Ok(stream) => crate::sse::decode_messages_from_stream_with_limits(stream, max_line_bytes, max_frame_bytes),
555            Err(error) => Box::pin(futures_stream::once(async move { Err(error) })),
556        }
557    }
558
559    /// Decodes body stream as internal SSE records for reconnect state handling.
560    ///
561    /// # Returns
562    /// Stream of internal records, or one error item when the body cannot be opened.
563    pub(crate) fn sse_records(mut self) -> crate::sse::SseRecordStream {
564        let max_line_bytes = self.options.sse_max_line_bytes;
565        let max_frame_bytes = self.options.sse_max_frame_bytes;
566        match self.stream() {
567            Ok(stream) => crate::sse::decode_records_from_stream_with_limits(stream, max_line_bytes, max_frame_bytes),
568            Err(error) => Box::pin(futures_stream::once(async move { Err(error) })),
569        }
570    }
571
572    /// Decodes SSE `data:` lines as JSON chunks using this response's SSE JSON mode, done-marker
573    /// policy, and line/frame limits (see [`Self::sse_json_mode`], [`Self::sse_done_marker_policy`],
574    /// [`Self::sse_max_line_bytes`], [`Self::sse_max_frame_bytes`]).
575    pub fn sse_chunks<T>(mut self) -> SseChunkStream<T>
576    where
577        T: DeserializeOwned + Send + 'static,
578    {
579        let done_policy = self.options.sse_done_marker_policy.clone();
580        let mode = self.options.sse_json_mode;
581        let max_line_bytes = self.options.sse_max_line_bytes;
582        let max_frame_bytes = self.options.sse_max_frame_bytes;
583        match self.stream() {
584            Ok(stream) => crate::sse::decode_json_chunks_from_stream_with_limits(
585                stream,
586                done_policy,
587                mode,
588                max_line_bytes,
589                max_frame_bytes,
590            ),
591            Err(error) => Box::pin(futures_stream::once(async move { Err(error) })),
592        }
593    }
594
595    /// Returns a buffered body reference for response logging if available.
596    ///
597    /// # Returns
598    /// `Some(&Bytes)` when response body has already been buffered.
599    pub(crate) fn buffered_body_for_logging(&self) -> Option<&Bytes> {
600        self.buffered_body.as_ref()
601    }
602
603    /// Returns whether logger may safely buffer the full body for logging.
604    ///
605    /// # Parameters
606    /// - `body_log_limit`: Configured logging body preview limit in bytes.
607    ///
608    /// # Returns
609    /// `true` only when this response is not SSE, has an explicit `Content-Length`,
610    /// and declared length is within `body_log_limit`.
611    pub(crate) fn can_buffer_body_for_logging(&self, body_log_limit: usize) -> bool {
612        if self.backend.is_none() {
613            return false;
614        }
615        if self.is_sse_response() {
616            return false;
617        }
618        self.content_length_hint()
619            .is_some_and(|content_length| content_length <= body_log_limit as u64)
620    }
621
622    /// Reads bounded preview bytes from a response body for status error messages.
623    ///
624    /// # Errors
625    /// Returns [`HttpErrorKind::Cancelled`](crate::HttpErrorKind::Cancelled)
626    /// when the supplied cancellation token fires while waiting for preview
627    /// bytes.
628    async fn read_error_body_preview(
629        &self,
630        mut response: reqwest::Response,
631        max_bytes: usize,
632        content_type: Option<String>,
633    ) -> HttpResult<String> {
634        let limit = max_bytes.max(1);
635        let read_timeout = self.runtime.read_timeout;
636        let cancellation_token = self.runtime.cancellation_token.clone();
637        let method = self.meta.method().clone();
638        let url = self.runtime.request_url.clone();
639        let status = self.meta.status();
640        let mut preview = Vec::new();
641        let mut truncated = false;
642        let capture_limit = limit.saturating_add(1);
643
644        loop {
645            let next = if let Some(token) = cancellation_token.as_ref() {
646                tokio::select! {
647                    _ = token.cancelled() => {
648                        return Err(HttpError::cancelled(
649                            "Request cancelled while reading status error response body preview",
650                        )
651                        .with_method(&method)
652                        .with_url(&url)
653                        .with_status(status));
654                    }
655                    item = tokio::time::timeout(read_timeout, response.chunk()) => item,
656                }
657            } else {
658                tokio::time::timeout(read_timeout, response.chunk()).await
659            };
660            match next {
661                Ok(Ok(Some(chunk))) => {
662                    let remaining = capture_limit.saturating_sub(preview.len());
663                    if chunk.len() > remaining {
664                        preview.extend_from_slice(&chunk[..remaining]);
665                        truncated = true;
666                        break;
667                    }
668                    preview.extend_from_slice(&chunk);
669                    if preview.len() > limit {
670                        truncated = true;
671                        break;
672                    }
673                }
674                Ok(Ok(None)) => break,
675                Ok(Err(error)) => {
676                    let error = error.without_url();
677                    return Ok(format!(
678                        "<error body unavailable: failed to read response body: {}>",
679                        error
680                    ));
681                }
682                Err(_) => {
683                    return Ok(format!(
684                        "<error body unavailable: read timeout after {:?}>",
685                        read_timeout
686                    ));
687                }
688            }
689        }
690        let source_len = preview.len();
691        if preview.len() > limit {
692            preview.truncate(limit);
693            truncated = true;
694        }
695        Ok(Self::render_error_body_preview(
696            &preview,
697            source_len,
698            truncated,
699            content_type.as_deref(),
700            &self.options.log_sanitizer,
701        ))
702    }
703
704    /// Returns a cancellation error with response read context when cancelled.
705    ///
706    /// # Parameters
707    /// - `message`: Cancellation message to include in the error.
708    ///
709    /// # Returns
710    /// `Some(HttpError)` when this response has a cancelled token; otherwise
711    /// `None`.
712    fn cancelled_error_if_needed(&self, message: &str) -> Option<HttpError> {
713        if self
714            .runtime
715            .cancellation_token
716            .as_ref()
717            .is_some_and(CancellationToken::is_cancelled)
718        {
719            Some(
720                HttpError::cancelled(message.to_string())
721                    .with_method(self.meta.method())
722                    .with_url(&self.runtime.request_url),
723            )
724        } else {
725            None
726        }
727    }
728
729    /// Returns `Content-Length` parsed from response headers when present and valid.
730    fn content_length_hint(&self) -> Option<u64> {
731        self.meta
732            .headers()
733            .get(CONTENT_LENGTH)
734            .and_then(|value| value.to_str().ok())
735            .and_then(|value| value.parse::<u64>().ok())
736    }
737
738    /// Returns whether response content-type is SSE (`text/event-stream`).
739    fn is_sse_response(&self) -> bool {
740        self.meta
741            .headers()
742            .get(CONTENT_TYPE)
743            .and_then(|value| value.to_str().ok())
744            .is_some_and(content_type::is_sse)
745    }
746
747    fn render_error_body_preview(
748        bytes: &[u8],
749        source_len: usize,
750        truncated: bool,
751        content_type: Option<&str>,
752        log_sanitizer: &LogSanitizer,
753    ) -> String {
754        let preview = BodyPreview::from_limited_bytes(bytes, source_len, truncated, BodyLogContext::ErrorResponse);
755        let preview = if let Some(content_type) = content_type {
756            preview.with_content_type(content_type)
757        } else {
758            preview
759        };
760        log_sanitizer.sanitize_body_preview(&preview)
761    }
762
763    /// Extracts a UTF-8 Content-Type header value.
764    ///
765    /// # Parameters
766    /// - `headers`: Headers to inspect.
767    ///
768    /// # Returns
769    /// Owned Content-Type value when present and UTF-8.
770    fn content_type_value(headers: &HeaderMap) -> Option<String> {
771        headers
772            .get(CONTENT_TYPE)
773            .and_then(|value| value.to_str().ok())
774            .map(str::to_string)
775    }
776}
qubit_http/response/http_response.rs

qubit_http/response/
http_response.rs
