hey_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")]
25#[cfg_attr(docsrs, doc(cfg(feature = "reqwest")))]
26pub use self::reqwest::ReqwestClient;
27
28/// Sends one HTTP request and answers with the response, its body still unread.
29///
30/// The Go SDK takes an `*http.Client` and leaves the transport to it. Rust takes a trait
31/// instead: the shells this SDK exists for bring the platform's own HTTP stack, and a
32/// second one in the binary is the wrong price for a client library. The SDK does the rest
33/// above this seam — credentials, retries, redirects, the response cache, the body caps —
34/// so an implementation is only a transport.
35///
36/// An implementation **must not follow redirects**. The SDK follows them itself, so it can
37/// keep credentials on the HEY origin and read the `Location` a form request was made for.
38/// One that follows them anyway loses that `Location`, and the SDK cannot tell that it did.
39///
40/// Timeouts belong to the implementation, since the SDK has no way to interrupt a transport
41/// it does not know. A failure to get an answer at all — no connection, a timeout, a broken
42/// stream — is [`Error::network`]; a response with any status is `Ok`.
43#[async_trait]
44pub trait HttpClient: Send + Sync {
45    /// Sends the request and answers with whatever came back, redirects included.
46    async fn send(&self, request: Request<Bytes>) -> Result<Response<Body>, Error>;
47}
48
49/// A response body as it arrives, read once.
50///
51/// An implementation builds one with [`Body::from_stream`], passing along the length the
52/// transport knows so a body declared past the caller's cap is refused before a byte of it
53/// is read. The SDK reads it with [`Body::chunk`] or [`Body::collect`].
54pub struct Body {
55    stream: Pin<Box<dyn Stream<Item = Result<Bytes, Error>> + Send>>,
56    content_length: Option<u64>,
57}
58
59impl Body {
60    /// A body over the transport's stream of chunks, with the length it declared, if any.
61    pub fn from_stream(
62        stream: impl Stream<Item = Result<Bytes, Error>> + Send + 'static,
63        content_length: Option<u64>,
64    ) -> Body {
65        Body {
66            stream: Box::pin(stream),
67            content_length,
68        }
69    }
70
71    /// A body with nothing in it, for an answer that carries none.
72    pub fn empty() -> Body {
73        Body::from(Bytes::new())
74    }
75
76    /// What the transport declared the body's length to be, when it declared one.
77    pub fn content_length(&self) -> Option<u64> {
78        self.content_length
79    }
80
81    /// The next piece of the body, or `None` once it has all arrived.
82    pub async fn chunk(&mut self) -> Result<Option<Bytes>, Error> {
83        self.stream.next().await.transpose()
84    }
85
86    /// Reads the body whole, up to `limit` bytes, and answers `too_large` on the first byte
87    /// past. A body exactly at the limit reads whole; one declared past it never starts.
88    pub async fn collect(
89        mut self,
90        limit: usize,
91        too_large: impl Fn() -> Error,
92    ) -> Result<Bytes, Error> {
93        if self
94            .content_length
95            .is_some_and(|length| length > limit as u64)
96        {
97            return Err(too_large());
98        }
99        let mut body = BytesMut::new();
100        while let Some(chunk) = self.chunk().await? {
101            if body.len() + chunk.len() > limit {
102                return Err(too_large());
103            }
104            body.extend_from_slice(&chunk);
105        }
106        Ok(body.freeze())
107    }
108}
109
110impl From<Bytes> for Body {
111    fn from(bytes: Bytes) -> Body {
112        let content_length = Some(bytes.len() as u64);
113        Body::from_stream(stream::once(async move { Ok(bytes) }), content_length)
114    }
115}
116
117impl From<Vec<u8>> for Body {
118    fn from(bytes: Vec<u8>) -> Body {
119        Body::from(Bytes::from(bytes))
120    }
121}
122
123impl From<&'static str> for Body {
124    fn from(text: &'static str) -> Body {
125        Body::from(Bytes::from_static(text.as_bytes()))
126    }
127}
128
129impl std::fmt::Debug for Body {
130    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
131        f.debug_struct("Body")
132            .field("content_length", &self.content_length)
133            .finish_non_exhaustive()
134    }
135}
136
137#[cfg(test)]
138mod tests {
139    use super::*;
140
141    fn too_large() -> Error {
142        Error::api(0, "too large")
143    }
144
145    #[tokio::test]
146    async fn collects_a_body_up_to_the_limit() {
147        let body = Body::from(Bytes::from_static(b"hello"));
148        assert_eq!(body.content_length(), Some(5));
149        assert_eq!(body.collect(5, too_large).await.unwrap(), "hello");
150    }
151
152    #[tokio::test]
153    async fn refuses_a_body_declared_past_the_limit_before_reading_it() {
154        let body = Body::from_stream(
155            stream::once(async { panic!("the body should never be read") }),
156            Some(6),
157        );
158        assert_eq!(
159            body.collect(5, too_large).await.unwrap_err().to_string(),
160            "too large"
161        );
162    }
163
164    #[tokio::test]
165    async fn refuses_an_undeclared_body_on_the_first_byte_past_the_limit() {
166        let chunks = stream::iter([
167            Ok(Bytes::from_static(b"hel")),
168            Ok(Bytes::from_static(b"lo!")),
169        ]);
170        let body = Body::from_stream(chunks, None);
171        assert_eq!(
172            body.collect(5, too_large).await.unwrap_err().to_string(),
173            "too large"
174        );
175    }
176
177    #[tokio::test]
178    async fn a_failing_stream_fails_the_read() {
179        let chunks = stream::iter([
180            Ok(Bytes::from_static(b"hel")),
181            Err(Error::api(0, "cut off")),
182        ]);
183        let body = Body::from_stream(chunks, None);
184        assert_eq!(
185            body.collect(100, too_large).await.unwrap_err().to_string(),
186            "cut off"
187        );
188    }
189}
hey_sdk/http/mod.rs

hey_sdk/http/
mod.rs
