fizzy_sdk/http/
mod.rs

1//! The HTTP layer the client sends on, and the seam for replacing it.
2//!
3//! Everything the SDK sends goes out through one [`HttpClient`]. The one it ships,
4//! [`ReqwestClient`], is what [`crate::Client::new`] builds; an application that already
5//! has an HTTP stack — a mobile shell on the platform's own, a test on a canned answer —
6//! implements the trait and hands it to [`crate::ClientBuilder::http_client`].
7//!
8//! The request and response types are the `http` crate's, re-exported here so a caller
9//! needs no dependency of its own to name a [`Method`] or read a [`StatusCode`].
10
11use std::pin::Pin;
12
13use async_trait::async_trait;
14use bytes::{Bytes, BytesMut};
15use futures_util::{Stream, StreamExt, stream};
16
17use crate::error::Error;
18
19pub use ::http::header::{self, HeaderMap, HeaderName, HeaderValue};
20pub use ::http::{Method, Request, Response, StatusCode, Version};
21
22#[cfg(feature = "reqwest")]
23mod reqwest;
24#[cfg(feature = "reqwest")]
25pub use self::reqwest::ReqwestClient;
26
27/// Sends one HTTP request and answers with the response, its body still unread.
28///
29/// The Go SDK takes an `*http.Client` and leaves the transport to it. Rust takes a trait
30/// instead: the shells this SDK exists for bring the platform's own HTTP stack, and a
31/// second one in the binary is the wrong price for a client library. The SDK does the rest
32/// above this seam — credentials, retries, redirects, the response cache, the body caps —
33/// so an implementation is only a transport.
34///
35/// An implementation **must not follow redirects**. The SDK follows them itself, so it can
36/// keep credentials on the Fizzy origin and read the `Location` a form request was made for.
37/// One that follows them anyway loses that `Location`, and the SDK cannot tell that it did.
38///
39/// Timeouts belong to the implementation, since the SDK has no way to interrupt a transport
40/// it does not know. A failure to get an answer at all — no connection, a timeout, a broken
41/// stream — is [`Error::network`]; a response with any status is `Ok`.
42#[async_trait]
43pub trait HttpClient: Send + Sync {
44    /// Sends the request and answers what came back, body unread.
45    async fn send(&self, request: Request<Bytes>) -> Result<Response<Body>, Error>;
46}
47
48/// A response body as it arrives, read once.
49///
50/// An implementation builds one with [`Body::from_stream`], passing along the length the
51/// transport knows so a body declared past the caller's cap is refused before a byte of it
52/// is read. The SDK reads it with [`Body::chunk`] or [`Body::collect`].
53pub struct Body {
54    stream: Pin<Box<dyn Stream<Item = Result<Bytes, Error>> + Send>>,
55    content_length: Option<u64>,
56}
57
58impl Body {
59    /// A body over a stream of chunks, with the length the transport declared, if any.
60    pub fn from_stream(
61        stream: impl Stream<Item = Result<Bytes, Error>> + Send + 'static,
62        content_length: Option<u64>,
63    ) -> Body {
64        Body {
65            stream: Box::pin(stream),
66            content_length,
67        }
68    }
69
70    /// No body at all.
71    pub fn empty() -> Body {
72        Body::from(Bytes::new())
73    }
74
75    /// What the transport declared the body's length to be, when it declared one.
76    pub fn content_length(&self) -> Option<u64> {
77        self.content_length
78    }
79
80    /// The next piece of the body, or `None` once it has all arrived.
81    pub async fn chunk(&mut self) -> Result<Option<Bytes>, Error> {
82        self.stream.next().await.transpose()
83    }
84
85    /// Reads the body whole, up to `limit` bytes, and answers `too_large` on the first byte
86    /// past. A body exactly at the limit reads whole; one declared past it never starts.
87    pub async fn collect(
88        mut self,
89        limit: usize,
90        too_large: impl Fn() -> Error,
91    ) -> Result<Bytes, Error> {
92        if self
93            .content_length
94            .is_some_and(|length| length > limit as u64)
95        {
96            return Err(too_large());
97        }
98        let mut body = BytesMut::new();
99        while let Some(chunk) = self.chunk().await? {
100            if body.len() + chunk.len() > limit {
101                return Err(too_large());
102            }
103            body.extend_from_slice(&chunk);
104        }
105        Ok(body.freeze())
106    }
107
108    /// Reads the first `limit` bytes of the body and stops there, leaving the rest unread.
109    pub async fn prefix(mut self, limit: usize) -> Result<Bytes, Error> {
110        let mut body = BytesMut::new();
111        while body.len() < limit
112            && let Some(chunk) = self.chunk().await?
113        {
114            let room = limit - body.len();
115            body.extend_from_slice(&chunk[..chunk.len().min(room)]);
116        }
117        Ok(body.freeze())
118    }
119}
120
121impl From<Bytes> for Body {
122    fn from(bytes: Bytes) -> Body {
123        let content_length = Some(bytes.len() as u64);
124        Body::from_stream(stream::once(async move { Ok(bytes) }), content_length)
125    }
126}
127
128impl From<Vec<u8>> for Body {
129    fn from(bytes: Vec<u8>) -> Body {
130        Body::from(Bytes::from(bytes))
131    }
132}
133
134impl From<&'static str> for Body {
135    fn from(text: &'static str) -> Body {
136        Body::from(Bytes::from_static(text.as_bytes()))
137    }
138}
139
140impl std::fmt::Debug for Body {
141    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
142        f.debug_struct("Body")
143            .field("content_length", &self.content_length)
144            .finish_non_exhaustive()
145    }
146}
147
148#[cfg(test)]
149#[allow(clippy::unwrap_used)]
150mod tests {
151    use super::*;
152
153    fn too_large() -> Error {
154        Error::api(0, "too large")
155    }
156
157    #[tokio::test]
158    async fn collects_a_body_up_to_the_limit() {
159        let body = Body::from(Bytes::from_static(b"hello"));
160        assert_eq!(body.content_length(), Some(5));
161        assert_eq!(body.collect(5, too_large).await.unwrap(), "hello");
162    }
163
164    #[tokio::test]
165    async fn refuses_a_body_declared_past_the_limit_before_reading_it() {
166        let body = Body::from_stream(
167            stream::once(async { panic!("the body should never be read") }),
168            Some(6),
169        );
170        assert_eq!(
171            body.collect(5, too_large).await.unwrap_err().to_string(),
172            "too large"
173        );
174    }
175
176    #[tokio::test]
177    async fn refuses_an_undeclared_body_on_the_first_byte_past_the_limit() {
178        let chunks = stream::iter([
179            Ok(Bytes::from_static(b"hel")),
180            Ok(Bytes::from_static(b"lo!")),
181        ]);
182        let body = Body::from_stream(chunks, None);
183        assert_eq!(
184            body.collect(5, too_large).await.unwrap_err().to_string(),
185            "too large"
186        );
187    }
188
189    #[tokio::test]
190    async fn a_prefix_stops_at_the_limit_without_reading_further() {
191        let chunks = stream::iter([
192            Ok(Bytes::from_static(b"hel")),
193            Ok(Bytes::from_static(b"lo!")),
194            Err(Error::api(0, "never reached")),
195        ]);
196        let body = Body::from_stream(chunks, None);
197        assert_eq!(body.prefix(5).await.unwrap(), "hello");
198    }
199
200    #[tokio::test]
201    async fn a_failing_stream_fails_the_read() {
202        let chunks = stream::iter([
203            Ok(Bytes::from_static(b"hel")),
204            Err(Error::api(0, "cut off")),
205        ]);
206        let body = Body::from_stream(chunks, None);
207        assert_eq!(
208            body.collect(100, too_large).await.unwrap_err().to_string(),
209            "cut off"
210        );
211    }
212}
fizzy_sdk/http/mod.rs

fizzy_sdk/http/
mod.rs
