mxlink 1.15.0

A library for connecting (linking) to the Matrix Client Server API (for building bots, etc.)
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106

use std::sync::Arc;
use std::time::Duration;

use matrix_sdk::{LoopCtrl, config::SyncSettings, ruma::api::client::filter::FilterDefinition};

use thiserror::Error;

use crate::SessionPersistenceError;
use crate::utils::is_potentially_transient_sdk_error;

const SYNC_INITIAL_DELAY_DURATION: Duration = Duration::from_secs(3);
const SYNC_MAX_DELAY_DURATION: Duration = Duration::from_secs(30);

#[derive(Error, Debug)]
pub enum SyncError {
    #[error("Error from the matrix SDK: {0}")]
    Sdk(#[from] matrix_sdk::Error),

    #[error("Error persisting/restoring session: {0}")]
    SessionPersistence(SessionPersistenceError),
}

#[derive(Clone)]
pub struct Syncing {
    matrix_link: super::MatrixLink,
}

impl Syncing {
    pub(super) fn new(matrix_link: super::MatrixLink) -> Self {
        Self { matrix_link }
    }

    /// Setup the client to listen to new messages.
    pub async fn start(&self) -> Result<(), SyncError> {
        // Enable room members lazy-loading, it will speed up the initial sync a lot
        // with accounts in lots of rooms.
        // See <https://spec.matrix.org/v1.6/client-server-api/#lazy-loading-room-members>.
        let filter = FilterDefinition::with_lazy_loading();

        let mut sync_settings = SyncSettings::default().filter(filter.into());

        // We restore the sync where we left.
        if let Some(sync_token) = &self.matrix_link.inner.initial_sync_token {
            sync_settings = sync_settings.token(sync_token);
        }

        let delay = Arc::new(tokio::sync::Mutex::new(SYNC_INITIAL_DELAY_DURATION));

        let persistence_manager = &self.matrix_link.inner.persistence_manager;

        tracing::info!("Syncing..");

        self.matrix_link
            .inner
            .client
            .sync_with_result_callback(sync_settings, {
                let delay = Arc::clone(&delay);
                move |sync_result| {
                    let delay = Arc::clone(&delay);
                    async move {
                        match sync_result {
                            Ok(response) => {
                                // Reset delay on successful sync
                                let mut current_delay = delay.lock().await;
                                *current_delay = SYNC_INITIAL_DELAY_DURATION;

                                // We persist the token each time to be able to restore our session
                                if let Err(err) = persistence_manager
                                    .persist_sync_token(response.next_batch.clone())
                                    .await
                                {
                                    return Err(matrix_sdk::Error::UnknownError(err.into()));
                                }

                                Ok(LoopCtrl::Continue)
                            }
                            Err(err) => {
                                if !is_potentially_transient_sdk_error(&err) {
                                    tracing::error!(?err, "Sync failed with a permanent error");
                                    return Err(err);
                                }

                                let mut current_delay = delay.lock().await;

                                tracing::warn!(
                                    ?err,
                                    ?current_delay,
                                    "A potentially-transient error occurred during sync. Retrying after delay.."
                                );

                                tokio::time::sleep(*current_delay).await;

                                *current_delay = std::cmp::min(*current_delay * 2, SYNC_MAX_DELAY_DURATION);

                                Ok(LoopCtrl::Continue)
                            }
                        }
                    }
                }
            })
            .await
            .map_err(SyncError::Sdk)?;

        Ok(())
    }
}