rig_core/http_client/
mod.rs

1use crate::http_client::sse::BoxedStream;
2use bytes::Bytes;
3pub use http::{HeaderMap, HeaderValue, Method, Request, Response, Uri, request::Builder};
4use http::{HeaderName, StatusCode};
5use reqwest::Body;
6pub mod multipart;
7pub mod retry;
8pub mod sse;
9use crate::wasm_compat::*;
10pub use multipart::MultipartForm;
11pub use reqwest::Client as ReqwestClient;
12use std::pin::Pin;
13
14#[derive(Debug, thiserror::Error)]
15pub enum Error {
16    #[error("Http error: {0}")]
17    Protocol(#[from] http::Error),
18    #[error("Invalid status code: {0}")]
19    InvalidStatusCode(StatusCode),
20    #[error("Invalid status code {0} with message: {1}")]
21    InvalidStatusCodeWithMessage(StatusCode, String),
22    #[error("Header value outside of legal range: {0}")]
23    InvalidHeaderValue(#[from] http::header::InvalidHeaderValue),
24    #[error("Request in error state, cannot access headers")]
25    NoHeaders,
26    #[error("Stream ended")]
27    StreamEnded,
28    #[error("Invalid content type was returned: {0:?}")]
29    InvalidContentType(HeaderValue),
30    #[cfg(not(target_family = "wasm"))]
31    #[error("Http client error: {0}")]
32    Instance(#[from] Box<dyn std::error::Error + Send + Sync + 'static>),
33
34    #[cfg(target_family = "wasm")]
35    #[error("Http client error: {0}")]
36    Instance(#[from] Box<dyn std::error::Error + 'static>),
37}
38
39impl Error {
40    pub(crate) fn non_success_status(&self) -> Option<StatusCode> {
41        match self {
42            Self::InvalidStatusCode(status) | Self::InvalidStatusCodeWithMessage(status, _) => {
43                Some(*status)
44            }
45            _ => None,
46        }
47    }
48
49    pub(crate) fn non_success_body(&self) -> Option<&str> {
50        match self {
51            Self::InvalidStatusCodeWithMessage(_, body) => Some(body.as_str()),
52            _ => None,
53        }
54    }
55}
56
57pub type Result<T> = std::result::Result<T, Error>;
58
59#[cfg(not(target_family = "wasm"))]
60pub(crate) fn instance_error<E: std::error::Error + Send + Sync + 'static>(error: E) -> Error {
61    Error::Instance(error.into())
62}
63
64#[cfg(target_family = "wasm")]
65fn instance_error<E: std::error::Error + 'static>(error: E) -> Error {
66    Error::Instance(error.into())
67}
68
69async fn non_success_status_error(response: reqwest::Response) -> Error {
70    let status = response.status();
71    let message = response
72        .text()
73        .await
74        .unwrap_or_else(|error| format!("failed to read error response body: {error}"));
75    Error::InvalidStatusCodeWithMessage(status, message)
76}
77
78pub type LazyBytes = WasmBoxedFuture<'static, Result<Bytes>>;
79pub type LazyBody<T> = WasmBoxedFuture<'static, Result<T>>;
80
81pub type StreamingResponse = Response<BoxedStream>;
82
83#[derive(Debug, Clone, Copy)]
84pub struct NoBody;
85
86impl From<NoBody> for Bytes {
87    fn from(_: NoBody) -> Self {
88        Bytes::new()
89    }
90}
91
92impl From<NoBody> for Body {
93    fn from(_: NoBody) -> Self {
94        reqwest::Body::default()
95    }
96}
97
98pub async fn text(response: Response<LazyBody<Vec<u8>>>) -> Result<String> {
99    let text = response.into_body().await?;
100    Ok(String::from(String::from_utf8_lossy(&text)))
101}
102
103pub fn make_auth_header(key: impl AsRef<str>) -> Result<(HeaderName, HeaderValue)> {
104    Ok((
105        http::header::AUTHORIZATION,
106        HeaderValue::from_str(&format!("Bearer {}", key.as_ref()))?,
107    ))
108}
109
110pub fn bearer_auth_header(headers: &mut HeaderMap, key: impl AsRef<str>) -> Result<()> {
111    let (k, v) = make_auth_header(key)?;
112
113    headers.insert(k, v);
114
115    Ok(())
116}
117
118pub fn with_bearer_auth(mut req: Builder, auth: &str) -> Result<Builder> {
119    bearer_auth_header(req.headers_mut().ok_or(Error::NoHeaders)?, auth)?;
120
121    Ok(req)
122}
123
124/// A helper trait to make generic requests (both regular and SSE) possible.
125pub trait HttpClientExt: WasmCompatSend + WasmCompatSync {
126    /// Send a HTTP request, get a response back (as bytes). Response must be able to be turned back into Bytes.
127    fn send<T, U>(
128        &self,
129        req: Request<T>,
130    ) -> impl Future<Output = Result<Response<LazyBody<U>>>> + WasmCompatSend + 'static
131    where
132        T: Into<Bytes>,
133        T: WasmCompatSend,
134        U: From<Bytes>,
135        U: WasmCompatSend + 'static;
136
137    /// Send a HTTP request with a multipart body, get a response back (as bytes). Response must be able to be turned back into Bytes (although usually for the response, you will probably want to specify Bytes anyway).
138    fn send_multipart<U>(
139        &self,
140        req: Request<MultipartForm>,
141    ) -> impl Future<Output = Result<Response<LazyBody<U>>>> + WasmCompatSend + 'static
142    where
143        U: From<Bytes>,
144        U: WasmCompatSend + 'static;
145
146    /// Send a HTTP request, get a streamed response back (as a stream of [`bytes::Bytes`].)
147    fn send_streaming<T>(
148        &self,
149        req: Request<T>,
150    ) -> impl Future<Output = Result<StreamingResponse>> + WasmCompatSend
151    where
152        T: Into<Bytes> + WasmCompatSend;
153}
154
155async fn into_lazy_response<U>(response: reqwest::Response) -> Result<Response<LazyBody<U>>>
156where
157    U: From<Bytes>,
158    U: WasmCompatSend + 'static,
159{
160    if !response.status().is_success() {
161        return Err(non_success_status_error(response).await);
162    }
163
164    let mut res = Response::builder().status(response.status());
165
166    if let Some(headers) = res.headers_mut() {
167        *headers = response.headers().clone();
168    }
169
170    let body: LazyBody<U> = Box::pin(async {
171        let bytes = response.bytes().await.map_err(instance_error)?;
172        Ok(U::from(bytes))
173    });
174
175    res.body(body).map_err(Error::Protocol)
176}
177
178macro_rules! impl_http_client_ext {
179    ($(#[$attribute:meta])* $client:ty) => {
180        $(#[$attribute])*
181        impl HttpClientExt for $client {
182            fn send<T, U>(
183                &self,
184                req: Request<T>,
185            ) -> impl Future<Output = Result<Response<LazyBody<U>>>> + WasmCompatSend + 'static
186            where
187                T: Into<Bytes>,
188                U: From<Bytes> + WasmCompatSend + 'static,
189            {
190                let (parts, body) = req.into_parts();
191                let req = self
192                    .request(parts.method, parts.uri.to_string())
193                    .headers(parts.headers)
194                    .body(body.into());
195
196                async move {
197                    let response = req.send().await.map_err(instance_error)?;
198                    into_lazy_response(response).await
199                }
200            }
201
202            fn send_multipart<U>(
203                &self,
204                req: Request<MultipartForm>,
205            ) -> impl Future<Output = Result<Response<LazyBody<U>>>> + WasmCompatSend + 'static
206            where
207                U: From<Bytes>,
208                U: WasmCompatSend + 'static,
209            {
210                let (parts, body) = req.into_parts();
211                let body = reqwest::multipart::Form::from(body);
212
213                let req = self
214                    .request(parts.method, parts.uri.to_string())
215                    .headers(parts.headers)
216                    .multipart(body);
217
218                async move {
219                    let response = req.send().await.map_err(instance_error)?;
220                    into_lazy_response(response).await
221                }
222            }
223
224            fn send_streaming<T>(
225                &self,
226                req: Request<T>,
227            ) -> impl Future<Output = Result<StreamingResponse>> + WasmCompatSend
228            where
229                T: Into<Bytes> + WasmCompatSend,
230            {
231                let (parts, body) = req.into_parts();
232
233                let client = self.clone();
234
235                async move {
236                    let req = self
237                        .request(parts.method, parts.uri.to_string())
238                        .headers(parts.headers)
239                        .body(body.into())
240                        .build()
241                        .map_err(|error| Error::Instance(error.into()))?;
242                    let response: reqwest::Response =
243                        client.execute(req).await.map_err(instance_error)?;
244                    if !response.status().is_success() {
245                        return Err(non_success_status_error(response).await);
246                    }
247
248                    #[cfg(not(target_family = "wasm"))]
249                    let mut res = Response::builder()
250                        .status(response.status())
251                        .version(response.version());
252
253                    #[cfg(target_family = "wasm")]
254                    let mut res = Response::builder().status(response.status());
255
256                    if let Some(hs) = res.headers_mut() {
257                        *hs = response.headers().clone();
258                    }
259
260                    use futures::StreamExt;
261
262                    let mapped_stream: Pin<
263                        Box<dyn WasmCompatSendStream<InnerItem = Result<Bytes>>>,
264                    > = Box::pin(
265                        response
266                            .bytes_stream()
267                            .map(|chunk| chunk.map_err(|e| Error::Instance(Box::new(e)))),
268                    );
269
270                    res.body(mapped_stream).map_err(Error::Protocol)
271                }
272            }
273        }
274    };
275}
276
277impl_http_client_ext!(reqwest::Client);
278
279impl_http_client_ext!(
280    #[cfg(feature = "reqwest-middleware")]
281    #[cfg_attr(docsrs, doc(cfg(feature = "reqwest-middleware")))]
282    reqwest_middleware::ClientWithMiddleware
283);
rig_core/http_client/mod.rs

rig_core/http_client/
mod.rs
