1use std::{
16 any::type_name,
17 fmt::Debug,
18 num::NonZeroUsize,
19 sync::{
20 atomic::{AtomicU64, Ordering},
21 Arc,
22 },
23 time::Duration,
24};
25
26use bytes::{Bytes, BytesMut};
27use bytesize::ByteSize;
28use eyeball::SharedObservable;
29use http::Method;
30use ruma::api::{
31 error::{FromHttpResponseError, IntoHttpError},
32 AuthScheme, OutgoingRequest, SendAccessToken, SupportedVersions,
33};
34use tokio::sync::{Semaphore, SemaphorePermit};
35use tracing::{debug, field::debug, instrument, trace};
36
37use crate::{config::RequestConfig, error::HttpError};
38
39#[cfg(not(target_family = "wasm"))]
40mod native;
41#[cfg(target_family = "wasm")]
42mod wasm;
43
44#[cfg(not(target_family = "wasm"))]
45pub(crate) use native::HttpSettings;
46
47pub(crate) const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
48
49#[derive(Clone, Debug)]
50struct MaybeSemaphore(Arc<Option<Semaphore>>);
51
52#[allow(dead_code)] struct MaybeSemaphorePermit<'a>(Option<SemaphorePermit<'a>>);
54
55impl MaybeSemaphore {
56 fn new(max: Option<NonZeroUsize>) -> Self {
57 let inner = max.map(|i| Semaphore::new(i.into()));
58 MaybeSemaphore(Arc::new(inner))
59 }
60
61 async fn acquire(&self) -> MaybeSemaphorePermit<'_> {
62 match self.0.as_ref() {
63 Some(inner) => {
64 MaybeSemaphorePermit(inner.acquire().await.ok())
67 }
68 None => MaybeSemaphorePermit(None),
69 }
70 }
71}
72
73#[derive(Clone, Debug)]
74pub(crate) struct HttpClient {
75 pub(crate) inner: reqwest::Client,
76 pub(crate) request_config: RequestConfig,
77 concurrent_request_semaphore: MaybeSemaphore,
78 next_request_id: Arc<AtomicU64>,
79}
80
81impl HttpClient {
82 pub(crate) fn new(inner: reqwest::Client, request_config: RequestConfig) -> Self {
83 HttpClient {
84 inner,
85 request_config,
86 concurrent_request_semaphore: MaybeSemaphore::new(
87 request_config.max_concurrent_requests,
88 ),
89 next_request_id: AtomicU64::new(0).into(),
90 }
91 }
92
93 fn get_request_id(&self) -> String {
94 let request_id = self.next_request_id.fetch_add(1, Ordering::SeqCst);
95 format!("REQ-{request_id}")
96 }
97
98 fn serialize_request<R>(
99 &self,
100 request: R,
101 config: RequestConfig,
102 homeserver: String,
103 access_token: Option<&str>,
104 supported_versions: &SupportedVersions,
105 ) -> Result<http::Request<Bytes>, IntoHttpError>
106 where
107 R: OutgoingRequest + Debug,
108 {
109 trace!(request_type = type_name::<R>(), "Serializing request");
110
111 let send_access_token = match access_token {
112 Some(access_token) => {
113 if config.force_auth {
114 SendAccessToken::Always(access_token)
115 } else {
116 SendAccessToken::IfRequired(access_token)
117 }
118 }
119 None => SendAccessToken::None,
120 };
121
122 let request = request
123 .try_into_http_request::<BytesMut>(&homeserver, send_access_token, supported_versions)?
124 .map(|body| body.freeze());
125
126 Ok(request)
127 }
128
129 #[allow(clippy::too_many_arguments)]
130 #[instrument(
131 skip(self, request, config, homeserver, access_token, supported_versions, send_progress),
132 fields(
133 uri,
134 method,
135 request_id,
136 request_size,
137 request_duration,
138 status,
139 response_size,
140 sentry_event_id
141 )
142 )]
143 pub async fn send<R>(
144 &self,
145 request: R,
146 config: Option<RequestConfig>,
147 homeserver: String,
148 access_token: Option<&str>,
149 supported_versions: &SupportedVersions,
150 send_progress: SharedObservable<TransmissionProgress>,
151 ) -> Result<R::IncomingResponse, HttpError>
152 where
153 R: OutgoingRequest + Debug,
154 HttpError: From<FromHttpResponseError<R::EndpointError>>,
155 {
156 let config = match config {
157 Some(config) => config,
158 None => self.request_config,
159 };
160
161 let request = {
164 let request_id = self.get_request_id();
165 let span = tracing::Span::current();
166
167 span.record("config", debug(config)).record("request_id", request_id);
170
171 let auth_scheme = R::METADATA.authentication;
172 match auth_scheme {
173 AuthScheme::AccessToken
174 | AuthScheme::AccessTokenOptional
175 | AuthScheme::AppserviceToken
176 | AuthScheme::AppserviceTokenOptional
177 | AuthScheme::None => {}
178 AuthScheme::ServerSignatures => {
179 return Err(HttpError::NotClientRequest);
180 }
181 }
182
183 let request = self
184 .serialize_request(request, config, homeserver, access_token, supported_versions)
185 .map_err(HttpError::IntoHttp)?;
186
187 let method = request.method();
188
189 let mut uri_parts = request.uri().clone().into_parts();
190 if let Some(path_and_query) = &mut uri_parts.path_and_query {
191 *path_and_query =
192 path_and_query.path().try_into().expect("path is valid PathAndQuery");
193 }
194 let uri = http::Uri::from_parts(uri_parts).expect("created from valid URI");
195
196 span.record("method", debug(method)).record("uri", uri.to_string());
197
198 if [Method::POST, Method::PUT, Method::PATCH].contains(method) {
201 let request_size = request.body().len().try_into().unwrap_or(u64::MAX);
202 span.record(
203 "request_size",
204 ByteSize(request_size).display().si_short().to_string(),
205 );
206 }
207
208 request
209 };
210
211 let _handle = self.concurrent_request_semaphore.acquire().await;
213
214 match Box::pin(self.send_request::<R>(request, config, send_progress)).await {
217 Ok(response) => {
218 debug!("Got response");
219 Ok(response)
220 }
221 Err(e) => {
222 debug!("Error while sending request: {e:?}");
223 Err(e)
224 }
225 }
226 }
227}
228
229#[derive(Clone, Copy, Debug, Default)]
231pub struct TransmissionProgress {
232 pub current: usize,
234 pub total: usize,
236}
237
238async fn response_to_http_response(
239 mut response: reqwest::Response,
240) -> Result<http::Response<Bytes>, reqwest::Error> {
241 let status = response.status();
242
243 let mut http_builder = http::Response::builder().status(status);
244 let headers = http_builder.headers_mut().expect("Can't get the response builder headers");
245
246 for (k, v) in response.headers_mut().drain() {
247 if let Some(key) = k {
248 headers.insert(key, v);
249 }
250 }
251
252 let body = response.bytes().await?;
253
254 Ok(http_builder.body(body).expect("Can't construct a response using the given body"))
255}
256
257#[cfg(all(test, not(target_family = "wasm")))]
258mod tests {
259 use std::{
260 num::NonZeroUsize,
261 sync::{
262 atomic::{AtomicU8, Ordering},
263 Arc,
264 },
265 time::Duration,
266 };
267
268 use matrix_sdk_common::executor::spawn;
269 use matrix_sdk_test::{async_test, test_json};
270 use wiremock::{
271 matchers::{method, path},
272 Mock, Request, ResponseTemplate,
273 };
274
275 use crate::{
276 http_client::RequestConfig,
277 test_utils::{set_client_session, test_client_builder_with_server},
278 };
279
280 #[async_test]
281 async fn test_ensure_concurrent_request_limit_is_observed() {
282 let (client_builder, server) = test_client_builder_with_server().await;
283 let client = client_builder
284 .request_config(RequestConfig::default().max_concurrent_requests(NonZeroUsize::new(5)))
285 .build()
286 .await
287 .unwrap();
288
289 set_client_session(&client).await;
290
291 let counter = Arc::new(AtomicU8::new(0));
292 let inner_counter = counter.clone();
293
294 Mock::given(method("GET"))
295 .and(path("/_matrix/client/versions"))
296 .respond_with(ResponseTemplate::new(200).set_body_json(&*test_json::VERSIONS))
297 .mount(&server)
298 .await;
299
300 Mock::given(method("GET"))
301 .and(path("_matrix/client/r0/account/whoami"))
302 .respond_with(move |_req: &Request| {
303 inner_counter.fetch_add(1, Ordering::SeqCst);
304 ResponseTemplate::new(200).set_delay(Duration::from_secs(60))
306 })
307 .mount(&server)
308 .await;
309
310 let bg_task = spawn(async move {
311 futures_util::future::join_all((0..10).map(|_| client.whoami())).await
312 });
313
314 tokio::time::sleep(Duration::from_millis(300)).await;
316
317 assert_eq!(
318 counter.load(Ordering::SeqCst),
319 5,
320 "More requests passed than the limit we configured"
321 );
322 bg_task.abort();
323 }
324
325 #[async_test]
326 async fn test_ensure_no_max_concurrent_request_does_not_limit() {
327 let (client_builder, server) = test_client_builder_with_server().await;
328 let client = client_builder
329 .request_config(RequestConfig::default().max_concurrent_requests(None))
330 .build()
331 .await
332 .unwrap();
333
334 set_client_session(&client).await;
335
336 let counter = Arc::new(AtomicU8::new(0));
337 let inner_counter = counter.clone();
338
339 Mock::given(method("GET"))
340 .and(path("/_matrix/client/versions"))
341 .respond_with(ResponseTemplate::new(200).set_body_json(&*test_json::VERSIONS))
342 .mount(&server)
343 .await;
344
345 Mock::given(method("GET"))
346 .and(path("_matrix/client/r0/account/whoami"))
347 .respond_with(move |_req: &Request| {
348 inner_counter.fetch_add(1, Ordering::SeqCst);
349 ResponseTemplate::new(200).set_delay(Duration::from_secs(60))
350 })
351 .mount(&server)
352 .await;
353
354 let bg_task = spawn(async move {
355 futures_util::future::join_all((0..254).map(|_| client.whoami())).await
356 });
357
358 tokio::time::sleep(Duration::from_secs(1)).await;
360
361 assert_eq!(counter.load(Ordering::SeqCst), 254, "Not all requests passed through");
362 bg_task.abort();
363 }
364}