dittolive_ditto/sync/
sync_subscription.rs

1use std::{
2    cmp::Ordering,
3    hash::{self, Hash},
4};
5
6use ffi_sdk::{
7    ffi_utils::{c_slice, char_p, repr_c},
8    FfiSyncSubscription,
9};
10
11use crate::{
12    ditto::Ditto,
13    error::DittoError,
14    utils::{extension_traits::FfiResultIntoRustResult, zstr::zstr},
15};
16
17/// Use [`ditto.sync().register_subscription(...)`] to create a `SyncSubscription`
18///
19/// The subscription will remain active until either:
20///
21/// - the `SyncSubscription` is explicitly canceled via [`.cancel()`], or
22/// - the owning [`Ditto`] object goes out of scope
23///
24/// [See the `sync` module documentation for more details][0].
25///
26/// [`ditto.sync().register_subscription(...)`]: crate::sync::Sync::register_subscription
27/// [`.cancel()`]: Self::cancel
28/// [0]: crate::sync
29pub struct SyncSubscription {
30    pub(crate) handle: repr_c::Box<FfiSyncSubscription>,
31}
32
33impl SyncSubscription {
34    pub(crate) fn new(
35        ditto: &Ditto,
36        query: &zstr,
37        query_args: Option<&[u8]>,
38    ) -> Result<Self, DittoError> {
39        let handle = ffi_sdk::dittoffi_sync_register_subscription_throws(
40            &ditto.ditto,
41            query.into(),
42            query_args.map(|a| a.into()),
43        )
44        .into_rust_result()?;
45
46        Ok(Self { handle })
47    }
48
49    /// The DQL query string of this [`SyncSubscription`] (as passed while
50    /// registering it).
51    ///
52    /// This is the original query string supplied when calling
53    /// [`ditto.sync().register_subscription(...)`], not the resolved query with
54    /// arguments substituted in.
55    ///
56    /// # Example
57    ///
58    /// ```
59    /// use dittolive_ditto::prelude::*;
60    ///
61    /// # fn main() -> anyhow::Result<()> {
62    /// # let (_root, ditto) = dittolive_ditto::doctest_helpers::doctest_ditto();
63    /// let sync_subscription = ditto.sync().register_subscription("SELECT * FROM cars")?;
64    ///
65    /// assert_eq!(sync_subscription.query_string(), "SELECT * FROM cars");
66    /// # Ok(())
67    /// # }
68    /// ```
69    ///
70    /// [`ditto.sync().register_subscription(...)`]: crate::sync::Sync::register_subscription
71    pub fn query_string(&self) -> String {
72        let cbox: char_p::Box = ffi_sdk::dittoffi_sync_subscription_query_string(&self.handle);
73        cbox.into_string()
74    }
75
76    /// The DQL query arguments of this [`SyncSubscription`] (as passed while
77    /// registering it).
78    ///
79    /// The returned value is not guaranteed to be strictly equal to the value
80    /// provided when calling [`ditto.sync().register_subscription(...)`], as
81    /// the arguments will have gone through a serialization roundtrip. This
82    /// might affect equality checks, particularly for non-primitive values. If
83    /// you want more control over how to deserialize the query arguments then
84    /// use [`query_arguments_cbor_data`](Self::query_arguments_cbor_data) or
85    /// [`query_arguments_json_str`](Self::query_arguments_json_str).
86    ///
87    /// # Example
88    ///
89    /// ```
90    /// use dittolive_ditto::prelude::*;
91    ///
92    /// # fn main() -> anyhow::Result<()> {
93    /// # let (_root, ditto) = dittolive_ditto::doctest_helpers::doctest_ditto();
94    /// let sync_subscription = ditto
95    ///     .sync()
96    ///     .register_subscription((
97    ///         "SELECT * FROM cars WHERE color = :color",
98    ///         serde_json::json!({
99    ///             "color": "red",
100    ///         })
101    ///     ))?;
102    ///
103    /// let maybe_args = sync_subscription.query_arguments();
104    /// let args = maybe_args.expect("expected query arguments");
105    /// let args_json = serde_json::to_value(&args)?;
106    ///
107    /// assert_eq!(args_json, serde_json::json!({
108    ///    "color": "red",
109    /// }));
110    ///
111    /// # Ok(())
112    /// # }
113    /// ```
114    ///
115    /// [`ditto.sync().register_subscription(...)`]: crate::sync::Sync::register_subscription
116    pub fn query_arguments(&self) -> Option<serde_cbor::Value> {
117        let buffer: c_slice::Box<u8> =
118            ffi_sdk::dittoffi_sync_subscription_query_arguments_cbor(&self.handle)?;
119
120        let cbor = serde_cbor::from_slice(buffer.as_slice())
121            .unwrap_or_else(|error| panic!("bug: failed to deserialize CBOR from FFI: {error}"));
122        Some(cbor)
123    }
124
125    /// Returns the DQL query arguments as raw CBOR-encoded bytes.
126    ///
127    /// # Example
128    ///
129    /// ```
130    /// use dittolive_ditto::prelude::*;
131    ///
132    /// # fn main() -> anyhow::Result<()> {
133    /// # let (_root, ditto) = dittolive_ditto::doctest_helpers::doctest_ditto();
134    /// let sync_subscription = ditto
135    ///     .sync()
136    ///     .register_subscription((
137    ///         "SELECT * FROM cars WHERE color = :color",
138    ///         serde_json::json!({
139    ///             "color": "red",
140    ///         })
141    ///     ))?;
142    ///
143    /// let maybe_args = sync_subscription.query_arguments_cbor_data();
144    /// let args = maybe_args.expect("expected query arguments");
145    /// let args_json: serde_json::Value = serde_cbor::from_slice(&args)?;
146    ///
147    /// assert_eq!(args_json, serde_json::json!({
148    ///    "color": "red",
149    /// }));
150    ///
151    /// # Ok(())
152    /// # }
153    /// ```
154    pub fn query_arguments_cbor_data(&self) -> Option<Vec<u8>> {
155        let buffer: c_slice::Box<u8> =
156            ffi_sdk::dittoffi_sync_subscription_query_arguments_cbor(&self.handle)?;
157        Some(buffer.as_slice().to_vec())
158    }
159
160    /// Returns the DQL query arguments as a JSON string.
161    ///
162    /// # Example
163    ///
164    /// ```
165    /// use dittolive_ditto::prelude::*;
166    ///
167    /// # fn main() -> anyhow::Result<()> {
168    /// # let (_root, ditto) = dittolive_ditto::doctest_helpers::doctest_ditto();
169    /// let args = serde_json::json!({
170    ///     "color": "red",
171    /// });
172    /// let query = (
173    ///     "SELECT * FROM cars WHERE color = :color",
174    ///     args.clone(),
175    /// );
176    /// let sync_subscription = ditto.sync().register_subscription(query)?;
177    ///
178    /// let maybe_args = sync_subscription.query_arguments_json_str();
179    /// let args_str = maybe_args.expect("expected query arguments");
180    ///
181    /// assert_eq!(args_str, serde_json::to_string(&args).unwrap());
182    ///
183    /// # Ok(())
184    /// # }
185    /// ```
186    pub fn query_arguments_json_str(&self) -> Option<String> {
187        let buffer: c_slice::Box<u8> =
188            ffi_sdk::dittoffi_sync_subscription_query_arguments_json(&self.handle)?;
189
190        let json = String::from_utf8(buffer.as_slice().to_vec())
191            .unwrap_or_else(|error| panic!("bug: failed to deserialize JSON from FFI: {error}"));
192        Some(json)
193    }
194
195    /// Cancels this [`SyncSubscription`], so that changes matching the query are no longer
196    /// synced from other peers to this one.
197    ///
198    /// # Example
199    ///
200    /// ```
201    /// use dittolive_ditto::Ditto;
202    ///
203    /// # fn main() -> anyhow::Result<()> {
204    /// # let (_root, ditto) = dittolive_ditto::doctest_helpers::doctest_ditto();
205    /// let subscription = ditto.sync().register_subscription("SELECT * FROM cars")?;
206    /// assert!(!subscription.is_cancelled());
207    ///
208    /// subscription.cancel();
209    /// assert!(subscription.is_cancelled());
210    /// # Ok(())
211    /// # }
212    /// ```
213    pub fn cancel(&self) {
214        ffi_sdk::dittoffi_sync_subscription_cancel(&self.handle);
215    }
216
217    /// Returns `true` if this [`SyncSubscription`] has been cancelled, `false` otherwise.
218    ///
219    /// # Example
220    ///
221    /// ```
222    /// use dittolive_ditto::Ditto;
223    ///
224    /// # fn main() -> anyhow::Result<()> {
225    /// # let (_root, ditto) = dittolive_ditto::doctest_helpers::doctest_ditto();
226    /// let subscription = ditto.sync().register_subscription("SELECT * FROM cars")?;
227    /// assert!(!subscription.is_cancelled());
228    ///
229    /// subscription.cancel();
230    /// assert!(subscription.is_cancelled());
231    /// # Ok(())
232    /// # }
233    /// ```
234    pub fn is_cancelled(&self) -> bool {
235        ffi_sdk::dittoffi_sync_subscription_is_cancelled(&self.handle)
236    }
237
238    /// Intentionally left non-public as the ID representation is an implementation detail
239    ///
240    /// The only reason this is here is to power the Ord/Hash/Debug impls
241    fn id(&self) -> impl '_ + Ord + Hash + core::fmt::Debug {
242        ffi_sdk::dittoffi_sync_subscription_id(&self.handle)
243    }
244}
245
246impl std::fmt::Debug for SyncSubscription {
247    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
248        f.debug_struct("SyncSubscription")
249            .field("id", &self.id())
250            .finish_non_exhaustive()
251    }
252}
253
254impl std::fmt::Display for SyncSubscription {
255    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
256        std::fmt::Debug::fmt(self, f)
257    }
258}
259
260impl Ord for SyncSubscription {
261    fn cmp(&self, other: &Self) -> Ordering {
262        Ord::cmp(&self.id(), &other.id())
263    }
264}
265
266impl PartialOrd for SyncSubscription {
267    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
268        Some(Ord::cmp(self, other))
269    }
270}
271
272impl Eq for SyncSubscription {}
273impl PartialEq for SyncSubscription {
274    fn eq(&self, other: &Self) -> bool {
275        self.id() == other.id()
276    }
277}
278
279impl Hash for SyncSubscription {
280    fn hash<H: hash::Hasher>(&self, h: &mut H) {
281        self.id().hash(h)
282    }
283}
dittolive_ditto/sync/sync_subscription.rs

dittolive_ditto/sync/
sync_subscription.rs
