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

matrix_sdk/sliding_sync/
builder.rs
