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

qubit_http/response/
http_response.rs
