matrix_sdk/sliding_sync/
builder.rs

1use std::{
2    collections::BTreeMap,
3    fmt::Debug,
4    sync::{Arc, RwLock as StdRwLock},
5    time::Duration,
6};
7
8use matrix_sdk_common::timer;
9use ruma::{OwnedRoomId, api::client::sync::sync_events::v5 as http};
10use tokio::sync::{Mutex as AsyncMutex, RwLock as AsyncRwLock, broadcast::channel};
11
12use super::{
13    Error, SlidingSync, SlidingSyncInner, SlidingSyncListBuilder, SlidingSyncPositionMarkers,
14    Version, cache::format_storage_key_prefix,
15};
16use crate::{Client, Result};
17
18/// Configuration for a Sliding Sync instance.
19///
20/// Get a new builder with methods like [`crate::Client::sliding_sync`], or
21/// [`crate::SlidingSync::builder`].
22#[derive(Debug, Clone)]
23pub struct SlidingSyncBuilder {
24    id: String,
25    storage_key: String,
26    version: Option<Version>,
27    client: Client,
28    lists: Vec<SlidingSyncListBuilder>,
29    extensions: Option<http::request::Extensions>,
30    room_subscriptions: BTreeMap<OwnedRoomId, http::request::RoomSubscription>,
31    poll_timeout: Duration,
32    network_timeout: Duration,
33    #[cfg(feature = "e2e-encryption")]
34    share_pos: bool,
35}
36
37impl SlidingSyncBuilder {
38    pub(super) fn new(id: String, client: Client) -> Result<Self, Error> {
39        if id.len() > 16 {
40            Err(Error::InvalidSlidingSyncIdentifier)
41        } else {
42            let storage_key =
43                format_storage_key_prefix(&id, client.user_id().ok_or(Error::UnauthenticatedUser)?);
44
45            Ok(Self {
46                id,
47                storage_key,
48                version: None,
49                client,
50                lists: Vec::new(),
51                extensions: None,
52                room_subscriptions: BTreeMap::new(),
53                poll_timeout: Duration::from_secs(30),
54                network_timeout: Duration::from_secs(30),
55                #[cfg(feature = "e2e-encryption")]
56                share_pos: false,
57            })
58        }
59    }
60
61    /// Set a specific version that will override the one from the [`Client`].
62    pub fn version(mut self, version: Version) -> Self {
63        self.version = Some(version);
64        self
65    }
66
67    /// Add the given list to the lists.
68    ///
69    /// Replace any list with the same name.
70    pub fn add_list(mut self, list_builder: SlidingSyncListBuilder) -> Self {
71        self.lists.push(list_builder);
72        self
73    }
74
75    /// Enroll the list in caching, reloads it from the cache if possible, and
76    /// adds it to the list of lists.
77    ///
78    /// This will raise an error if there was a I/O error reading from the
79    /// cache.
80    ///
81    /// Replace any list with the same name.
82    pub async fn add_cached_list(self, mut list: SlidingSyncListBuilder) -> Result<Self> {
83        let _timer = timer!(format!("restoring (loading+processing) list {}", list.name));
84
85        list.set_cached_and_reload(&self.client, &self.storage_key).await?;
86
87        Ok(self.add_list(list))
88    }
89
90    /// Activate e2ee, to-device-message, account data, typing, receipt and
91    /// (if compiled in) sticky events extensions if not yet configured.
92    ///
93    /// Will leave any extension configuration found untouched, so the order
94    /// does not matter.
95    pub fn with_all_extensions(mut self) -> Self {
96        {
97            let cfg = self.extensions.get_or_insert_with(Default::default);
98            if cfg.to_device.enabled.is_none() {
99                cfg.to_device.enabled = Some(true);
100            }
101
102            if cfg.e2ee.enabled.is_none() {
103                cfg.e2ee.enabled = Some(true);
104            }
105
106            if cfg.account_data.enabled.is_none() {
107                cfg.account_data.enabled = Some(true);
108            }
109
110            if cfg.receipts.enabled.is_none() {
111                cfg.receipts.enabled = Some(true);
112            }
113
114            if cfg.typing.enabled.is_none() {
115                cfg.typing.enabled = Some(true);
116            }
117
118            #[cfg(feature = "unstable-msc4354")]
119            if cfg.sticky_events.enabled.is_none() {
120                cfg.sticky_events.enabled = Some(true);
121            }
122        }
123        self
124    }
125
126    /// Set the E2EE extension configuration.
127    pub fn with_e2ee_extension(mut self, e2ee: http::request::E2EE) -> Self {
128        self.extensions.get_or_insert_with(Default::default).e2ee = e2ee;
129        self
130    }
131
132    /// Unset the E2EE extension configuration.
133    pub fn without_e2ee_extension(mut self) -> Self {
134        self.extensions.get_or_insert_with(Default::default).e2ee = http::request::E2EE::default();
135        self
136    }
137
138    /// Set the ToDevice extension configuration.
139    pub fn with_to_device_extension(mut self, to_device: http::request::ToDevice) -> Self {
140        self.extensions.get_or_insert_with(Default::default).to_device = to_device;
141        self
142    }
143
144    /// Unset the ToDevice extension configuration.
145    pub fn without_to_device_extension(mut self) -> Self {
146        self.extensions.get_or_insert_with(Default::default).to_device =
147            http::request::ToDevice::default();
148        self
149    }
150
151    /// Set the account data extension configuration.
152    pub fn with_account_data_extension(mut self, account_data: http::request::AccountData) -> Self {
153        self.extensions.get_or_insert_with(Default::default).account_data = account_data;
154        self
155    }
156
157    /// Unset the account data extension configuration.
158    pub fn without_account_data_extension(mut self) -> Self {
159        self.extensions.get_or_insert_with(Default::default).account_data =
160            http::request::AccountData::default();
161        self
162    }
163
164    /// Set the Typing extension configuration.
165    pub fn with_typing_extension(mut self, typing: http::request::Typing) -> Self {
166        self.extensions.get_or_insert_with(Default::default).typing = typing;
167        self
168    }
169
170    /// Unset the Typing extension configuration.
171    pub fn without_typing_extension(mut self) -> Self {
172        self.extensions.get_or_insert_with(Default::default).typing =
173            http::request::Typing::default();
174        self
175    }
176
177    /// Set the Receipt extension configuration.
178    pub fn with_receipt_extension(mut self, receipt: http::request::Receipts) -> Self {
179        self.extensions.get_or_insert_with(Default::default).receipts = receipt;
180        self
181    }
182
183    /// Unset the Receipt extension configuration.
184    pub fn without_receipt_extension(mut self) -> Self {
185        self.extensions.get_or_insert_with(Default::default).receipts =
186            http::request::Receipts::default();
187        self
188    }
189
190    /// Set the Threads subscriptions extension configuration.
191    pub fn with_thread_subscriptions_extension(
192        mut self,
193        thread_subscriptions: http::request::ThreadSubscriptions,
194    ) -> Self {
195        self.extensions.get_or_insert_with(Default::default).thread_subscriptions =
196            thread_subscriptions;
197        self
198    }
199
200    /// Unset the Threads subscriptions extension configuration.
201    pub fn without_thread_subscriptions_extension(mut self) -> Self {
202        self.extensions.get_or_insert_with(Default::default).thread_subscriptions =
203            Default::default();
204        self
205    }
206
207    /// Set the Profiles extension configuration.
208    pub fn with_profiles_extension(mut self, profiles: http::request::Profiles) -> Self {
209        self.extensions.get_or_insert_with(Default::default).profiles = profiles;
210        self
211    }
212
213    /// Unset the Profiles extension configuration.
214    pub fn without_profiles_extension(mut self) -> Self {
215        self.extensions.get_or_insert_with(Default::default).profiles = Default::default();
216        self
217    }
218
219    /// Set the sticky events (MSC4480) extension configuration.
220    ///
221    /// The `since` token of the extension is managed by the sliding sync
222    /// itself, any value set here is ignored.
223    #[cfg(feature = "unstable-msc4354")]
224    pub fn with_sticky_events_extension(
225        mut self,
226        sticky_events: http::request::StickyEvents,
227    ) -> Self {
228        self.extensions.get_or_insert_with(Default::default).sticky_events = sticky_events;
229        self
230    }
231
232    /// Unset the sticky events (MSC4480) extension configuration.
233    #[cfg(feature = "unstable-msc4354")]
234    pub fn without_sticky_events_extension(mut self) -> Self {
235        self.extensions.get_or_insert_with(Default::default).sticky_events = Default::default();
236        self
237    }
238
239    /// Sets a custom timeout duration for the sliding sync polling endpoint.
240    ///
241    /// This is the maximum time to wait before the sliding sync server returns
242    /// the long-polling request. If no events (or other data) become available
243    /// before this time elapses, the server will a return a response with empty
244    /// fields.
245    ///
246    /// There's an additional network timeout on top of that that can be
247    /// configured with [`Self::network_timeout`].
248    pub fn poll_timeout(mut self, timeout: Duration) -> Self {
249        self.poll_timeout = timeout;
250        self
251    }
252
253    /// Sets a custom network timeout for the sliding sync polling.
254    ///
255    /// This is not the polling timeout that can be configured with
256    /// [`Self::poll_timeout`], but an additional timeout that will be added to
257    /// the former.
258    pub fn network_timeout(mut self, timeout: Duration) -> Self {
259        self.network_timeout = timeout;
260        self
261    }
262
263    /// Should the sliding sync instance share its sync position through
264    /// storage?
265    ///
266    /// In general, sliding sync instances will cache the sync position (`pos`
267    /// field in the request) in internal memory. It can be useful, in
268    /// multi-process scenarios, to save it into some shared storage so that one
269    /// sliding sync instance running across two different processes can
270    /// continue with the same sync position it had before being stopped.
271    #[cfg(feature = "e2e-encryption")]
272    pub fn share_pos(mut self) -> Self {
273        self.share_pos = true;
274        self
275    }
276
277    /// Build the Sliding Sync.
278    #[allow(clippy::unused_async)] // Async is only used if the e2e-encryption feature is enabled.
279    pub async fn build(self) -> Result<SlidingSync> {
280        let client = self.client;
281
282        let version = self.version.unwrap_or_else(|| client.sliding_sync_version());
283
284        if matches!(version, Version::None) {
285            return Err(crate::error::Error::SlidingSync(Box::new(Error::VersionIsMissing)));
286        }
287
288        let (internal_channel_sender, _internal_channel_receiver) = channel(8);
289
290        let mut lists = BTreeMap::new();
291
292        for list_builder in self.lists {
293            let list = list_builder.build(internal_channel_sender.clone());
294
295            lists.insert(list.name().to_owned(), list);
296        }
297
298        let (share_pos, pos) = {
299            cfg_select! {
300                feature = "e2e-encryption" => {
301                    if self.share_pos {
302                        // If the sliding sync instance is configured to share
303                        // its current sync position, we will restore it from
304                        // the cache.
305                        (
306                            true,
307                            super::cache::restore_sliding_sync_state(&client, &self.storage_key)
308                                .await?
309                                .and_then(|fields| fields.pos),
310                        )
311                    } else {
312                        (false, None)
313                    }
314                }
315                _ => (false, None),
316            }
317        };
318
319        let lists = AsyncRwLock::new(lists);
320
321        Ok(SlidingSync::new(SlidingSyncInner {
322            id: self.id,
323
324            client,
325            storage_key: self.storage_key,
326            share_pos,
327
328            lists,
329
330            position: Arc::new(AsyncMutex::new(SlidingSyncPositionMarkers {
331                pos,
332                #[cfg(feature = "unstable-msc4354")]
333                sticky_events_since: None,
334            })),
335
336            room_subscriptions: StdRwLock::new(self.room_subscriptions),
337            extensions: self.extensions.unwrap_or_default(),
338
339            internal_channel: internal_channel_sender,
340
341            poll_timeout: self.poll_timeout,
342            network_timeout: self.network_timeout,
343        }))
344    }
345}
matrix_sdk/sliding_sync/builder.rs

matrix_sdk/sliding_sync/
builder.rs
