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::sse::{
44    DoneMarkerPolicy,
45    SseChunkStream,
46    SseJsonMode,
47    SseMessageStream,
48};
49use crate::{
50    BodyLogContext,
51    BodyPreview,
52    HttpByteStream,
53    HttpError,
54    HttpErrorKind,
55    HttpResult,
56    LogSanitizer,
57};
58
59use super::{
60    HttpResponseMeta,
61    HttpResponseOptions,
62};
63
64/// Snapshot of a body read failure retained after the backend body is consumed.
65#[derive(Debug, Clone)]
66struct BodyReadFailure {
67    /// Error category from the original failure.
68    kind: HttpErrorKind,
69    /// Original failure message.
70    message: String,
71    /// Optional request method context.
72    method: Option<Method>,
73    /// Optional request URL context.
74    url: Option<Url>,
75    /// Optional HTTP status context.
76    status: Option<StatusCode>,
77}
78
79impl BodyReadFailure {
80    /// Captures cloneable diagnostic fields from a read error.
81    ///
82    /// # Parameters
83    /// - `error`: Original body read failure.
84    ///
85    /// # Returns
86    /// Cloneable failure snapshot for subsequent reads.
87    fn from_error(error: &HttpError) -> Self {
88        Self {
89            kind: error.kind,
90            message: error.message.clone(),
91            method: error.method.clone(),
92            url: error.url.clone(),
93            status: error.status,
94        }
95    }
96
97    /// Rebuilds an [`HttpError`] for a later read attempt.
98    ///
99    /// # Returns
100    /// Error preserving the original kind and cloneable context.
101    fn to_error(&self) -> HttpError {
102        let mut error = HttpError::new(
103            self.kind,
104            format!("previous response body read failed: {}", self.message),
105        );
106        if let Some(method) = self.method.as_ref() {
107            error = error.with_method(method);
108        }
109        if let Some(url) = self.url.as_ref() {
110            error = error.with_url(url);
111        }
112        if let Some(status) = self.status {
113            error = error.with_status(status);
114        }
115        error
116    }
117}
118
119/// Shared response body failure state.
120type BodyReadFailureState = Arc<Mutex<Option<BodyReadFailure>>>;
121
122/// Maps one backend response body read error into this crate's error model.
123///
124/// # Parameters
125/// - `error`: Backend error returned while reading response body bytes.
126/// - `method`: Request method used for diagnostics.
127/// - `url`: Request URL used for diagnostics.
128///
129/// # Returns
130/// [`HttpErrorKind::ReadTimeout`] for timeout errors, otherwise
131/// [`HttpErrorKind::Transport`] for backend body read failures.
132fn map_response_read_error(
133    error: reqwest::Error,
134    method: Method,
135    url: Url,
136    status: StatusCode,
137) -> HttpError {
138    if error.is_timeout() {
139        return map_reqwest_error(
140            error,
141            HttpErrorKind::Transport,
142            ReqwestErrorPhase::Read,
143            method,
144            url,
145        )
146        .with_status(status);
147    }
148
149    let error = error.without_url();
150    HttpError::transport(format!("Failed to read response body: {}", error))
151        .with_method(&method)
152        .with_url(&url)
153        .with_status(status)
154        .with_source(error)
155}
156
157/// Runtime state bound to one response instance.
158#[derive(Debug, Clone)]
159struct HttpResponseRuntime {
160    /// Per-response read timeout inherited from request/client.
161    read_timeout: Duration,
162    /// Optional cancellation token inherited from request.
163    cancellation_token: Option<CancellationToken>,
164    /// Request URL used in read/cancellation error context.
165    request_url: Url,
166    /// First response body read failure, if the backend stream failed after
167    /// being taken from this response.
168    body_read_failure: BodyReadFailureState,
169}
170
171impl HttpResponseRuntime {
172    fn new(
173        read_timeout: Duration,
174        cancellation_token: Option<CancellationToken>,
175        request_url: Url,
176    ) -> Self {
177        Self {
178            read_timeout,
179            cancellation_token,
180            request_url,
181            body_read_failure: Arc::new(Mutex::new(None)),
182        }
183    }
184}
185
186/// Unified HTTP response with lazily consumed body.
187pub struct HttpResponse {
188    /// Response metadata (status, headers, final URL, request method).
189    pub(crate) meta: HttpResponseMeta,
190    /// Raw backend response until consumed.
191    backend: Option<reqwest::Response>,
192    /// Cached full body bytes after eager or lazy read.
193    buffered_body: Option<Bytes>,
194    /// Runtime state inherited from request/client.
195    runtime: HttpResponseRuntime,
196    /// Decode and error-preview options inherited from client options.
197    options: HttpResponseOptions,
198}
199
200impl fmt::Debug for HttpResponse {
201    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
202        let sanitizer = LogSanitizer::for_debug(self.options.log_sanitizer.policy());
203        let url = sanitizer.sanitize_url(self.meta.url());
204        let request_url = sanitizer.sanitize_url(&self.runtime.request_url);
205        formatter
206            .debug_struct("HttpResponse")
207            .field("status", &self.meta.status())
208            .field(
209                "headers",
210                &sanitizer.sanitize_header_map(self.meta.headers()),
211            )
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 message = format!(
353            "{} with status {} for {} {}; response body preview: {}",
354            message_prefix,
355            status,
356            method,
357            diagnostic_sanitizer.sanitize_url(&url),
358            body_preview
359        );
360        let mut mapped = HttpError::status(status, message)
361            .with_method(&method)
362            .with_url(&url)
363            .with_response_body_preview(body_preview);
364        if let Some(retry_after) = retry_after {
365            mapped = mapped.with_retry_after(retry_after);
366        }
367        Err(mapped)
368    }
369
370    /// Consumes this response and returns a bounded body preview for status errors.
371    ///
372    /// # Errors
373    /// Returns [`HttpErrorKind::Cancelled`](crate::HttpErrorKind::Cancelled)
374    /// when the request cancellation token fires while preview bytes are being
375    /// read.
376    pub(crate) async fn into_error_body_preview(mut self, max_bytes: usize) -> HttpResult<String> {
377        let limit = max_bytes.max(1);
378        let Some(backend) = self.backend.take() else {
379            return Ok("<empty>".to_string());
380        };
381        let content_type = Self::content_type_value(self.meta.headers());
382        self.read_error_body_preview(backend, limit, content_type)
383            .await
384    }
385
386    /// Returns full body bytes, consuming backend stream lazily on first call.
387    pub async fn bytes(&mut self) -> HttpResult<Bytes> {
388        if let Some(body) = &self.buffered_body {
389            return Ok(body.clone());
390        }
391        if let Some(error) = self.previous_body_read_error() {
392            return Err(error);
393        }
394        let Some(mut backend) = self.backend.take() else {
395            self.buffered_body = Some(Bytes::new());
396            return Ok(Bytes::new());
397        };
398
399        let method = self.meta.method().clone();
400        let url = self.runtime.request_url.clone();
401        let status = self.meta.status();
402        let read_timeout = self.runtime.read_timeout;
403        let cancellation_token = self.runtime.cancellation_token.clone();
404        let mut body = bytes::BytesMut::new();
405
406        loop {
407            let next = if let Some(token) = &cancellation_token {
408                tokio::select! {
409                    _ = token.cancelled() => {
410                        let error = HttpError::cancelled("Request cancelled while reading response body")
411                            .with_method(&method)
412                            .with_url(&url)
413                            .with_status(status);
414                        self.remember_body_read_failure(&error);
415                        return Err(error);
416                    }
417                    item = tokio::time::timeout(read_timeout, backend.chunk()) => item,
418                }
419            } else {
420                tokio::time::timeout(read_timeout, backend.chunk()).await
421            };
422
423            match next {
424                Ok(Ok(Some(chunk))) => body.extend_from_slice(&chunk),
425                Ok(Ok(None)) => {
426                    let body = body.freeze();
427                    self.buffered_body = Some(body.clone());
428                    return Ok(body);
429                }
430                Ok(Err(error)) => {
431                    let error = map_response_read_error(error, method, url, status);
432                    self.remember_body_read_failure(&error);
433                    return Err(error);
434                }
435                Err(_) => {
436                    let error = HttpError::read_timeout(format!(
437                        "Read timeout after {:?} while reading response body",
438                        read_timeout
439                    ))
440                    .with_method(self.meta.method())
441                    .with_url(&self.runtime.request_url)
442                    .with_status(status);
443                    self.remember_body_read_failure(&error);
444                    return Err(error);
445                }
446            }
447        }
448    }
449
450    /// Returns body as stream; if already buffered, returns stream backed by cached bytes.
451    pub fn stream(&mut self) -> HttpResult<HttpByteStream> {
452        if let Some(body) = self.buffered_body.as_ref() {
453            let bytes = body.clone();
454            return Ok(Box::pin(futures_stream::once(async move { Ok(bytes) })));
455        }
456        if let Some(error) = self.previous_body_read_error() {
457            return Err(error);
458        }
459        if let Some(error) = self
460            .cancelled_error_if_needed("Streaming response cancelled before reading response body")
461        {
462            return Err(error);
463        }
464        let Some(backend) = self.backend.take() else {
465            return Ok(Box::pin(futures_stream::empty()));
466        };
467
468        let method = self.meta.method().clone();
469        let url = self.runtime.request_url.clone();
470        let status = self.meta.status();
471        let read_timeout = self.runtime.read_timeout;
472        let cancellation_token = self.runtime.cancellation_token.clone();
473        let body_read_failure = self.runtime.body_read_failure.clone();
474        let mut stream = backend.bytes_stream();
475        let wrapped = stream! {
476            loop {
477                let next = if let Some(token) = &cancellation_token {
478                    tokio::select! {
479                        _ = token.cancelled() => {
480                            let error = HttpError::cancelled("Streaming response cancelled while reading body")
481                                .with_method(&method)
482                                .with_url(&url)
483                                .with_status(status);
484                            Self::remember_body_read_failure_state(&body_read_failure, &error);
485                            yield Err(error);
486                            break;
487                        }
488                        item = tokio::time::timeout(read_timeout, stream.next()) => item,
489                    }
490                } else {
491                    tokio::time::timeout(read_timeout, stream.next()).await
492                };
493                match next {
494                    Ok(Some(Ok(bytes))) => yield Ok(bytes),
495                    Ok(Some(Err(error))) => {
496                        let mapped = map_response_read_error(error, method.clone(), url.clone(), status);
497                        Self::remember_body_read_failure_state(&body_read_failure, &mapped);
498                        yield Err(mapped);
499                        break;
500                    }
501                    Ok(None) => break,
502                    Err(_) => {
503                        let error = HttpError::read_timeout(format!(
504                            "Read timeout after {:?} while streaming response",
505                            read_timeout
506                        ))
507                        .with_method(&method)
508                        .with_url(&url)
509                        .with_status(status);
510                        Self::remember_body_read_failure_state(&body_read_failure, &error);
511                        yield Err(error);
512                        break;
513                    }
514                }
515            }
516        };
517        Ok(Box::pin(wrapped))
518    }
519
520    /// Interprets response body as UTF-8 text.
521    pub async fn text(&mut self) -> HttpResult<String> {
522        let body = self.bytes().await?;
523        String::from_utf8(body.to_vec()).map_err(|error| {
524            HttpError::decode(format!(
525                "Failed to decode response body as UTF-8: {}",
526                error
527            ))
528            .with_status(self.meta.status())
529            .with_url(self.meta.url())
530        })
531    }
532
533    /// Deserializes response body as JSON.
534    pub async fn json<T>(&mut self) -> HttpResult<T>
535    where
536        T: DeserializeOwned,
537    {
538        let body = self.bytes().await?;
539        serde_json::from_slice(&body).map_err(|error| {
540            HttpError::decode(format!("Failed to decode response JSON: {}", error))
541                .with_status(self.meta.status())
542                .with_url(self.meta.url())
543        })
544    }
545
546    /// Overrides the maximum allowed size (in bytes) for one SSE line on this response.
547    ///
548    /// Values below 1 are clamped to 1. Returns `self` so callers can chain configuration
549    /// before consuming the body with [`Self::sse_messages`] or [`Self::sse_chunks`]
550    /// (together with [`Self::sse_json_mode`], [`Self::sse_done_marker_policy`], etc.).
551    #[inline]
552    pub fn sse_max_line_bytes(mut self, max_line_bytes: usize) -> Self {
553        self.options.sse_max_line_bytes = max_line_bytes.max(1);
554        self
555    }
556
557    /// Overrides the maximum allowed size (in bytes) for one SSE frame on this response.
558    ///
559    /// Values below 1 are clamped to 1. Returns `self` for chained configuration.
560    #[inline]
561    pub fn sse_max_frame_bytes(mut self, max_frame_bytes: usize) -> Self {
562        self.options.sse_max_frame_bytes = max_frame_bytes.max(1);
563        self
564    }
565
566    /// Overrides the JSON decoding mode used by [`Self::sse_chunks`] on this response.
567    #[inline]
568    pub fn sse_json_mode(mut self, mode: SseJsonMode) -> Self {
569        self.options.sse_json_mode = mode;
570        self
571    }
572
573    /// Overrides how [`Self::sse_chunks`] detects end-of-stream from trimmed `data:` payloads.
574    #[inline]
575    pub fn sse_done_marker_policy(mut self, policy: DoneMarkerPolicy) -> Self {
576        self.options.sse_done_marker_policy = policy;
577        self
578    }
579
580    /// Decodes body stream as SSE messages using this response's SSE line/frame byte limits (from
581    /// client defaults unless overridden via [`Self::sse_max_line_bytes`] /
582    /// [`Self::sse_max_frame_bytes`]).
583    pub fn sse_messages(mut self) -> SseMessageStream {
584        let max_line_bytes = self.options.sse_max_line_bytes;
585        let max_frame_bytes = self.options.sse_max_frame_bytes;
586        match self.stream() {
587            Ok(stream) => crate::sse::decode_messages_from_stream_with_limits(
588                stream,
589                max_line_bytes,
590                max_frame_bytes,
591            ),
592            Err(error) => Box::pin(futures_stream::once(async move { Err(error) })),
593        }
594    }
595
596    /// Decodes body stream as internal SSE records for reconnect state handling.
597    ///
598    /// # Returns
599    /// Stream of internal records, or one error item when the body cannot be opened.
600    pub(crate) fn sse_records(mut self) -> crate::sse::SseRecordStream {
601        let max_line_bytes = self.options.sse_max_line_bytes;
602        let max_frame_bytes = self.options.sse_max_frame_bytes;
603        match self.stream() {
604            Ok(stream) => crate::sse::decode_records_from_stream_with_limits(
605                stream,
606                max_line_bytes,
607                max_frame_bytes,
608            ),
609            Err(error) => Box::pin(futures_stream::once(async move { Err(error) })),
610        }
611    }
612
613    /// Decodes SSE `data:` lines as JSON chunks using this response's SSE JSON mode, done-marker
614    /// policy, and line/frame limits (see [`Self::sse_json_mode`], [`Self::sse_done_marker_policy`],
615    /// [`Self::sse_max_line_bytes`], [`Self::sse_max_frame_bytes`]).
616    pub fn sse_chunks<T>(mut self) -> SseChunkStream<T>
617    where
618        T: DeserializeOwned + Send + 'static,
619    {
620        let done_policy = self.options.sse_done_marker_policy.clone();
621        let mode = self.options.sse_json_mode;
622        let max_line_bytes = self.options.sse_max_line_bytes;
623        let max_frame_bytes = self.options.sse_max_frame_bytes;
624        match self.stream() {
625            Ok(stream) => crate::sse::decode_json_chunks_from_stream_with_limits(
626                stream,
627                done_policy,
628                mode,
629                max_line_bytes,
630                max_frame_bytes,
631            ),
632            Err(error) => Box::pin(futures_stream::once(async move { Err(error) })),
633        }
634    }
635
636    /// Returns a buffered body reference for response logging if available.
637    ///
638    /// # Returns
639    /// `Some(&Bytes)` when response body has already been buffered.
640    pub(crate) fn buffered_body_for_logging(&self) -> Option<&Bytes> {
641        self.buffered_body.as_ref()
642    }
643
644    /// Returns whether logger may safely buffer the full body for logging.
645    ///
646    /// # Parameters
647    /// - `body_log_limit`: Configured logging body preview limit in bytes.
648    ///
649    /// # Returns
650    /// `true` only when this response is not SSE, has an explicit `Content-Length`,
651    /// and declared length is within `body_log_limit`.
652    pub(crate) fn can_buffer_body_for_logging(&self, body_log_limit: usize) -> bool {
653        if self.backend.is_none() {
654            return false;
655        }
656        if self.is_sse_response() {
657            return false;
658        }
659        self.content_length_hint()
660            .is_some_and(|content_length| content_length <= body_log_limit as u64)
661    }
662
663    /// Reads bounded preview bytes from a response body for status error messages.
664    ///
665    /// # Errors
666    /// Returns [`HttpErrorKind::Cancelled`](crate::HttpErrorKind::Cancelled)
667    /// when the supplied cancellation token fires while waiting for preview
668    /// bytes.
669    async fn read_error_body_preview(
670        &self,
671        mut response: reqwest::Response,
672        max_bytes: usize,
673        content_type: Option<String>,
674    ) -> HttpResult<String> {
675        let limit = max_bytes.max(1);
676        let read_timeout = self.runtime.read_timeout;
677        let cancellation_token = self.runtime.cancellation_token.clone();
678        let method = self.meta.method().clone();
679        let url = self.runtime.request_url.clone();
680        let status = self.meta.status();
681        let mut preview = Vec::new();
682        let mut truncated = false;
683
684        loop {
685            let next = if let Some(token) = cancellation_token.as_ref() {
686                tokio::select! {
687                    _ = token.cancelled() => {
688                        return Err(HttpError::cancelled(
689                            "Request cancelled while reading status error response body preview",
690                        )
691                        .with_method(&method)
692                        .with_url(&url)
693                        .with_status(status));
694                    }
695                    item = tokio::time::timeout(read_timeout, response.chunk()) => item,
696                }
697            } else {
698                tokio::time::timeout(read_timeout, response.chunk()).await
699            };
700            match next {
701                Ok(Ok(Some(chunk))) => {
702                    if preview.len() >= limit {
703                        truncated = true;
704                        break;
705                    }
706                    let remaining = limit - preview.len();
707                    if chunk.len() > remaining {
708                        preview.extend_from_slice(&chunk[..remaining]);
709                        truncated = true;
710                        break;
711                    }
712                    preview.extend_from_slice(&chunk);
713                }
714                Ok(Ok(None)) => break,
715                Ok(Err(error)) => {
716                    let error = error.without_url();
717                    return Ok(format!(
718                        "<error body unavailable: failed to read response body: {}>",
719                        error
720                    ));
721                }
722                Err(_) => {
723                    return Ok(format!(
724                        "<error body unavailable: read timeout after {:?}>",
725                        read_timeout
726                    ));
727                }
728            }
729        }
730        Ok(Self::render_error_body_preview(
731            &preview,
732            preview.len(),
733            truncated,
734            content_type.as_deref(),
735            &self.options.log_sanitizer,
736        ))
737    }
738
739    /// Returns a cancellation error with response read context when cancelled.
740    ///
741    /// # Parameters
742    /// - `message`: Cancellation message to include in the error.
743    ///
744    /// # Returns
745    /// `Some(HttpError)` when this response has a cancelled token; otherwise
746    /// `None`.
747    fn cancelled_error_if_needed(&self, message: &str) -> Option<HttpError> {
748        if self
749            .runtime
750            .cancellation_token
751            .as_ref()
752            .is_some_and(CancellationToken::is_cancelled)
753        {
754            Some(
755                HttpError::cancelled(message.to_string())
756                    .with_method(self.meta.method())
757                    .with_url(&self.runtime.request_url),
758            )
759        } else {
760            None
761        }
762    }
763
764    /// Returns `Content-Length` parsed from response headers when present and valid.
765    fn content_length_hint(&self) -> Option<u64> {
766        self.meta
767            .headers()
768            .get(CONTENT_LENGTH)
769            .and_then(|value| value.to_str().ok())
770            .and_then(|value| value.parse::<u64>().ok())
771    }
772
773    /// Returns whether response content-type is SSE (`text/event-stream`).
774    fn is_sse_response(&self) -> bool {
775        self.meta
776            .headers()
777            .get(CONTENT_TYPE)
778            .and_then(|value| value.to_str().ok())
779            .is_some_and(content_type::is_sse)
780    }
781
782    fn render_error_body_preview(
783        bytes: &[u8],
784        source_len: usize,
785        truncated: bool,
786        content_type: Option<&str>,
787        log_sanitizer: &crate::LogSanitizer,
788    ) -> String {
789        let preview = BodyPreview::from_limited_bytes(
790            bytes,
791            source_len,
792            truncated,
793            BodyLogContext::ErrorResponse,
794        );
795        let preview = if let Some(content_type) = content_type {
796            preview.with_content_type(content_type)
797        } else {
798            preview
799        };
800        log_sanitizer.sanitize_body_preview(&preview)
801    }
802
803    /// Extracts a UTF-8 Content-Type header value.
804    ///
805    /// # Parameters
806    /// - `headers`: Headers to inspect.
807    ///
808    /// # Returns
809    /// Owned Content-Type value when present and UTF-8.
810    fn content_type_value(headers: &HeaderMap) -> Option<String> {
811        headers
812            .get(CONTENT_TYPE)
813            .and_then(|value| value.to_str().ok())
814            .map(str::to_string)
815    }
816}
qubit_http/response/http_response.rs

qubit_http/response/
http_response.rs
