dittolive_ditto/store/attachment/
fetcher.rs

1use_prelude!();
2
3use core::ffi::c_void;
4use std::sync::{
5    atomic::{self, AtomicU64},
6    Arc, Weak,
7};
8
9use ffi_sdk::BoxedAttachmentHandle;
10
11use crate::{
12    error::DittoError,
13    store::{
14        ditto_attachment::DittoAttachment, ditto_attachment_fetch_event::DittoAttachmentFetchEvent,
15        ditto_attachment_token::DittoAttachmentToken, DittoFields,
16    },
17};
18
19/// Type-level `enum` to distinguish between a [`DittoAttachmentFetcher`] that
20/// can and must be cancelled, and the legacy behavior where it was auto-cancelled on drop.
21///
22/// ```rust ,ignore
23/// // It represents the following, but at the type-level ("const generic" sort to speak).
24/// enum FetcherVersion /* : Sealed */ {
25///     /// The deprecated `.collection.fetch_attachment(...)` API.
26///     V1,
27///
28///     /// The proper `.store.fetch_attachment(...)` API.
29///     V2,
30/// }
31/// ```
32#[allow(nonstandard_style)]
33pub mod FetcherVersion {
34    /// The deprecated
35    /// [`collection.fetch_attachment(...)`][crate::store::Collection::fetch_attachment()] API.
36    pub enum V1 {}
37    /// The proper [`store.fetch_attachment(...)`][crate::store::Store::fetch_attachment()] API.
38    pub enum V2 {}
39
40    mod seal {
41        pub trait Sealed: 'static {}
42    }
43    pub(crate) use seal::Sealed;
44
45    impl Sealed for V1 {}
46    impl Sealed for V2 {}
47}
48
49/// The output of [`store.fetch_attachment()`](crate::store::Store::fetch_attachment).
50///
51///   - In the deprecated [`FetcherVersion::V1`] case, they must be kept alive for the fetching of
52///     the attachment to proceed and for you to be notified once the status of the fetch request
53///     has changed.
54///
55///   - In the proper [`FetcherVersion::V2`] case, they are only
56///     [`.cancel()`][DittoAttachmentFetcher::cancel]-ed *explicitly*, that is, they are safe to
57///     discard / not to safe-keep.
58pub struct DittoAttachmentFetcher<'a, Version: FetcherVersion::Sealed = FetcherVersion::V1> {
59    pub(crate) context: Arc<DittoAttachmentFetcherCtx<'a>>,
60    _phantom: ::core::marker::PhantomData<fn() -> Version>,
61}
62
63pub(crate) type DittoAttachmentFetcherV2 = DittoAttachmentFetcher<'static, FetcherVersion::V2>;
64
65impl Clone for DittoAttachmentFetcherV2 {
66    fn clone(&self) -> Self {
67        Self {
68            context: self.context.retain(),
69            ..*self
70        }
71    }
72}
73
74impl<'a, Version: FetcherVersion::Sealed> DittoAttachmentFetcher<'a, Version> {
75    pub(crate) fn new(
76        token: DittoAttachmentToken,
77        ditto: Option<&Arc<DittoFields>>,
78        raw_ditto: &Arc<BoxedDitto>,
79        on_fetch_event: impl 'a + Send + Sync + Fn(DittoAttachmentFetchEvent, &AtomicU64),
80    ) -> Result<Self, DittoError> {
81        let context = Arc::new(DittoAttachmentFetcherCtx {
82            token: token.clone(),
83            ditto: ditto.map_or_else(::std::sync::Weak::new, Arc::downgrade),
84            raw_ditto: Arc::downgrade(raw_ditto),
85            on_fetch_event: Box::new(on_fetch_event),
86            cancel_token: 0.into(),
87        });
88        let raw_context = Arc::as_ptr(&context) as *mut c_void;
89
90        let cancel_token = unsafe {
91            ffi_sdk::ditto_resolve_attachment(
92                raw_ditto,
93                token.id.as_ref().into(),
94                raw_context,
95                Some(DittoAttachmentFetcherCtx::retain),
96                Some(DittoAttachmentFetcherCtx::release),
97                DittoAttachmentFetcherCtx::on_complete_cb,
98                DittoAttachmentFetcherCtx::on_progress_cb,
99                DittoAttachmentFetcherCtx::on_deleted_cb,
100            )
101            .ok()?
102        };
103        // HACK: until `dittoffi` exposes the `cancel_token` / `resolve_id`
104        // to the callback itself, we manually back-smuggle it using the shared
105        // context.
106        //
107        // This value is later accessed in the callback registered by
108        // `Store::fetch_attachment()`, in the `on_complete` case, after having
109        // acquired a lock which happens to be held during this whole `new()` call,
110        // hence why `Relaxed` suffices.
111        context
112            .cancel_token
113            .store(cancel_token as _, atomic::Ordering::Relaxed);
114
115        Ok(Self {
116            context,
117            _phantom: ::core::marker::PhantomData,
118        })
119    }
120}
121
122impl<Version: FetcherVersion::Sealed> DittoAttachmentFetcher<'_, Version> {
123    pub(crate) fn cancel_token(&self) -> u64 {
124        self.context.cancel_token.load(atomic::Ordering::SeqCst)
125    }
126}
127
128impl DittoAttachmentFetcherV2 {
129    /// Yields `true` if the original `cancel_token` was `0`.
130    pub(crate) fn cancel_token_ensure_unique(&self) -> (u64, bool) {
131        let mut cancel_token = self.cancel_token();
132        let was_zero = cancel_token == 0;
133        if was_zero {
134            // Currently, FFI `.resolve_attachment()` will yield `0` for fetchers
135            // it believes to have invoked synchronously (fast path); even
136            // though our callback is —outside of Wasm— dispatching these
137            // invocations onto the `attachments_signal_sender` queue.
138            //
139            // So, to keep the property of unicity of these tokens across
140            // the lifetime of a ditto instance (TODO: process-wide?), we
141            // currently hack a fallback unicity mechanism. The `core`
142            // auto-increments off `1..`, so let us do `(..-2).rev()`.
143            static FALLBACK_UNIQUE_CANCEL_TOKEN: AtomicU64 = AtomicU64::new(u64::MAX - 1);
144            cancel_token =
145                FALLBACK_UNIQUE_CANCEL_TOKEN.fetch_sub(1, ::std::sync::atomic::Ordering::Relaxed);
146            self.context
147                .cancel_token
148                .store(cancel_token, atomic::Ordering::SeqCst);
149        };
150        (cancel_token, was_zero)
151    }
152}
153
154impl DittoAttachmentFetcherV2 {
155    /// Stops fetching the fetcher's associated attachment and cleans up any
156    /// associated resources.
157    ///
158    /// Note that you are not required to call it once your attachment
159    /// fetch operation has finished. The method primarily exists to allow you
160    /// to cancel an attachment fetch request while it is ongoing if you no
161    /// longer wish for the attachment to be made available locally to the
162    /// device nor for its evolution to be observed.
163    pub fn cancel(&self) {
164        if let Some(ditto) = self.context.ditto.upgrade() {
165            ditto.store.unregister_fetcher(self.cancel_token(), None);
166        }
167    }
168}
169
170#[allow(deprecated)]
171impl crate::observer::Observer for DittoAttachmentFetcher<'_, FetcherVersion::V1> {}
172
173// Does nothing for V2.
174impl<V1: FetcherVersion::Sealed> Drop for DittoAttachmentFetcher<'_, V1> {
175    fn drop(&mut self) {
176        /// Poorman's specialization.
177        use core::any::TypeId;
178
179        if TypeId::of::<V1>() == TypeId::of::<FetcherVersion::V1>() {
180            if let Some(ditto) = self.context.raw_ditto.upgrade() {
181                let status = ffi_sdk::ditto_cancel_resolve_attachment(
182                    &ditto,
183                    self.context.token.id.as_ref().into(),
184                    self.cancel_token(),
185                );
186                if status != 0 {
187                    error!("failed to clean up attachment fetcher");
188                }
189            }
190        }
191    }
192}
193
194pub(crate) struct DittoAttachmentFetcherCtx<'a> {
195    pub(crate) token: DittoAttachmentToken,
196    ditto: Weak<DittoFields>,
197    raw_ditto: Weak<BoxedDitto>,
198    #[allow(clippy::type_complexity)]
199    on_fetch_event: Box<dyn Fn(DittoAttachmentFetchEvent, &AtomicU64) + Send + Sync + 'a>,
200    cancel_token: atomic::AtomicU64,
201}
202
203impl DittoAttachmentFetcherCtx<'_> {
204    pub(crate) unsafe extern "C" fn retain(ctx: *mut c_void, _: Blocking) {
205        let ptr = ctx.cast::<DittoAttachmentFetcherCtx<'_>>();
206        Arc::increment_strong_count(ptr);
207    }
208
209    pub(crate) unsafe extern "C" fn release(ctx: *mut c_void, _: Blocking) {
210        let ptr = ctx.cast::<DittoAttachmentFetcherCtx<'_>>();
211        Arc::decrement_strong_count(ptr);
212    }
213
214    pub(crate) unsafe extern "C" fn on_complete_cb(
215        ctx: *mut c_void,
216        attachment_handle: BoxedAttachmentHandle,
217        _: Blocking,
218    ) {
219        let ctx_ref = ctx
220            .cast::<DittoAttachmentFetcherCtx<'_>>()
221            .as_ref()
222            .expect("got null");
223
224        let ditto_attachment = DittoAttachment::new_with_token(
225            ctx_ref.token.clone(),
226            ctx_ref.raw_ditto.clone(),
227            attachment_handle,
228        );
229        let event = DittoAttachmentFetchEvent::Completed {
230            attachment: ditto_attachment,
231        };
232        (ctx_ref.on_fetch_event)(event, &ctx_ref.cancel_token);
233    }
234
235    pub(crate) unsafe extern "C" fn on_progress_cb(
236        ctx: *mut c_void,
237        downloaded_bytes: u64,
238        total_bytes: u64,
239        _: Blocking,
240    ) {
241        let ctx_ref = ctx
242            .cast::<DittoAttachmentFetcherCtx<'_>>()
243            .as_ref()
244            .expect("got null");
245
246        let event = DittoAttachmentFetchEvent::Progress {
247            downloaded_bytes,
248            total_bytes,
249        };
250        (ctx_ref.on_fetch_event)(event, &ctx_ref.cancel_token);
251    }
252
253    pub(crate) unsafe extern "C" fn on_deleted_cb(ctx: *mut c_void, _: Blocking) {
254        let ctx_ref = ctx
255            .cast::<DittoAttachmentFetcherCtx<'_>>()
256            .as_ref()
257            .expect("got null");
258
259        let event = DittoAttachmentFetchEvent::Deleted;
260        (ctx_ref.on_fetch_event)(event, &ctx_ref.cancel_token);
261    }
262}
dittolive_ditto/store/attachment/fetcher.rs

dittolive_ditto/store/attachment/
fetcher.rs
