rama_http/service/client/
ext.rs

1use crate::{Method, Request, Response};
2use rama_core::error::BoxErrorExt as _;
3use rama_core::{
4    Service,
5    error::{BoxError, ErrorExt},
6    extensions::{Extension, Extensions, ExtensionsRef},
7};
8use rama_http_headers::authorization::Credentials;
9use rama_http_types::StreamingBody;
10use rama_net::uri::Uri;
11use std::time::Duration;
12
13/// Extends an Http Client with high level features,
14/// to facilitate the creation and sending of http requests,
15/// in a more ergonomic way.
16pub trait HttpClientExt: private::HttpClientExtSealed + Sized + Send + Sync + 'static {
17    /// The response type returned by the `execute` method.
18    type ExecuteResponse;
19    /// The error type returned by the `execute` method.
20    type ExecuteError;
21
22    /// Convenience method to make a `GET` request to a URL.
23    ///
24    /// # Errors
25    ///
26    /// This method fails whenever the supplied [`Url`] cannot be parsed.
27    ///
28    /// [`Url`]: rama_net::uri::Uri
29    fn get(&self, url: impl IntoUrl) -> RequestBuilder<'_, Self, Self::ExecuteResponse>;
30
31    /// Convenience method to make a `POST` request to a URL.
32    ///
33    /// # Errors
34    ///
35    /// This method fails whenever the supplied [`Url`] cannot be parsed.
36    ///
37    /// [`Url`]: rama_net::uri::Uri
38    fn post(&self, url: impl IntoUrl) -> RequestBuilder<'_, Self, Self::ExecuteResponse>;
39
40    /// Convenience method to make a `PUT` request to a URL.
41    ///
42    /// # Errors
43    ///
44    /// This method fails whenever the supplied [`Url`] cannot be parsed.
45    ///
46    /// [`Url`]: rama_net::uri::Uri
47    fn put(&self, url: impl IntoUrl) -> RequestBuilder<'_, Self, Self::ExecuteResponse>;
48
49    /// Convenience method to make a `PATCH` request to a URL.
50    ///
51    /// # Errors
52    ///
53    /// This method fails whenever the supplied [`Url`] cannot be parsed.
54    ///
55    /// [`Url`]: rama_net::uri::Uri
56    fn patch(&self, url: impl IntoUrl) -> RequestBuilder<'_, Self, Self::ExecuteResponse>;
57
58    /// Convenience method to make a `DELETE` request to a URL.
59    ///
60    /// # Errors
61    ///
62    /// This method fails whenever the supplied [`Url`] cannot be parsed.
63    ///
64    /// [`Url`]: rama_net::uri::Uri
65    fn delete(&self, url: impl IntoUrl) -> RequestBuilder<'_, Self, Self::ExecuteResponse>;
66
67    /// Convenience method to make a `HEAD` request to a URL.
68    ///
69    /// # Errors
70    ///
71    /// This method fails whenever the supplied `Url` cannot be parsed.
72    fn head(&self, url: impl IntoUrl) -> RequestBuilder<'_, Self, Self::ExecuteResponse>;
73
74    /// Convenience method to make a `CONNECT` request to a URL.
75    ///
76    /// # Errors
77    ///
78    /// This method fails whenever the supplied `Url` cannot be parsed.
79    fn connect(&self, url: impl IntoUrl) -> RequestBuilder<'_, Self, Self::ExecuteResponse>;
80
81    /// Start building a [`Request`] with the [`Method`] and [`Url`].
82    ///
83    /// Returns a [`RequestBuilder`], which will allow setting headers and
84    /// the request body before sending.
85    ///
86    /// [`Request`]: crate::Request
87    /// [`Method`]: crate::Method
88    /// [`Url`]: rama_net::uri::Uri
89    ///
90    /// # Errors
91    ///
92    /// This method fails whenever the supplied [`Url`] cannot be parsed.
93    fn request(
94        &self,
95        method: Method,
96        url: impl IntoUrl,
97    ) -> RequestBuilder<'_, Self, Self::ExecuteResponse>;
98
99    /// Start building a [`Request`], using the given [`Request`].
100    ///
101    /// Returns a [`RequestBuilder`], which will allow setting headers and
102    /// the request body before sending.
103    fn build_from_request<Body: Into<crate::Body>>(
104        &self,
105        request: Request<Body>,
106    ) -> RequestBuilder<'_, Self, Self::ExecuteResponse>;
107
108    /// Executes a `Request`.
109    ///
110    /// # Errors
111    ///
112    /// This method fails if there was an error while sending request.
113    fn execute(
114        &self,
115
116        request: Request,
117    ) -> impl Future<Output = Result<Self::ExecuteResponse, Self::ExecuteError>>;
118}
119
120impl<S, Body> HttpClientExt for S
121where
122    S: Service<Request, Output = Response<Body>, Error: Into<BoxError>>,
123{
124    type ExecuteResponse = Response<Body>;
125    type ExecuteError = S::Error;
126
127    fn get(&self, url: impl IntoUrl) -> RequestBuilder<'_, Self, Self::ExecuteResponse> {
128        self.request(Method::GET, url)
129    }
130
131    fn post(&self, url: impl IntoUrl) -> RequestBuilder<'_, Self, Self::ExecuteResponse> {
132        self.request(Method::POST, url)
133    }
134
135    fn put(&self, url: impl IntoUrl) -> RequestBuilder<'_, Self, Self::ExecuteResponse> {
136        self.request(Method::PUT, url)
137    }
138
139    fn patch(&self, url: impl IntoUrl) -> RequestBuilder<'_, Self, Self::ExecuteResponse> {
140        self.request(Method::PATCH, url)
141    }
142
143    fn delete(&self, url: impl IntoUrl) -> RequestBuilder<'_, Self, Self::ExecuteResponse> {
144        self.request(Method::DELETE, url)
145    }
146
147    fn head(&self, url: impl IntoUrl) -> RequestBuilder<'_, Self, Self::ExecuteResponse> {
148        self.request(Method::HEAD, url)
149    }
150
151    fn connect(&self, url: impl IntoUrl) -> RequestBuilder<'_, Self, Self::ExecuteResponse> {
152        self.request(Method::CONNECT, url)
153    }
154
155    fn request(
156        &self,
157        method: Method,
158        url: impl IntoUrl,
159    ) -> RequestBuilder<'_, Self, Self::ExecuteResponse> {
160        let uri = match url.into_url() {
161            Ok(uri) => uri,
162            Err(err) => {
163                return RequestBuilder {
164                    http_client_service: self,
165                    state: RequestBuilderState::Error(err),
166                    _phantom: std::marker::PhantomData,
167                    _mode: std::marker::PhantomData,
168                };
169            }
170        };
171
172        let builder = crate::request::Builder::new().method(method).uri(uri);
173
174        RequestBuilder {
175            http_client_service: self,
176            state: RequestBuilderState::PreBody(builder),
177            _phantom: std::marker::PhantomData,
178            _mode: std::marker::PhantomData,
179        }
180    }
181
182    fn build_from_request<RequestBody: Into<crate::Body>>(
183        &self,
184        request: Request<RequestBody>,
185    ) -> RequestBuilder<'_, Self, Self::ExecuteResponse> {
186        RequestBuilder {
187            http_client_service: self,
188            state: RequestBuilderState::PostBody(request.map(Into::into)),
189            _phantom: std::marker::PhantomData,
190            _mode: std::marker::PhantomData,
191        }
192    }
193
194    fn execute(
195        &self,
196
197        request: Request,
198    ) -> impl Future<Output = Result<Self::ExecuteResponse, Self::ExecuteError>> {
199        Service::serve(self, request)
200    }
201}
202
203/// A trait to try to convert some type into a [`Url`].
204///
205/// This trait is "sealed", such that only types within rama can implement it.
206///
207/// [`Url`]: rama_net::uri::Uri
208pub trait IntoUrl: private::IntoUrlSealed {}
209
210impl IntoUrl for Uri {}
211impl IntoUrl for &str {}
212impl IntoUrl for String {}
213impl IntoUrl for &String {}
214
215/// A trait to try to convert some type into a [`HeaderName`].
216///
217/// This trait is "sealed", such that only types within rama can implement it.
218///
219/// [`HeaderName`]: crate::HeaderName
220pub trait IntoHeaderName: private::IntoHeaderNameSealed {}
221
222impl IntoHeaderName for crate::HeaderName {}
223impl IntoHeaderName for Option<crate::HeaderName> {}
224impl IntoHeaderName for &str {}
225impl IntoHeaderName for String {}
226impl IntoHeaderName for &String {}
227impl IntoHeaderName for &[u8] {}
228
229/// A trait to try to convert some type into a [`HeaderValue`].
230///
231/// This trait is "sealed", such that only types within rama can implement it.
232///
233/// [`HeaderValue`]: crate::HeaderValue
234pub trait IntoHeaderValue: private::IntoHeaderValueSealed {}
235
236impl IntoHeaderValue for crate::HeaderValue {}
237impl IntoHeaderValue for &str {}
238impl IntoHeaderValue for String {}
239impl IntoHeaderValue for &String {}
240impl IntoHeaderValue for &[u8] {}
241
242mod private {
243
244    use rama_http_types::HeaderName;
245
246    use super::*;
247
248    pub trait IntoUrlSealed {
249        fn into_url(self) -> Result<Uri, BoxError>;
250    }
251
252    impl IntoUrlSealed for Uri {
253        fn into_url(self) -> Result<Uri, BoxError> {
254            // any scheme goes: layers such as `FileUriLayer` / `DataUriLayer`
255            // extend a client beyond http, and the connector rejects what it
256            // cannot serve anyway
257            if self.scheme().is_some() {
258                Ok(self)
259            } else {
260                Err(BoxError::from_static_str("Missing scheme in URI"))
261            }
262        }
263    }
264
265    impl IntoUrlSealed for &str {
266        fn into_url(self) -> Result<Uri, BoxError> {
267            match self.parse::<Uri>() {
268                Ok(uri) => uri.into_url(),
269                Err(_) => {
270                    Err(BoxError::from_static_str("invalid url").context_str_field("raw_str", self))
271                }
272            }
273        }
274    }
275
276    impl IntoUrlSealed for String {
277        #[inline(always)]
278        fn into_url(self) -> Result<Uri, BoxError> {
279            self.as_str().into_url()
280        }
281    }
282
283    impl IntoUrlSealed for &String {
284        #[inline(always)]
285        fn into_url(self) -> Result<Uri, BoxError> {
286            self.as_str().into_url()
287        }
288    }
289
290    pub trait IntoHeaderNameSealed {
291        fn into_header_name(self) -> Result<crate::HeaderName, BoxError>;
292    }
293
294    impl IntoHeaderNameSealed for HeaderName {
295        fn into_header_name(self) -> Result<crate::HeaderName, BoxError> {
296            Ok(self)
297        }
298    }
299
300    impl IntoHeaderNameSealed for Option<HeaderName> {
301        fn into_header_name(self) -> Result<crate::HeaderName, BoxError> {
302            match self {
303                Some(name) => Ok(name),
304                None => Err(BoxError::from_static_str("Header name is required")),
305            }
306        }
307    }
308
309    impl IntoHeaderNameSealed for &str {
310        fn into_header_name(self) -> Result<crate::HeaderName, BoxError> {
311            let name = self.parse::<crate::HeaderName>()?;
312            Ok(name)
313        }
314    }
315
316    impl IntoHeaderNameSealed for String {
317        #[inline(always)]
318        fn into_header_name(self) -> Result<crate::HeaderName, BoxError> {
319            self.as_str().into_header_name()
320        }
321    }
322
323    impl IntoHeaderNameSealed for &String {
324        #[inline(always)]
325        fn into_header_name(self) -> Result<crate::HeaderName, BoxError> {
326            self.as_str().into_header_name()
327        }
328    }
329
330    impl IntoHeaderNameSealed for &[u8] {
331        fn into_header_name(self) -> Result<crate::HeaderName, BoxError> {
332            let name = crate::HeaderName::from_bytes(self)?;
333            Ok(name)
334        }
335    }
336
337    pub trait IntoHeaderValueSealed {
338        fn into_header_value(self) -> Result<crate::HeaderValue, BoxError>;
339    }
340
341    impl IntoHeaderValueSealed for crate::HeaderValue {
342        fn into_header_value(self) -> Result<crate::HeaderValue, BoxError> {
343            Ok(self)
344        }
345    }
346
347    impl IntoHeaderValueSealed for &str {
348        fn into_header_value(self) -> Result<crate::HeaderValue, BoxError> {
349            let value = self.parse::<crate::HeaderValue>()?;
350            Ok(value)
351        }
352    }
353
354    impl IntoHeaderValueSealed for String {
355        fn into_header_value(self) -> Result<crate::HeaderValue, BoxError> {
356            self.as_str().into_header_value()
357        }
358    }
359
360    impl IntoHeaderValueSealed for &String {
361        #[inline(always)]
362        fn into_header_value(self) -> Result<crate::HeaderValue, BoxError> {
363            self.as_str().into_header_value()
364        }
365    }
366
367    impl IntoHeaderValueSealed for &[u8] {
368        fn into_header_value(self) -> Result<crate::HeaderValue, BoxError> {
369            let value = crate::HeaderValue::from_bytes(self)?;
370            Ok(value)
371        }
372    }
373
374    pub trait HttpClientExtSealed {}
375
376    impl<S, Body> HttpClientExtSealed for S where
377        S: Service<Request, Output = Response<Body>, Error: Into<BoxError>>
378    {
379    }
380}
381
382/// A builder to construct the properties of a [`Request`].
383///
384/// Constructed using [`HttpClientExt`].
385pub struct RequestBuilder<'a, S, Response, Mode = request_builder_mode::Async> {
386    pub(super) http_client_service: &'a S,
387    pub(super) state: RequestBuilderState,
388    _phantom: std::marker::PhantomData<fn(Response) -> ()>,
389    _mode: std::marker::PhantomData<fn(Mode) -> ()>,
390}
391
392/// A request builder whose terminal operations block the calling thread.
393pub type BlockingRequestBuilder<'a, S, Response> =
394    RequestBuilder<'a, S, Response, request_builder_mode::Blocking>;
395
396/// Request builder execution modes.
397pub mod request_builder_mode {
398    /// Asynchronous terminal request operations.
399    #[derive(Debug)]
400    #[non_exhaustive]
401    pub struct Async;
402
403    /// Blocking terminal request operations.
404    #[derive(Debug)]
405    #[non_exhaustive]
406    pub struct Blocking;
407}
408
409impl<S, Response, Mode> std::fmt::Debug for RequestBuilder<'_, S, Response, Mode>
410where
411    S: std::fmt::Debug,
412{
413    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
414        f.debug_struct("RequestBuilder")
415            .field("http_client_service", &self.http_client_service)
416            .field("state", &self.state)
417            .finish()
418    }
419}
420
421#[derive(Debug)]
422pub(super) enum RequestBuilderState {
423    PreBody(crate::request::Builder),
424    PostBody(crate::Request),
425    Error(BoxError),
426}
427
428pub(super) fn request_builder_state(method: Method, url: impl IntoUrl) -> RequestBuilderState {
429    match private::IntoUrlSealed::into_url(url) {
430        Ok(uri) => {
431            RequestBuilderState::PreBody(crate::request::Builder::new().method(method).uri(uri))
432        }
433        Err(err) => RequestBuilderState::Error(err),
434    }
435}
436
437impl<'a, S, Body, Mode> RequestBuilder<'a, S, Response<Body>, Mode> {
438    pub(super) fn from_state(http_client_service: &'a S, state: RequestBuilderState) -> Self {
439        Self {
440            http_client_service,
441            state,
442            _phantom: std::marker::PhantomData,
443            _mode: std::marker::PhantomData,
444        }
445    }
446
447    /// Add a `Header` to this [`Request`].
448    #[must_use]
449    pub fn header<K, V>(mut self, key: K, value: V) -> Self
450    where
451        K: IntoHeaderName,
452        V: IntoHeaderValue,
453    {
454        match self.state {
455            RequestBuilderState::PreBody(builder) => {
456                let key = match key.into_header_name() {
457                    Ok(key) => key,
458                    Err(err) => {
459                        self.state = RequestBuilderState::Error(err);
460                        return self;
461                    }
462                };
463                let value = match value.into_header_value() {
464                    Ok(value) => value,
465                    Err(err) => {
466                        self.state = RequestBuilderState::Error(err);
467                        return self;
468                    }
469                };
470                self.state = RequestBuilderState::PreBody(builder.header(key, value));
471                self
472            }
473            RequestBuilderState::PostBody(mut request) => {
474                let key = match key.into_header_name() {
475                    Ok(key) => key,
476                    Err(err) => {
477                        self.state = RequestBuilderState::Error(err);
478                        return self;
479                    }
480                };
481                let value = match value.into_header_value() {
482                    Ok(value) => value,
483                    Err(err) => {
484                        self.state = RequestBuilderState::Error(err);
485                        return self;
486                    }
487                };
488                request.headers_mut().append(key, value);
489                self.state = RequestBuilderState::PostBody(request);
490                self
491            }
492            RequestBuilderState::Error(err) => {
493                self.state = RequestBuilderState::Error(err);
494                self
495            }
496        }
497    }
498
499    /// Add a typed [`HeaderEncode`] to this [`Request`].
500    ///
501    /// [`HeaderEncode`]: crate::headers::HeaderEncode
502    #[must_use]
503    pub fn typed_header<H>(self, header: H) -> Self
504    where
505        H: crate::headers::HeaderEncode,
506    {
507        if let Some(value) = header.encode_to_value() {
508            self.header(H::name().clone(), value)
509        } else {
510            self
511        }
512    }
513
514    /// Add all `Headers` from the [`HeaderMap`] to this [`Request`].
515    ///
516    /// [`HeaderMap`]: crate::HeaderMap
517    #[must_use]
518    pub fn headers(mut self, headers: crate::HeaderMap) -> Self {
519        for (key, value) in headers.into_iter() {
520            self = self.header(key, value);
521        }
522        self
523    }
524
525    /// Overwrite a `Header` to this [`Request`].
526    #[must_use]
527    pub fn overwrite_header<K, V>(mut self, key: K, value: V) -> Self
528    where
529        K: IntoHeaderName,
530        V: IntoHeaderValue,
531    {
532        match self.state {
533            RequestBuilderState::PreBody(mut builder) => {
534                // None in case builder has errors
535                if let Some(headers) = builder.headers_mut() {
536                    let key = match key.into_header_name() {
537                        Ok(key) => key,
538                        Err(err) => {
539                            self.state = RequestBuilderState::Error(err);
540                            return self;
541                        }
542                    };
543                    let value = match value.into_header_value() {
544                        Ok(value) => value,
545                        Err(err) => {
546                            self.state = RequestBuilderState::Error(err);
547                            return self;
548                        }
549                    };
550                    _ = headers.insert(key, value);
551                }
552
553                self.state = RequestBuilderState::PreBody(builder);
554                self
555            }
556            RequestBuilderState::PostBody(mut request) => {
557                let key = match key.into_header_name() {
558                    Ok(key) => key,
559                    Err(err) => {
560                        self.state = RequestBuilderState::Error(err);
561                        return self;
562                    }
563                };
564                let value = match value.into_header_value() {
565                    Ok(value) => value,
566                    Err(err) => {
567                        self.state = RequestBuilderState::Error(err);
568                        return self;
569                    }
570                };
571                _ = request.headers_mut().insert(key, value);
572                self.state = RequestBuilderState::PostBody(request);
573                self
574            }
575            RequestBuilderState::Error(err) => {
576                self.state = RequestBuilderState::Error(err);
577                self
578            }
579        }
580    }
581
582    /// Overwrite a typed [`HeaderEncode`] to this [`Request`].
583    ///
584    /// [`HeaderEncode`]: crate::headers::HeaderEncode
585    #[must_use]
586    pub fn overwrite_typed_header<H>(self, header: H) -> Self
587    where
588        H: crate::headers::HeaderEncode,
589    {
590        if let Some(value) = header.encode_to_value() {
591            self.overwrite_header(H::name().clone(), value)
592        } else {
593            self
594        }
595    }
596
597    /// Enable HTTP authentication.
598    #[must_use]
599    pub fn auth(self, credentials: impl Credentials) -> Self {
600        let header = crate::headers::Authorization::new(credentials);
601        self.typed_header(header)
602    }
603
604    /// Adds an extension to this builder
605    #[must_use]
606    pub fn extension<T>(mut self, extension: T) -> Self
607    where
608        T: Extension,
609    {
610        match self.state {
611            RequestBuilderState::PreBody(builder) => {
612                let builder = builder.extension(extension);
613                self.state = RequestBuilderState::PreBody(builder);
614                self
615            }
616            RequestBuilderState::PostBody(request) => {
617                request.extensions().insert(extension);
618                self.state = RequestBuilderState::PostBody(request);
619                self
620            }
621            state @ RequestBuilderState::Error(_) => {
622                self.state = state;
623                self
624            }
625        }
626    }
627
628    /// Get access to the underlying [`Extensions`]
629    ///
630    /// This function will return None if [`Extensions`] are not available,
631    /// or if this builder is in an error state
632    pub fn extensions(&self) -> Option<&Extensions> {
633        match &self.state {
634            RequestBuilderState::PreBody(builder) => builder.extensions(),
635            RequestBuilderState::PostBody(request) => Some(request.extensions()),
636            RequestBuilderState::Error(_) => None,
637        }
638    }
639
640    /// Set the [`Request`]'s [`Body`].
641    ///
642    /// Re-syncs `Content-Length` with the new body: when the body has a
643    /// known exact size, `Content-Length` is set to that value, replacing
644    /// any prior value. When the size is unknown, any pre-set
645    /// `Content-Length` is removed so the request doesn't ship with stale
646    /// framing.
647    ///
648    /// [`Body`]: crate::Body
649    #[must_use]
650    pub fn body<T>(mut self, body: T) -> Self
651    where
652        T: TryInto<crate::Body, Error: Into<BoxError>>,
653    {
654        self.state = match self.state {
655            RequestBuilderState::PreBody(mut builder) => match body.try_into() {
656                Ok(body) => {
657                    sync_content_length_pre_body(&mut builder, &body);
658                    match builder.body(body) {
659                        Ok(req) => RequestBuilderState::PostBody(req),
660                        Err(err) => RequestBuilderState::Error(BoxError::from(err)),
661                    }
662                }
663                Err(err) => RequestBuilderState::Error(err.into()),
664            },
665            RequestBuilderState::PostBody(mut req) => match body.try_into() {
666                Ok(body) => {
667                    sync_content_length_post_body(&mut req, &body);
668                    *req.body_mut() = body;
669                    RequestBuilderState::PostBody(req)
670                }
671                Err(err) => RequestBuilderState::Error(BoxError::from(err.into())),
672            },
673            RequestBuilderState::Error(err) => RequestBuilderState::Error(err),
674        };
675        self
676    }
677
678    /// Set the given value as a URL-Encoded Form [`Body`] in the [`Request`].
679    ///
680    /// Sets `Content-Type: application/x-www-form-urlencoded` if no
681    /// `Content-Type` header is already set. Sets `Content-Length` to the
682    /// encoded body's size, replacing any prior value.
683    ///
684    /// [`Body`]: crate::Body
685    #[must_use]
686    pub fn form<T: serde::Serialize + ?Sized>(mut self, form: &T) -> Self {
687        self.state = match self.state {
688            RequestBuilderState::PreBody(mut builder) => match serde_html_form::to_string(form) {
689                Ok(body) => {
690                    let len = body.len() as u64;
691                    let builder = match builder.headers_mut() {
692                        Some(headers) => {
693                            if !headers.contains_key(crate::header::CONTENT_TYPE) {
694                                headers.insert(
695                                    crate::header::CONTENT_TYPE,
696                                    crate::HeaderValue::from_static(
697                                        "application/x-www-form-urlencoded",
698                                    ),
699                                );
700                            }
701                            headers.insert(
702                                crate::header::CONTENT_LENGTH,
703                                crate::HeaderValue::from(len),
704                            );
705                            builder
706                        }
707                        None => builder
708                            .header(
709                                crate::header::CONTENT_TYPE,
710                                crate::HeaderValue::from_static(
711                                    "application/x-www-form-urlencoded",
712                                ),
713                            )
714                            .header(crate::header::CONTENT_LENGTH, crate::HeaderValue::from(len)),
715                    };
716                    match builder.body(body.into()) {
717                        Ok(req) => RequestBuilderState::PostBody(req),
718                        Err(err) => RequestBuilderState::Error(BoxError::from(err)),
719                    }
720                }
721                Err(err) => RequestBuilderState::Error(BoxError::from(err)),
722            },
723            RequestBuilderState::PostBody(mut req) => match serde_html_form::to_string(form) {
724                Ok(body) => {
725                    let len = body.len() as u64;
726                    if !req.headers().contains_key(crate::header::CONTENT_TYPE) {
727                        req.headers_mut().insert(
728                            crate::header::CONTENT_TYPE,
729                            crate::HeaderValue::from_static("application/x-www-form-urlencoded"),
730                        );
731                    }
732                    req.headers_mut()
733                        .insert(crate::header::CONTENT_LENGTH, crate::HeaderValue::from(len));
734                    *req.body_mut() = body.into();
735                    RequestBuilderState::PostBody(req)
736                }
737                Err(err) => RequestBuilderState::Error(BoxError::from(err)),
738            },
739            RequestBuilderState::Error(err) => RequestBuilderState::Error(err),
740        };
741        self
742    }
743
744    /// Set the given value as a JSON [`Body`] in the [`Request`].
745    ///
746    /// Sets `Content-Type: application/json` if no `Content-Type` header is
747    /// already set. Sets `Content-Length` to the encoded body's size,
748    /// replacing any prior value.
749    ///
750    /// [`Body`]: crate::Body
751    #[must_use]
752    pub fn json<T: serde::Serialize + ?Sized>(mut self, json: &T) -> Self {
753        self.state = match self.state {
754            RequestBuilderState::PreBody(mut builder) => match serde_json::to_vec(json) {
755                Ok(body) => {
756                    let len = body.len() as u64;
757                    let builder = match builder.headers_mut() {
758                        Some(headers) => {
759                            if !headers.contains_key(crate::header::CONTENT_TYPE) {
760                                headers.insert(
761                                    crate::header::CONTENT_TYPE,
762                                    crate::HeaderValue::from_static("application/json"),
763                                );
764                            }
765                            headers.insert(
766                                crate::header::CONTENT_LENGTH,
767                                crate::HeaderValue::from(len),
768                            );
769                            builder
770                        }
771                        None => builder
772                            .header(
773                                crate::header::CONTENT_TYPE,
774                                crate::HeaderValue::from_static("application/json"),
775                            )
776                            .header(crate::header::CONTENT_LENGTH, crate::HeaderValue::from(len)),
777                    };
778                    match builder.body(body.into()) {
779                        Ok(req) => RequestBuilderState::PostBody(req),
780                        Err(err) => RequestBuilderState::Error(BoxError::from(err)),
781                    }
782                }
783                Err(err) => RequestBuilderState::Error(BoxError::from(err)),
784            },
785            RequestBuilderState::PostBody(mut req) => match serde_json::to_vec(json) {
786                Ok(body) => {
787                    let len = body.len() as u64;
788                    if !req.headers().contains_key(crate::header::CONTENT_TYPE) {
789                        req.headers_mut().insert(
790                            crate::header::CONTENT_TYPE,
791                            crate::HeaderValue::from_static("application/json"),
792                        );
793                    }
794                    req.headers_mut()
795                        .insert(crate::header::CONTENT_LENGTH, crate::HeaderValue::from(len));
796                    *req.body_mut() = body.into();
797                    RequestBuilderState::PostBody(req)
798                }
799                Err(err) => RequestBuilderState::Error(BoxError::from(err)),
800            },
801            RequestBuilderState::Error(err) => RequestBuilderState::Error(err),
802        };
803        self
804    }
805
806    /// Set a [`multipart::Form`](super::multipart::Form) as the body of this
807    /// [`Request`].
808    ///
809    /// Sets the `Content-Type` header to
810    /// `multipart/form-data; boundary={boundary}` and, when every part has a
811    /// known size, the `Content-Length` header. An existing `Content-Type`
812    /// header is overwritten.
813    #[cfg(feature = "multipart")]
814    #[must_use]
815    pub fn multipart(mut self, form: super::multipart::Form) -> Self {
816        let content_type = form.content_type();
817        let content_length = form.content_length();
818        let body = form.into_body();
819        self.state = match self.state {
820            RequestBuilderState::PreBody(mut builder) => {
821                if let Some(headers) = builder.headers_mut() {
822                    headers.insert(crate::header::CONTENT_TYPE, content_type);
823                    sync_content_length_in_map(headers, content_length);
824                }
825                // If `headers_mut` is None the builder is in an error
826                // state; `builder.body(...)` will surface that error.
827                match builder.body(body) {
828                    Ok(req) => RequestBuilderState::PostBody(req),
829                    Err(err) => RequestBuilderState::Error(BoxError::from(err)),
830                }
831            }
832            RequestBuilderState::PostBody(mut req) => {
833                let headers = req.headers_mut();
834                headers.insert(crate::header::CONTENT_TYPE, content_type);
835                sync_content_length_in_map(headers, content_length);
836                *req.body_mut() = body;
837                RequestBuilderState::PostBody(req)
838            }
839            RequestBuilderState::Error(err) => RequestBuilderState::Error(err),
840        };
841        self
842    }
843
844    /// Set the given bytes as an `application/octet-stream` [`Body`] in the [`Request`].
845    ///
846    /// Sets `Content-Type: application/octet-stream` if no `Content-Type`
847    /// header is already set on the request. Sets `Content-Length` to the
848    /// payload's size, replacing any prior value.
849    ///
850    /// [`Body`]: crate::Body
851    #[must_use]
852    pub fn octet_stream<T: Into<rama_core::bytes::Bytes>>(mut self, bytes: T) -> Self {
853        let bytes = bytes.into();
854        let len = bytes.len() as u64;
855        self.state = match self.state {
856            RequestBuilderState::PreBody(mut builder) => {
857                let builder = match builder.headers_mut() {
858                    Some(headers) => {
859                        if !headers.contains_key(crate::header::CONTENT_TYPE) {
860                            headers.insert(
861                                crate::header::CONTENT_TYPE,
862                                crate::HeaderValue::from_static("application/octet-stream"),
863                            );
864                        }
865                        headers
866                            .insert(crate::header::CONTENT_LENGTH, crate::HeaderValue::from(len));
867                        builder
868                    }
869                    None => builder
870                        .header(
871                            crate::header::CONTENT_TYPE,
872                            crate::HeaderValue::from_static("application/octet-stream"),
873                        )
874                        .header(crate::header::CONTENT_LENGTH, crate::HeaderValue::from(len)),
875                };
876                match builder.body(bytes.into()) {
877                    Ok(req) => RequestBuilderState::PostBody(req),
878                    Err(err) => RequestBuilderState::Error(BoxError::from(err)),
879                }
880            }
881            RequestBuilderState::PostBody(mut req) => {
882                if !req.headers().contains_key(crate::header::CONTENT_TYPE) {
883                    req.headers_mut().insert(
884                        crate::header::CONTENT_TYPE,
885                        crate::HeaderValue::from_static("application/octet-stream"),
886                    );
887                }
888                req.headers_mut()
889                    .insert(crate::header::CONTENT_LENGTH, crate::HeaderValue::from(len));
890                *req.body_mut() = bytes.into();
891                RequestBuilderState::PostBody(req)
892            }
893            RequestBuilderState::Error(err) => RequestBuilderState::Error(err),
894        };
895        self
896    }
897
898    /// Set the http [`Version`] of this [`Request`].
899    ///
900    /// [`Version`]: crate::Version
901    #[must_use]
902    pub fn version(mut self, version: crate::Version) -> Self {
903        match self.state {
904            RequestBuilderState::PreBody(builder) => {
905                self.state = RequestBuilderState::PreBody(builder.version(version));
906                self
907            }
908            RequestBuilderState::PostBody(mut request) => {
909                *request.version_mut() = version;
910                self.state = RequestBuilderState::PostBody(request);
911                self
912            }
913            RequestBuilderState::Error(err) => {
914                self.state = RequestBuilderState::Error(err);
915                self
916            }
917        }
918    }
919
920    /// Construct the [`Request`] without sending it.
921    ///
922    /// # Errors
923    ///
924    /// This method fails if there was an error while constructing the [`Request`].
925    pub fn build(self) -> Result<Request, BoxError> {
926        Ok(match self.state {
927            RequestBuilderState::PreBody(builder) => builder.body(crate::Body::empty())?,
928            RequestBuilderState::PostBody(request) => request,
929            RequestBuilderState::Error(err) => return Err(err),
930        })
931    }
932}
933
934impl<S, Body> RequestBuilder<'_, S, Response<Body>>
935where
936    S: Service<Request, Output = Response<Body>, Error: Into<BoxError>>,
937{
938    /// Constructs the [`Request`].
939    ///
940    /// # Errors
941    ///
942    /// This method fails if there was an error while constructing the [`Request`].
943    pub async fn try_into_request(self) -> Result<Request, BoxError> {
944        self.build()
945    }
946
947    /// Constructs the [`Request`] and sends it to the target [`Uri`], returning a future [`Response`].
948    ///
949    /// # Errors
950    ///
951    /// This method fails if there was an error while sending [`Request`].
952    pub async fn send(self) -> Result<Response<Body>, BoxError> {
953        let http_client_service = self.http_client_service;
954        let request = self.build()?;
955
956        let uri = request.uri().clone();
957        match http_client_service.serve(request).await {
958            Ok(response) => Ok(response),
959            Err(err) => Err(err.into().context(uri)),
960        }
961    }
962
963    /// Constructs the [`Request`] and sends it to the target [`Uri`],
964    /// returning a [`Response`] with a timeout that aborts with an error if
965    /// the roundtrip does not complete in time.
966    ///
967    /// Note that this does not apply to the collection of the response payload,
968    /// if there is any at all. The roundtrip is considered complete
969    /// after the response headers are received.
970    ///
971    /// # Errors
972    ///
973    /// This method fails if there was an error while sending [`Request`].
974    pub async fn send_with_timeout(self, t: Duration) -> Result<Response<Body>, BoxError> {
975        let http_client_service = self.http_client_service;
976        let request = self.build()?;
977
978        let uri = request.uri().clone();
979        match tokio::time::timeout(t, http_client_service.serve(request)).await {
980            Ok(Ok(response)) => Ok(response),
981            Err(err) => Err(err
982                .context("roundtrip timeout reached")
983                .context_debug_field("timeout", t)),
984            Ok(Err(err)) => Err(err.into().context(uri)),
985        }
986    }
987}
988
989/// Synchronise `Content-Length` with the new body on a pre-body request
990/// builder.
991///
992/// If the body has a known exact size, set `Content-Length` to that value
993/// (replacing any prior). If the size is unknown, remove any pre-set
994/// `Content-Length` so the request doesn't ship with stale framing.
995fn sync_content_length_pre_body(builder: &mut crate::request::Builder, body: &crate::Body) {
996    let Some(headers) = builder.headers_mut() else {
997        return;
998    };
999    let exact = body.size_hint().exact();
1000    sync_content_length_in_map(headers, exact);
1001}
1002
1003/// Synchronise `Content-Length` with the new body on a request that has
1004/// already entered the post-body state.
1005fn sync_content_length_post_body<B>(req: &mut crate::Request<B>, body: &crate::Body) {
1006    let exact = body.size_hint().exact();
1007    sync_content_length_in_map(req.headers_mut(), exact);
1008}
1009
1010fn sync_content_length_in_map(headers: &mut crate::HeaderMap, exact: Option<u64>) {
1011    match exact {
1012        Some(len) => {
1013            headers.insert(crate::header::CONTENT_LENGTH, crate::HeaderValue::from(len));
1014        }
1015        None => {
1016            headers.remove(crate::header::CONTENT_LENGTH);
1017        }
1018    }
1019}
1020
1021#[cfg(test)]
1022mod test {
1023    use rama_http_types::StatusCode;
1024
1025    use super::*;
1026    use crate::{
1027        StreamingBody,
1028        layer::{
1029            required_header::AddRequiredRequestHeadersLayer,
1030            retry::{ManagedPolicy, RetryLayer},
1031            trace::TraceLayer,
1032        },
1033        service::web::response::IntoResponse,
1034    };
1035    use rama_core::{
1036        layer::{Layer, MapResultLayer},
1037        service::{BoxService, service_fn},
1038    };
1039    use rama_utils::backoff::ExponentialBackoff;
1040    use std::convert::Infallible;
1041
1042    async fn fake_client_fn<Body>(request: Request<Body>) -> Result<Response, Infallible>
1043    where
1044        Body: StreamingBody<Data: Send + 'static, Error: Send + 'static> + Send + 'static,
1045    {
1046        let ua = request.headers().get(crate::header::USER_AGENT).unwrap();
1047        assert_eq!(
1048            ua.to_str().unwrap(),
1049            format!("{}/{}", rama_utils::info::NAME, rama_utils::info::VERSION)
1050        );
1051
1052        Ok(StatusCode::OK.into_response())
1053    }
1054
1055    fn map_internal_client_error<E, Body>(
1056        result: Result<Response<Body>, E>,
1057    ) -> Result<Response, rama_core::error::BoxError>
1058    where
1059        E: Into<rama_core::error::BoxError>,
1060        Body: StreamingBody<Data = rama_core::bytes::Bytes, Error: Into<BoxError>>
1061            + Send
1062            + Sync
1063            + 'static,
1064    {
1065        match result {
1066            Ok(response) => Ok(response.map(crate::Body::new)),
1067            Err(err) => Err(err.into()),
1068        }
1069    }
1070
1071    type HttpClient = BoxService<Request, Response, BoxError>;
1072
1073    fn client() -> HttpClient {
1074        let builder = (
1075            MapResultLayer::new(map_internal_client_error),
1076            TraceLayer::new_for_http(),
1077        );
1078
1079        #[cfg(feature = "compression")]
1080        let builder = (
1081            builder,
1082            crate::layer::decompression::DecompressionLayer::new(),
1083        );
1084
1085        (
1086            builder,
1087            RetryLayer::new(ManagedPolicy::default().with_backoff(ExponentialBackoff::default())),
1088            AddRequiredRequestHeadersLayer::default(),
1089        )
1090            .into_layer(service_fn(fake_client_fn))
1091            .boxed()
1092    }
1093
1094    #[tokio::test]
1095    async fn test_client_happy_path() {
1096        let response = client().get("http://127.0.0.1:8080").send().await.unwrap();
1097        assert_eq!(response.status(), StatusCode::OK);
1098    }
1099
1100    #[tokio::test]
1101    async fn test_client_with_timeout_happy_path() {
1102        let response = client()
1103            .get("http://127.0.0.1:8080")
1104            .send_with_timeout(Duration::from_secs(8))
1105            .await
1106            .unwrap();
1107        assert_eq!(response.status(), StatusCode::OK);
1108    }
1109
1110    #[tokio::test]
1111    async fn test_hanging_client_with_timeout() {
1112        let client = service_fn(async |_req| {
1113            _ = std::future::pending::<()>().await;
1114            Ok::<_, Infallible>(StatusCode::OK.into_response())
1115        });
1116
1117        let _ = client
1118            .get("http://127.0.0.1:8080")
1119            .send_with_timeout(Duration::from_micros(1))
1120            .await
1121            .unwrap_err();
1122    }
1123
1124    /// Capture the request the builder would send and return it as the
1125    /// response body so tests can inspect the outgoing headers.
1126    async fn echo_headers_client_fn(request: Request<crate::Body>) -> Result<Response, Infallible> {
1127        let mut headers_lines = Vec::new();
1128        for (name, value) in request.headers() {
1129            headers_lines.push(format!(
1130                "{}: {}",
1131                name.as_str(),
1132                value.to_str().unwrap_or("<bin>")
1133            ));
1134        }
1135        let body = headers_lines.join("\n");
1136        Ok(crate::Response::new(crate::Body::from(body)))
1137    }
1138
1139    fn echo_client() -> HttpClient {
1140        let builder = (
1141            MapResultLayer::new(map_internal_client_error),
1142            TraceLayer::new_for_http(),
1143        );
1144        (builder,)
1145            .into_layer(service_fn(echo_headers_client_fn))
1146            .boxed()
1147    }
1148
1149    async fn dump_headers(resp: Response) -> String {
1150        use crate::body::util::BodyExt as _;
1151        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
1152        String::from_utf8(bytes.to_vec()).unwrap()
1153    }
1154
1155    #[tokio::test]
1156    async fn test_any_scheme_reaches_the_inner_service() {
1157        // scheme-extending layers (file://, data:, ...) live above the
1158        // client, so IntoUrl must not gatekeep on the scheme
1159        let uri_client = || {
1160            (MapResultLayer::new(map_internal_client_error),)
1161                .into_layer(service_fn(async |req: Request| {
1162                    Ok::<_, Infallible>(req.uri().to_string().into_response())
1163                }))
1164                .boxed()
1165        };
1166
1167        for uri in [
1168            "wibble://example.com/x",
1169            "file:///tmp/pac.js",
1170            "data:,hello",
1171            "urn:isbn:0451450523",
1172        ] {
1173            let resp = uri_client().get(uri).send().await.unwrap();
1174            assert_eq!(dump_headers(resp).await, uri);
1175        }
1176
1177        // a scheme-less URI stays an error: there is nothing to route on
1178        let err = uri_client().get("/just/a/path").send().await.unwrap_err();
1179        assert!(err.to_string().contains("Missing scheme"), "{err}");
1180    }
1181
1182    #[tokio::test]
1183    async fn test_form_helper_sets_content_length_and_overwrites_stale() {
1184        let resp = echo_client()
1185            .post("http://x/")
1186            .header(crate::header::CONTENT_LENGTH, "9999") // stale, should be overwritten
1187            .form(&[("k", "vvv")])
1188            .send()
1189            .await
1190            .unwrap();
1191        let dump = dump_headers(resp).await;
1192        // serde_html_form encodes `[("k", "vvv")]` as `k=vvv` (5 bytes).
1193        assert!(
1194            dump.contains("content-length: 5"),
1195            "expected content-length: 5 in:\n{dump}"
1196        );
1197        assert!(
1198            dump.contains("content-type: application/x-www-form-urlencoded"),
1199            "missing CT in:\n{dump}"
1200        );
1201    }
1202
1203    #[tokio::test]
1204    async fn test_json_helper_sets_content_length_and_overwrites_stale() {
1205        let resp = echo_client()
1206            .post("http://x/")
1207            .header(crate::header::CONTENT_LENGTH, "9999")
1208            .json(&serde_json::json!({"a": 1}))
1209            .send()
1210            .await
1211            .unwrap();
1212        let dump = dump_headers(resp).await;
1213        // `{"a":1}` is 7 bytes.
1214        assert!(
1215            dump.contains("content-length: 7"),
1216            "expected content-length: 7 in:\n{dump}"
1217        );
1218        assert!(dump.contains("content-type: application/json"));
1219    }
1220
1221    #[tokio::test]
1222    async fn test_octet_stream_helper_sets_content_length_and_overwrites_stale() {
1223        let resp = echo_client()
1224            .post("http://x/")
1225            .header(crate::header::CONTENT_LENGTH, "9999")
1226            .octet_stream(b"abcd".as_slice())
1227            .send()
1228            .await
1229            .unwrap();
1230        let dump = dump_headers(resp).await;
1231        assert!(
1232            dump.contains("content-length: 4"),
1233            "expected content-length: 4 in:\n{dump}"
1234        );
1235        assert!(dump.contains("content-type: application/octet-stream"));
1236    }
1237
1238    #[tokio::test]
1239    async fn test_body_helper_clears_content_length_for_streaming_body() {
1240        // A streaming body with no exact size_hint must not ship with the
1241        // stale Content-Length the user pre-set.
1242        use rama_core::futures::stream;
1243        let stream = stream::iter([Ok::<_, BoxError>(rama_core::bytes::Bytes::from_static(
1244            b"chunk",
1245        ))]);
1246        let body = crate::Body::from_stream(stream);
1247
1248        let resp = echo_client()
1249            .post("http://x/")
1250            .header(crate::header::CONTENT_LENGTH, "9999")
1251            .body(body)
1252            .send()
1253            .await
1254            .unwrap();
1255        let dump = dump_headers(resp).await;
1256        assert!(
1257            !dump.to_ascii_lowercase().contains("content-length:"),
1258            "stale content-length leaked into:\n{dump}"
1259        );
1260    }
1261}
rama_http/service/client/ext.rs

rama_http/service/client/
ext.rs
