dittolive_ditto/sync/

mod.rs

//! Use [`ditto.sync()`] to access the [`Sync`] API of Ditto.
//!
//! The [`Sync`] API can be used to request documents to be synchronized with
//! other peers. Use [`ditto.sync().register_subscription(...)`] and provide
//! a DQL query to let Ditto know which documents you're interested in syncing.
//!
//! The returned [`SyncSubscription`] handle can be used to cancel the
//! subscription with [`.cancel()`], at which point Ditto will stop syncing
//! data for that subscription.
//!
//! # Example
//!
//! ```
//! use dittolive_ditto::prelude::*;
//! # fn main() -> anyhow::Result<()> {
//! # let (_root, ditto) = dittolive_ditto::doctest_helpers::doctest_ditto();
//!
//! let sync_subscription = ditto
//!     .sync()
//!     .register_subscription("SELECT * FROM cars WHERE color = 'blue'")?;
//!
//! // To cancel the sync subscription, use .cancel()
//! sync_subscription.cancel();
//! # Ok(())
//! # }
//! ```
//!
//! [`ditto.sync()`]: Ditto::sync
//! [`ditto.sync().register_subscription(...)`]: Sync::register_subscription
//! [`.cancel()`]: SyncSubscription::cancel

use std::{
    collections::HashSet,
    sync::{Arc, Weak},
};

use serde::Serialize;

pub use self::sync_subscription::SyncSubscription;

mod sync_subscription;

use crate::{
    ditto::{Ditto, DittoFields},
    dql::query::IntoQuery,
    error::DittoError,
};

/// Ditto's `Sync` API, obtained via [`ditto.sync()`].
///
/// [See the `sync` module documentation for more details][0].
///
/// [`ditto.sync()`]: crate::prelude::Ditto::sync
/// [0]: crate::sync
pub struct Sync {
    ditto: Weak<DittoFields>,
}

impl Sync {
    pub(crate) fn new(ditto: Weak<DittoFields>) -> Self {
        Self { ditto }
    }

    /// Returns a snapshot of handles to all active [`SyncSubscription`]s.
    ///
    /// Adding or removing [`SyncSubscription`]s from this set will not cause the
    /// underlying subscriptions to be updated.
    ///
    /// # Example
    ///
    /// ```
    /// # use dittolive_ditto::Ditto;
    /// # fn main() -> anyhow::Result<()> {
    /// # let (_root, ditto) = dittolive_ditto::doctest_helpers::doctest_ditto();
    /// let subscription = ditto.sync().register_subscription("SELECT * FROM cars")?;
    /// let subscriptions = ditto.sync().subscriptions();
    /// assert!(subscriptions.contains(&subscription));
    /// # Ok(())
    /// # }
    /// ```
    pub fn subscriptions(&self) -> HashSet<SyncSubscription> {
        let ditto = Ditto::upgrade(&self.ditto).expect("Ditto went out of scope");
        let sync_subscriptions = ffi_sdk::dittoffi_sync_subscriptions(&ditto.ditto);
        let sync_subscriptions: Vec<_> = sync_subscriptions.into();

        sync_subscriptions
            .into_iter()
            .map(|handle| SyncSubscription { handle })
            .collect::<HashSet<_>>()
    }

    /// Use a DQL query to subscribe to data on other Ditto peers.
    ///
    /// While the subscription is active, data matching this query on other
    /// peers will be synced to the local peer's data store.
    ///
    /// Note that dropping the `SyncSubscription` won't cancel it. To do that
    /// be sure to use [`sync_subscription.cancel()`].
    ///
    /// # Example
    ///
    /// ```
    /// use dittolive_ditto::prelude::*;
    /// # fn main() -> anyhow::Result<()> {
    /// # let (_root, ditto) = dittolive_ditto::doctest_helpers::doctest_ditto();
    /// let query = (
    ///     "SELECT * FROM cars WHERE color = :color",
    ///     serde_json::json!({"color": "blue"}),
    /// );
    ///
    /// let sync_subscription = ditto.sync().register_subscription(query)?;
    ///
    /// // Cancel the subscription with `.cancel()`
    /// sync_subscription.cancel();
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// [`sync_subscription.cancel()`]: crate::sync::SyncSubscription::cancel
    pub fn register_subscription<Q>(&self, query: Q) -> Result<Arc<SyncSubscription>, DittoError>
    where
        Q: IntoQuery,
        Q::Args: Serialize,
    {
        let ditto = Ditto::upgrade(&self.ditto)?;
        let query = query.into_query()?;
        let subscription =
            SyncSubscription::new(&ditto, &query.string, query.args_cbor.as_deref())?;
        let subscription = Arc::new(subscription);
        Ok(subscription)
    }

    /// Start syncing on all transports.
    ///
    /// By default, Ditto enables the stable peer-to-peer transports supported on
    /// the current platform. Reliable UDP multicast is excluded from this
    /// default and must be enabled explicitly. The network configuration can be
    /// customized using [`Ditto::set_transport_config`].
    ///
    /// # Example
    ///
    /// ```
    /// # use dittolive_ditto::Ditto;
    /// # use dittolive_ditto::prelude::DittoError;
    /// # fn main() -> Result<(), DittoError> {
    /// # let (_root, ditto) = dittolive_ditto::doctest_helpers::doctest_ditto();
    /// ditto.sync().start()?;
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Errors
    ///
    /// Returns an error if the Ditto instance has not been activated with a valid license.
    pub fn start(&self) -> Result<(), DittoError> {
        let ditto = Ditto::upgrade(&self.ditto)?;
        let result = ffi_sdk::dittoffi_ditto_try_start_sync(&ditto.ditto);
        if let Some(error) = result.error {
            if ffi_sdk::dittoffi_error_code(&*error)
                == ffi_sdk::FfiErrorCode::ActivationNotActivated
            {
                return Err(crate::error::ErrorKind::NotActivated.into());
            }
            return Err(DittoError::from(error));
        }
        Ok(())
    }

    /// Stop syncing on all transports.
    ///
    /// You may continue to use the Ditto store locally but no data
    /// will sync to or from other devices.
    ///
    /// # Example
    ///
    /// ```
    /// # use dittolive_ditto::Ditto;
    /// # let (_root, ditto) = dittolive_ditto::doctest_helpers::doctest_ditto();
    /// ditto.sync().stop();
    /// ```
    pub fn stop(&self) {
        let ditto = Ditto::upgrade(&self.ditto).expect("Ditto went out of scope");
        ffi_sdk::dittoffi_ditto_stop_sync(&ditto.ditto);
    }

    /// Returns `true` if sync is currently active, otherwise returns `false`.
    ///
    /// Use [`ditto.sync().start()`] to activate and [`ditto.sync().stop()`] to deactivate sync.
    ///
    /// [`ditto.sync().start()`]: Sync::start
    /// [`ditto.sync().stop()`]: Sync::stop
    ///
    /// # Example
    ///
    /// ```
    /// # use dittolive_ditto::Ditto;
    /// # let (_root, ditto) = dittolive_ditto::doctest_helpers::doctest_ditto();
    /// if ditto.sync().is_active() {
    ///     println!("Sync is active");
    /// }
    /// ```
    ///
    /// [`ditto.sync().stop()`]: Sync::stop
    pub fn is_active(&self) -> bool {
        let ditto = Ditto::upgrade(&self.ditto).expect("Ditto went out of scope");
        ffi_sdk::dittoffi_ditto_is_sync_active(&ditto.ditto)
    }
}