meilisearch_sdk/
reqwest.rs

1use std::{
2    pin::Pin,
3    task::{Context, Poll},
4};
5
6use async_trait::async_trait;
7use bytes::{Bytes, BytesMut};
8use futures_core::Stream;
9use futures_io::AsyncRead;
10use pin_project_lite::pin_project;
11use serde::{de::DeserializeOwned, Serialize};
12
13use crate::{
14    errors::Error,
15    request::{parse_response, HttpClient, Method},
16};
17
18#[derive(Debug, Clone, Default)]
19pub struct ReqwestClient {
20    client: reqwest::Client,
21}
22
23impl ReqwestClient {
24    pub fn new(api_key: Option<&str>) -> Result<Self, Error> {
25        use reqwest::{header, ClientBuilder};
26
27        let builder = ClientBuilder::new();
28        let mut headers = header::HeaderMap::new();
29        #[cfg(not(target_arch = "wasm32"))]
30        headers.insert(
31            header::USER_AGENT,
32            header::HeaderValue::from_str(&qualified_version()).unwrap(),
33        );
34        #[cfg(target_arch = "wasm32")]
35        headers.insert(
36            header::HeaderName::from_static("x-meilisearch-client"),
37            header::HeaderValue::from_str(&qualified_version()).unwrap(),
38        );
39
40        if let Some(api_key) = api_key {
41            headers.insert(
42                header::AUTHORIZATION,
43                header::HeaderValue::from_str(&format!("Bearer {api_key}")).unwrap(),
44            );
45        }
46
47        let builder = builder.default_headers(headers);
48        let client = builder.build()?;
49
50        Ok(ReqwestClient { client })
51    }
52
53    pub(crate) fn inner(&self) -> &reqwest::Client {
54        &self.client
55    }
56}
57
58#[cfg_attr(feature = "futures-unsend", async_trait(?Send))]
59#[cfg_attr(not(feature = "futures-unsend"), async_trait)]
60impl HttpClient for ReqwestClient {
61    async fn stream_request<
62        Query: Serialize + Send + Sync,
63        Body: futures_io::AsyncRead + Send + Sync + 'static,
64        Output: DeserializeOwned + 'static,
65    >(
66        &self,
67        url: &str,
68        method: Method<Query, Body>,
69        content_type: &str,
70        expected_status_code: u16,
71    ) -> Result<Output, Error> {
72        use reqwest::header;
73
74        let query = method.query();
75        let query = yaup::to_string(query)?;
76
77        let url = if query.is_empty() {
78            url.to_string()
79        } else {
80            format!("{url}{query}")
81        };
82
83        let mut request = self.client.request(verb(&method), &url);
84
85        if let Some(body) = method.into_body() {
86            // TODO: Currently reqwest doesn't support streaming data in wasm so we need to collect everything in RAM
87            #[cfg(not(target_arch = "wasm32"))]
88            {
89                let stream = ReaderStream::new(body);
90                let body = reqwest::Body::wrap_stream(stream);
91
92                request = request
93                    .header(header::CONTENT_TYPE, content_type)
94                    .body(body);
95            }
96            #[cfg(target_arch = "wasm32")]
97            {
98                use futures_util::AsyncReadExt;
99
100                let mut buf = Vec::new();
101                let mut body = std::pin::pin!(body);
102                body.read_to_end(&mut buf)
103                    .await
104                    .map_err(|err| Error::Other(Box::new(err)))?;
105                request = request.header(header::CONTENT_TYPE, content_type).body(buf);
106            }
107        }
108
109        let response = self.client.execute(request.build()?).await?;
110        let status = response.status().as_u16();
111        let mut body = response.text().await?;
112
113        if body.is_empty() {
114            body = "null".to_string();
115        }
116
117        parse_response(status, expected_status_code, &body, url.to_string())
118    }
119
120    fn is_tokio(&self) -> bool {
121        true
122    }
123}
124
125fn verb<Q, B>(method: &Method<Q, B>) -> reqwest::Method {
126    match method {
127        Method::Get { .. } => reqwest::Method::GET,
128        Method::Delete { .. } => reqwest::Method::DELETE,
129        Method::Post { .. } => reqwest::Method::POST,
130        Method::Put { .. } => reqwest::Method::PUT,
131        Method::Patch { .. } => reqwest::Method::PATCH,
132    }
133}
134
135pub fn qualified_version() -> String {
136    const VERSION: Option<&str> = option_env!("CARGO_PKG_VERSION");
137
138    format!("Meilisearch Rust (v{})", VERSION.unwrap_or("unknown"))
139}
140
141pin_project! {
142    #[derive(Debug)]
143    pub struct ReaderStream<R: AsyncRead> {
144        #[pin]
145        reader: R,
146        buf: BytesMut,
147        capacity: usize,
148    }
149}
150
151impl<R: AsyncRead> ReaderStream<R> {
152    pub fn new(reader: R) -> Self {
153        Self {
154            reader,
155            buf: BytesMut::new(),
156            // 8KiB of capacity, the default capacity used by `BufReader` in the std
157            capacity: 8 * 1024 * 1024,
158        }
159    }
160}
161
162impl<R: AsyncRead> Stream for ReaderStream<R> {
163    type Item = std::io::Result<Bytes>;
164
165    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
166        let this = self.as_mut().project();
167
168        if this.buf.capacity() == 0 {
169            this.buf.resize(*this.capacity, 0);
170        }
171
172        match AsyncRead::poll_read(this.reader, cx, this.buf) {
173            Poll::Pending => Poll::Pending,
174            Poll::Ready(Err(err)) => Poll::Ready(Some(Err(err))),
175            Poll::Ready(Ok(0)) => Poll::Ready(None),
176            Poll::Ready(Ok(i)) => {
177                let chunk = this.buf.split_to(i);
178                Poll::Ready(Some(Ok(chunk.freeze())))
179            }
180        }
181    }
182}
meilisearch_sdk/reqwest.rs

meilisearch_sdk/
reqwest.rs
