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 example(ditto: &Ditto) -> anyhow::Result<()> {
//!
//! let sync_subscription = ditto
//!     .sync()
//!     .register_subscription("SELECT * FROM cars WHERE color = 'blue'", None)?;
//!
//! // 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,
    ops::Deref,
    sync::{Arc, Weak},
};

use serde::Serialize;

pub use self::sync_subscription::SyncSubscription;

mod sync_subscription;

use crate::{
    ditto::{Ditto, DittoFields},
    dql::{query_v2::IntoQuery, *},
    error::DittoError,
    utils::SetArc,
};

/// 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 map will not cause the
    /// underlying subscriptions to be updated.
    ///
    /// # Example
    ///
    /// ```
    /// # use dittolive_ditto::Ditto;
    /// # fn example(ditto: &Ditto) -> anyhow::Result<()> {
    /// let subscription = ditto
    ///     .sync()
    ///     .register_subscription("SELECT * FROM cars", None)?;
    /// let subscriptions = ditto.sync().subscriptions();
    /// assert!(subscriptions.contains(&subscription));
    /// # Ok(())
    /// # }
    /// ```
    #[doc(hidden)]
    #[deprecated(note = "Use `ditto.sync().subscriptions_v2()` instead.")]
    pub fn subscriptions(&self) -> impl '_ + Deref<Target = SetArc<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();

        let sync_subscriptions = sync_subscriptions
            .into_iter()
            .map(|handle| Arc::new(SyncSubscription { handle }))
            .collect::<SetArc<_>>();

        Box::new(sync_subscriptions)
    }

    /// Returns a snapshot of handles to all active [`SyncSubscription`]s.
    ///
    /// Adding or removing [`SyncSubscription`]s from this map will not cause the
    /// underlying subscriptions to be updated.
    ///
    /// # Example
    ///
    /// ```
    /// # use dittolive_ditto::Ditto;
    /// # fn example(ditto: &Ditto) -> anyhow::Result<()> {
    /// let subscription = ditto
    ///     .sync()
    ///     .register_subscription("SELECT * FROM cars", None)?;
    /// let subscriptions = ditto.sync().subscriptions_v2();
    /// assert!(subscriptions.contains(&subscription));
    /// # Ok(())
    /// # }
    /// ```
    pub fn subscriptions_v2(&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<_>>()
    }

    /// Deprecated in favour of
    /// [`ditto.sync().register_subscription_v2()`][Sync::register_subscription_v2]
    #[deprecated = "Use `ditto.sync().register_subscription_v2(...) instead"]
    #[allow(deprecated)]
    #[doc(hidden)]
    pub fn register_subscription<Q>(
        &self,
        query: Q,
        query_args: Option<QueryArguments>,
    ) -> Result<Arc<SyncSubscription>, DittoError>
    where
        Q: TryInto<Query, Error = DittoError>,
    {
        let ditto = Ditto::upgrade(&self.ditto)?;
        let query: Query = query.try_into()?;
        let query_args = query_args.as_ref().map(|a| a.cbor());
        let subscription = SyncSubscription::new(&ditto, &query.inner_string, query_args)?;
        let subscription = Arc::new(subscription);
        Ok(subscription)
    }

    /// 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 example(ditto: &Ditto) -> anyhow::Result<()> {
    /// let query = (
    ///     "SELECT * FROM cars WHERE color = :color",
    ///     serde_json::json!({"color": "blue"}),
    /// );
    ///
    /// let sync_subscription = ditto.sync().register_subscription_v2(query)?;
    ///
    /// // Cancel the subscription with `.cancel()`
    /// sync_subscription.cancel();
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// [`sync_subscription.cancel()`]: crate::sync::SyncSubscription::cancel
    pub fn register_subscription_v2<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)
    }
}