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 ruma::api::{
23    OutgoingRequest,
24    client::media,
25    error::{ErrorKind, FromHttpResponseError},
26    path_builder::PathBuilder,
27};
28use tracing::{error, trace};
29
30use super::super::Client;
31use crate::{
32    Error, RefreshTokenError, TransmissionProgress,
33    config::RequestConfig,
34    error::{HttpError, HttpResult},
35    http_client::{SupportedAuthScheme, SupportedPathBuilder},
36    media::MediaError,
37};
38
39/// `IntoFuture` returned by [`Client::send`].
40#[allow(missing_debug_implementations)]
41pub struct SendRequest<R> {
42    pub(crate) client: Client,
43    pub(crate) request: R,
44    pub(crate) config: Option<RequestConfig>,
45    pub(crate) send_progress: SharedObservable<TransmissionProgress>,
46}
47
48impl<R> SendRequest<R> {
49    /// Replace the default `SharedObservable` used for tracking upload
50    /// progress.
51    ///
52    /// Note that any subscribers obtained from
53    /// [`subscribe_to_send_progress`][Self::subscribe_to_send_progress]
54    /// will be invalidated by this.
55    pub fn with_send_progress_observable(
56        mut self,
57        send_progress: SharedObservable<TransmissionProgress>,
58    ) -> Self {
59        self.send_progress = send_progress;
60        self
61    }
62
63    /// Use the given [`RequestConfig`] for this send request, instead of the
64    /// one provided by default.
65    pub fn with_request_config(mut self, request_config: impl Into<Option<RequestConfig>>) -> Self {
66        self.config = request_config.into();
67        self
68    }
69
70    /// Get a subscriber to observe the progress of sending the request body.
71    pub fn subscribe_to_send_progress(&self) -> Subscriber<TransmissionProgress> {
72        self.send_progress.subscribe()
73    }
74}
75
76impl<R> IntoFuture for SendRequest<R>
77where
78    R: OutgoingRequest + Clone + Debug + SendOutsideWasm + SyncOutsideWasm + 'static,
79    R::Authentication: SupportedAuthScheme,
80    R::PathBuilder: SupportedPathBuilder,
81    for<'a> <R::PathBuilder as PathBuilder>::Input<'a>: SendOutsideWasm + SyncOutsideWasm,
82    R::IncomingResponse: SendOutsideWasm + SyncOutsideWasm,
83    HttpError: From<FromHttpResponseError<R::EndpointError>>,
84{
85    type Output = HttpResult<R::IncomingResponse>;
86    boxed_into_future!();
87
88    fn into_future(self) -> Self::IntoFuture {
89        enum RetryRequest {
90            Yes,
91            No,
92        }
93
94        // split out so this only get compiled once, not monomorphized per
95        // request type
96        async fn handle_unknown_token_error(
97            e: &HttpError,
98            client: &Client,
99        ) -> HttpResult<RetryRequest> {
100            // An `M_UNKNOWN_TOKEN` error can potentially be fixed with a token
101            // refresh.
102            let Some(ErrorKind::UnknownToken(unknown_token_data)) = e.client_api_error_kind()
103            else {
104                return Ok(RetryRequest::No);
105            };
106
107            trace!("Token refresh: Unknown token error received.");
108
109            // If automatic token refresh isn't supported, there is nothing more
110            // to do.
111            if !client.inner.auth_ctx.handle_refresh_tokens {
112                trace!("Token refresh: Automatic refresh disabled.");
113                client.broadcast_unknown_token(unknown_token_data);
114                return Ok(RetryRequest::No);
115            }
116
117            // Try to refresh the token and retry the request.
118            if let Err(refresh_error) = client.refresh_access_token().await {
119                match &refresh_error {
120                    RefreshTokenError::RefreshTokenRequired => {
121                        trace!("Token refresh: The session doesn't have a refresh token.");
122                        // Refreshing access tokens is not supported by this
123                        // `Session`, ignore.
124                        client.broadcast_unknown_token(unknown_token_data);
125                        Ok(RetryRequest::No)
126                    }
127
128                    RefreshTokenError::OAuth(_) | RefreshTokenError::MatrixAuth(_) => {
129                        if refresh_error.is_session_rejected() {
130                            error!(
131                                "Token refresh: the refresh token was rejected: {refresh_error}"
132                            );
133                            // The refresh was denied, signal to sign out the
134                            // user.
135                            client.broadcast_unknown_token(unknown_token_data);
136                        } else {
137                            trace!("Token refresh: the refresh request itself failed.");
138                        }
139                        Err(HttpError::RefreshToken(refresh_error))
140                    }
141                }
142            } else {
143                trace!("Token refresh: Refresh succeeded, retrying request.");
144                Ok(RetryRequest::Yes)
145            }
146        }
147
148        let Self { client, request, config, send_progress } = self;
149
150        Box::pin(async move {
151            let res =
152                Box::pin(client.send_inner(request.clone(), config, send_progress.clone())).await;
153
154            if let Err(e) = &res
155                && let RetryRequest::Yes = handle_unknown_token_error(e, &client).await?
156            {
157                return Box::pin(client.send_inner(request, config, send_progress)).await;
158            }
159
160            res
161        })
162    }
163}
164
165/// `IntoFuture` used to send media upload requests. It wraps another
166/// [`SendRequest`], checking its size will be accepted by the homeserver before
167/// uploading.
168#[allow(missing_debug_implementations)]
169pub struct SendMediaUploadRequest {
170    send_request: SendRequest<media::create_content::v3::Request>,
171}
172
173impl SendMediaUploadRequest {
174    pub fn new(request: SendRequest<media::create_content::v3::Request>) -> Self {
175        Self { send_request: request }
176    }
177
178    /// Replace the default `SharedObservable` used for tracking upload
179    /// progress.
180    ///
181    /// Note that any subscribers obtained from
182    /// [`subscribe_to_send_progress`][Self::subscribe_to_send_progress]
183    /// will be invalidated by this.
184    pub fn with_send_progress_observable(
185        mut self,
186        send_progress: SharedObservable<TransmissionProgress>,
187    ) -> Self {
188        self.send_request = self.send_request.with_send_progress_observable(send_progress);
189        self
190    }
191
192    /// Get a subscriber to observe the progress of sending the request body.
193    pub fn subscribe_to_send_progress(&self) -> Subscriber<TransmissionProgress> {
194        self.send_request.send_progress.subscribe()
195    }
196}
197
198impl IntoFuture for SendMediaUploadRequest {
199    type Output = Result<media::create_content::v3::Response, Error>;
200    boxed_into_future!();
201
202    fn into_future(self) -> Self::IntoFuture {
203        let request_length = self.send_request.request.file.len();
204        let client = self.send_request.client.clone();
205        let send_request = self.send_request;
206
207        Box::pin(async move {
208            let max_upload_size = client.load_or_fetch_max_upload_size().await?;
209            let request_length = UInt::new_wrapping(request_length as u64);
210            if request_length > max_upload_size {
211                return Err(Error::Media(MediaError::MediaTooLargeToUpload {
212                    max: max_upload_size,
213                    current: request_length,
214                }));
215            }
216
217            send_request.into_future().await.map_err(Into::into)
218        })
219    }
220}
matrix_sdk/client/futures.rs

matrix_sdk/client/
futures.rs
