matrix_sdk/client/
futures.rs

1// Copyright 2023 The Matrix.org Foundation C.I.C.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15#![deny(unreachable_pub)]
16
17use std::{fmt::Debug, future::IntoFuture};
18
19use eyeball::{SharedObservable, Subscriber};
20use js_int::UInt;
21use matrix_sdk_common::{SendOutsideWasm, SyncOutsideWasm, boxed_into_future};
22use oauth2::{RequestTokenError, basic::BasicErrorResponseType};
23use ruma::api::{
24    OutgoingRequest,
25    client::media,
26    error::{ErrorKind, FromHttpResponseError},
27    path_builder::PathBuilder,
28};
29use tracing::{error, trace};
30
31use super::super::Client;
32use crate::{
33    Error, RefreshTokenError, TransmissionProgress,
34    authentication::oauth::OAuthError,
35    config::RequestConfig,
36    error::{HttpError, HttpResult},
37    http_client::{SupportedAuthScheme, SupportedPathBuilder},
38    media::MediaError,
39};
40
41/// `IntoFuture` returned by [`Client::send`].
42#[allow(missing_debug_implementations)]
43pub struct SendRequest<R> {
44    pub(crate) client: Client,
45    pub(crate) request: R,
46    pub(crate) config: Option<RequestConfig>,
47    pub(crate) send_progress: SharedObservable<TransmissionProgress>,
48}
49
50impl<R> SendRequest<R> {
51    /// Replace the default `SharedObservable` used for tracking upload
52    /// progress.
53    ///
54    /// Note that any subscribers obtained from
55    /// [`subscribe_to_send_progress`][Self::subscribe_to_send_progress]
56    /// will be invalidated by this.
57    pub fn with_send_progress_observable(
58        mut self,
59        send_progress: SharedObservable<TransmissionProgress>,
60    ) -> Self {
61        self.send_progress = send_progress;
62        self
63    }
64
65    /// Use the given [`RequestConfig`] for this send request, instead of the
66    /// one provided by default.
67    pub fn with_request_config(mut self, request_config: impl Into<Option<RequestConfig>>) -> Self {
68        self.config = request_config.into();
69        self
70    }
71
72    /// Get a subscriber to observe the progress of sending the request body.
73    pub fn subscribe_to_send_progress(&self) -> Subscriber<TransmissionProgress> {
74        self.send_progress.subscribe()
75    }
76}
77
78impl<R> IntoFuture for SendRequest<R>
79where
80    R: OutgoingRequest + Clone + Debug + SendOutsideWasm + SyncOutsideWasm + 'static,
81    R::Authentication: SupportedAuthScheme,
82    R::PathBuilder: SupportedPathBuilder,
83    for<'a> <R::PathBuilder as PathBuilder>::Input<'a>: SendOutsideWasm + SyncOutsideWasm,
84    R::IncomingResponse: SendOutsideWasm + SyncOutsideWasm,
85    HttpError: From<FromHttpResponseError<R::EndpointError>>,
86{
87    type Output = HttpResult<R::IncomingResponse>;
88    boxed_into_future!();
89
90    fn into_future(self) -> Self::IntoFuture {
91        enum RetryRequest {
92            Yes,
93            No,
94        }
95
96        // split out so this only get compiled once, not monomorphized per
97        // request type
98        async fn handle_unknown_token_error(
99            e: &HttpError,
100            client: &Client,
101        ) -> HttpResult<RetryRequest> {
102            // An `M_UNKNOWN_TOKEN` error can potentially be fixed with a token
103            // refresh.
104            let Some(ErrorKind::UnknownToken(unknown_token_data)) = e.client_api_error_kind()
105            else {
106                return Ok(RetryRequest::No);
107            };
108
109            trace!("Token refresh: Unknown token error received.");
110
111            // If automatic token refresh isn't supported, there is nothing more
112            // to do.
113            if !client.inner.auth_ctx.handle_refresh_tokens {
114                trace!("Token refresh: Automatic refresh disabled.");
115                client.broadcast_unknown_token(unknown_token_data);
116                return Ok(RetryRequest::No);
117            }
118
119            // Try to refresh the token and retry the request.
120            if let Err(refresh_error) = client.refresh_access_token().await {
121                match &refresh_error {
122                    RefreshTokenError::RefreshTokenRequired => {
123                        trace!("Token refresh: The session doesn't have a refresh token.");
124                        // Refreshing access tokens is not supported by this
125                        // `Session`, ignore.
126                        client.broadcast_unknown_token(unknown_token_data);
127                        Ok(RetryRequest::No)
128                    }
129
130                    RefreshTokenError::OAuth(oauth_error) => {
131                        match &**oauth_error {
132                            OAuthError::RefreshToken(RequestTokenError::ServerResponse(
133                                error_response,
134                            )) if *error_response.error()
135                                == BasicErrorResponseType::InvalidGrant =>
136                            {
137                                error!(
138                                    "Token refresh: OAuth 2.0 refresh_token rejected \
139                                         with invalid grant"
140                                );
141                                // The refresh was denied, signal to sign out
142                                // the user.
143                                client.broadcast_unknown_token(unknown_token_data);
144                            }
145                            _ => {
146                                trace!("Token refresh: OAuth 2.0 refresh encountered a problem.");
147                                // The refresh failed for other reasons, no need
148                                // to sign out.
149                            }
150                        }
151                        Err(HttpError::RefreshToken(refresh_error))
152                    }
153
154                    RefreshTokenError::MatrixAuth(http_error) => {
155                        // Only an answer from the homeserver means the token is
156                        // gone. A transport failure or a 5xx says nothing about
157                        // the session.
158                        let rejected = http_error
159                            .as_client_api_error()
160                            .is_some_and(|error| !error.status_code.is_server_error());
161
162                        if rejected {
163                            error!("Token refresh: the homeserver rejected the refresh token");
164                            client.broadcast_unknown_token(unknown_token_data);
165                        } else {
166                            trace!("Token refresh: the refresh request itself failed.");
167                        }
168                        Err(HttpError::RefreshToken(refresh_error))
169                    }
170                }
171            } else {
172                trace!("Token refresh: Refresh succeeded, retrying request.");
173                Ok(RetryRequest::Yes)
174            }
175        }
176
177        let Self { client, request, config, send_progress } = self;
178
179        Box::pin(async move {
180            let res =
181                Box::pin(client.send_inner(request.clone(), config, send_progress.clone())).await;
182
183            if let Err(e) = &res
184                && let RetryRequest::Yes = handle_unknown_token_error(e, &client).await?
185            {
186                return Box::pin(client.send_inner(request, config, send_progress)).await;
187            }
188
189            res
190        })
191    }
192}
193
194/// `IntoFuture` used to send media upload requests. It wraps another
195/// [`SendRequest`], checking its size will be accepted by the homeserver before
196/// uploading.
197#[allow(missing_debug_implementations)]
198pub struct SendMediaUploadRequest {
199    send_request: SendRequest<media::create_content::v3::Request>,
200}
201
202impl SendMediaUploadRequest {
203    pub fn new(request: SendRequest<media::create_content::v3::Request>) -> Self {
204        Self { send_request: request }
205    }
206
207    /// Replace the default `SharedObservable` used for tracking upload
208    /// progress.
209    ///
210    /// Note that any subscribers obtained from
211    /// [`subscribe_to_send_progress`][Self::subscribe_to_send_progress]
212    /// will be invalidated by this.
213    pub fn with_send_progress_observable(
214        mut self,
215        send_progress: SharedObservable<TransmissionProgress>,
216    ) -> Self {
217        self.send_request = self.send_request.with_send_progress_observable(send_progress);
218        self
219    }
220
221    /// Get a subscriber to observe the progress of sending the request body.
222    pub fn subscribe_to_send_progress(&self) -> Subscriber<TransmissionProgress> {
223        self.send_request.send_progress.subscribe()
224    }
225}
226
227impl IntoFuture for SendMediaUploadRequest {
228    type Output = Result<media::create_content::v3::Response, Error>;
229    boxed_into_future!();
230
231    fn into_future(self) -> Self::IntoFuture {
232        let request_length = self.send_request.request.file.len();
233        let client = self.send_request.client.clone();
234        let send_request = self.send_request;
235
236        Box::pin(async move {
237            let max_upload_size = client.load_or_fetch_max_upload_size().await?;
238            let request_length = UInt::new_wrapping(request_length as u64);
239            if request_length > max_upload_size {
240                return Err(Error::Media(MediaError::MediaTooLargeToUpload {
241                    max: max_upload_size,
242                    current: request_length,
243                }));
244            }
245
246            send_request.into_future().await.map_err(Into::into)
247        })
248    }
249}
matrix_sdk/client/futures.rs

matrix_sdk/client/
futures.rs
