dittolive_ditto/sync/

sync_subscription.rs

use std::{
    cmp::Ordering,
    hash::{self, Hash},
};

use ffi_sdk::{
    ffi_utils::{c_slice, char_p, repr_c},
    FfiSyncSubscription,
};

use crate::{
    ditto::Ditto,
    error::DittoError,
    utils::{extension_traits::FfiResultIntoRustResult, zstr::zstr},
};

/// Use [`ditto.sync().register_subscription(...)`] to create a `SyncSubscription`
///
/// The subscription will remain active until either:
///
/// - the `SyncSubscription` is explicitly canceled via [`.cancel()`], or
/// - the owning [`Ditto`] object goes out of scope
///
/// [See the `sync` module documentation for more details][0].
///
/// [`ditto.sync().register_subscription(...)`]: crate::sync::Sync::register_subscription
/// [`.cancel()`]: Self::cancel
/// [0]: crate::sync
pub struct SyncSubscription {
    pub(crate) handle: repr_c::Box<FfiSyncSubscription>,
}

impl SyncSubscription {
    pub(crate) fn new(
        ditto: &Ditto,
        query: &zstr,
        query_args: Option<&[u8]>,
    ) -> Result<Self, DittoError> {
        let handle = ffi_sdk::dittoffi_sync_register_subscription_throws(
            &ditto.ditto,
            query.into(),
            query_args.map(|a| a.into()),
        )
        .into_rust_result()?;

        Ok(Self { handle })
    }

    /// Returns the DQL query string that this [`SyncSubscription`] is subscribed to.
    ///
    /// # Example
    ///
    /// ```
    /// use dittolive_ditto::prelude::*;
    ///
    /// # fn example(ditto: &Ditto) -> anyhow::Result<()> {
    /// let sync_subscription = ditto
    ///     .sync()
    ///     .register_subscription("SELECT * FROM cars", None)?;
    ///
    /// assert_eq!(sync_subscription.query_string(), "SELECT * FROM cars");
    /// # Ok(())
    /// # }
    /// ```
    pub fn query_string(&self) -> String {
        let cbox: char_p::Box = ffi_sdk::dittoffi_sync_subscription_query_string(&self.handle);
        cbox.into_string()
    }

    /// Returns the DQL query arguments that this [`SyncSubscription`] is subscribed to.
    ///
    /// # Example
    ///
    /// ```
    /// use dittolive_ditto::prelude::*;
    ///
    /// # fn example(ditto: &Ditto) -> anyhow::Result<()> {
    /// let sync_subscription = ditto
    ///     .sync()
    ///     .register_subscription("SELECT * FROM cars", Some(serde_json::json!({
    ///         "color": "red",
    ///     }).into()))?;
    ///
    /// let maybe_args = sync_subscription.query_arguments();
    /// let args = maybe_args.expect("expected query arguments");
    /// let args_json = serde_json::to_value(&args)?;
    ///
    /// assert_eq!(args_json, serde_json::json!({
    ///    "color": "red",
    /// }));
    ///
    /// # Ok(())
    /// # }
    /// ```
    pub fn query_arguments(&self) -> Option<serde_cbor::Value> {
        let buffer: c_slice::Box<u8> =
            ffi_sdk::dittoffi_sync_subscription_query_arguments(&self.handle)?;

        let cbor = serde_cbor::from_slice(buffer.as_slice())
            .unwrap_or_else(|error| panic!("bug: failed to deserialize CBOR from FFI: {error}"));
        Some(cbor)
    }

    /// Cancels this [`SyncSubscription`], so that changes matching the query are no longer
    /// synced from other peers to this one.
    ///
    /// # Example
    ///
    /// ```
    /// use dittolive_ditto::Ditto;
    ///
    /// # fn example(ditto: &Ditto) -> anyhow::Result<()> {
    /// let subscription = ditto
    ///     .sync()
    ///     .register_subscription("SELECT * FROM cars", None)?;
    /// assert!(!subscription.is_cancelled());
    ///
    /// subscription.cancel();
    /// assert!(subscription.is_cancelled());
    /// # Ok(())
    /// # }
    /// ```
    pub fn cancel(&self) {
        ffi_sdk::dittoffi_sync_subscription_cancel(&self.handle);
    }

    /// Returns `true` if this [`SyncSubscription`] has been cancelled, `false` otherwise.
    ///
    /// # Example
    ///
    /// ```
    /// use dittolive_ditto::Ditto;
    ///
    /// # fn example(ditto: &Ditto) -> anyhow::Result<()> {
    /// let subscription = ditto
    ///     .sync()
    ///     .register_subscription("SELECT * FROM cars", None)?;
    /// assert!(!subscription.is_cancelled());
    ///
    /// subscription.cancel();
    /// assert!(subscription.is_cancelled());
    /// # Ok(())
    /// # }
    /// ```
    pub fn is_cancelled(&self) -> bool {
        ffi_sdk::dittoffi_sync_subscription_is_cancelled(&self.handle)
    }

    /// Intentionally left non-public as the ID representation is an implementation detail
    ///
    /// The only reason this is here is to power the Ord/Hash/Debug impls
    fn id(&self) -> impl '_ + Ord + Hash + core::fmt::Debug {
        ffi_sdk::dittoffi_sync_subscription_id(&self.handle)
    }
}

impl std::fmt::Debug for SyncSubscription {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SyncSubscription")
            .field("id", &self.id())
            .finish_non_exhaustive()
    }
}

impl std::fmt::Display for SyncSubscription {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        std::fmt::Debug::fmt(self, f)
    }
}

impl Ord for SyncSubscription {
    fn cmp(&self, other: &Self) -> Ordering {
        Ord::cmp(&self.id(), &other.id())
    }
}

impl PartialOrd for SyncSubscription {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(Ord::cmp(self, other))
    }
}

impl Eq for SyncSubscription {}
impl PartialEq for SyncSubscription {
    fn eq(&self, other: &Self) -> bool {
        self.id() == other.id()
    }
}

impl Hash for SyncSubscription {
    fn hash<H: hash::Hasher>(&self, h: &mut H) {
        self.id().hash(h)
    }
}