dittolive_ditto/store/query_builder/
subscription.rs

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
//! [`Subscription`]s are used to register interest in receiving updates
//! for specified [`DittoDocument`]s.

use_prelude!();

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

use ffi_sdk::COrderByParam;

use crate::error::{DittoError, ErrorKind};

/// Use [`cursor_op.subscribe()`] or [`id_op.subscribe()`] to sync documents from remote peers.
///
/// Holding a `Subscription` will cause documents matching the query to continue to sync
/// from remote peers.
///
/// Dropping a `Subscription` will cancel the sync of this query's documents from remote peers.
///
/// [`cursor_op.subscribe()`]: crate::store::query_builder::PendingCursorOperation::subscribe
/// [`id_op.subscribe()`]: crate::store::query_builder::PendingIdSpecificOperation::subscribe
#[must_use = "Dropping a `Subscription` will cancel it"]
pub struct Subscription {
    pub(super) ditto: Weak<BoxedDitto>,
    pub(super) collection_name: char_p::Box,
    pub(super) query: char_p::Box,
    pub(super) query_args: Option<Vec<u8>>,
    pub(super) order_by: Vec<(String, QuerySortDirection)>,
    pub(super) limit: i32,
    pub(super) offset: u32,
}

impl Subscription {
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn new(
        ditto: Arc<BoxedDitto>,
        collection_name: char_p::Box,
        query: &str,
        query_args: Option<Vec<u8>>,
        order_by: &'_ [COrderByParam<'_>],
        limit: i32,
        offset: u32,
    ) -> Self {
        let query = char_p::new(query);
        {
            let code = ffi_sdk::ditto_add_subscription(
                &ditto,
                collection_name.as_ref(),
                query.as_ref(),
                query_args.as_ref().map(|qa| (&qa[..]).into()),
                order_by.into(),
                limit,
                offset,
            );
            if code != 0 {
                error!(
                    collection = %collection_name,
                    %query,
                    error = %DittoError::from_ffi(ErrorKind::InvalidInput),
                    "error adding subscription",
                );
            }
        };
        let order_by: Vec<(String, QuerySortDirection)> = order_by
            .iter()
            // TODO: Doing `o.query_c_str.to_owned().into_string()` fails with
            // an InvalidNulTerminator error, but AFAICT it should be equivalent
            // to the below?
            .map(|o| (o.query_c_str.to_str_with_null().to_string(), o.direction))
            .collect();

        Subscription {
            ditto: Arc::downgrade(&ditto),
            collection_name,
            query,
            query_args,
            order_by,
            limit,
            offset,
        }
    }
}

#[allow(deprecated)]
impl crate::observer::Observer for Subscription {}

impl Drop for Subscription {
    fn drop(&mut self) {
        debug!(
            collection = %self.collection_name,
            query = %self.query,
            "dropping subscription",
        );
        if let Some(ditto) = self.ditto.upgrade() {
            {
                let order_by = self
                    .order_by
                    .iter()
                    .map(|query_and_direction| COrderByParam {
                        query_c_str: query_and_direction
                            .0
                            .as_str()
                            .try_into()
                            .expect("valid string"),
                        direction: query_and_direction.1,
                    })
                    .collect::<Vec<COrderByParam<'_>>>();

                let code = ffi_sdk::ditto_remove_subscription(
                    &ditto,
                    self.collection_name.as_ref(),
                    self.query.as_ref(),
                    self.query_args.as_ref().map(|qa| (&qa[..]).into()),
                    order_by[..].into(),
                    self.limit,
                    self.offset,
                );
                if code != 0 {
                    error!(
                        collection = %self.collection_name,
                        query = %self.query,
                        error = %DittoError::from_ffi(ErrorKind::InvalidInput),
                        "error removing subscription",
                    );
                }
            }
        }
    }
}