struct_db 0.4.6

Drop-in embedded database
Documentation 
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

use crate::watch::filter::{KeyFilter, TableFilter};
use crate::watch::request::WatcherRequest;
use crate::watch::{Event, MpscSender};
use std::collections::HashMap;
use std::sync::{Arc, Mutex};

pub(crate) struct Watchers(HashMap<u64, (TableFilter, Arc<Mutex<MpscSender<Event>>>)>);

impl Watchers {
    pub(crate) fn new() -> Self {
        Self(HashMap::new())
    }

    pub(crate) fn add_sender(
        &mut self,
        id: u64,
        table_filter: &TableFilter,
        event_sender: Arc<Mutex<MpscSender<Event>>>,
    ) {
        self.0.insert(id, (table_filter.clone(), event_sender));
    }

    pub(crate) fn remove_sender(&mut self, id: u64) {
        self.0.remove(&id);
    }

    pub(crate) fn find_senders(
        &self,
        request: &WatcherRequest,
    ) -> Vec<Arc<Mutex<MpscSender<Event>>>> {
        let mut event_senders = Vec::new();
        for (_, (filter, event_sender)) in &self.0 {
            if filter.table_name == request.table_name {
                match &filter.key_filter {
                    KeyFilter::Primary(value) => {
                        if let Some(key) = &value {
                            if key == &request.primary_key {
                                event_senders.push(Arc::clone(event_sender));
                            }
                        } else {
                            event_senders.push(Arc::clone(event_sender));
                        }
                    }
                    KeyFilter::PrimaryStartWith(key_prefix) => {
                        if request.primary_key.starts_with(key_prefix) {
                            event_senders.push(Arc::clone(event_sender));
                        }
                    }
                    KeyFilter::Secondary(key_def, key) => {
                        for (request_secondary_key_def, request_secondary_key) in
                            &request.secondary_keys_value
                        {
                            if key_def == request_secondary_key_def.as_bytes() {
                                if let Some(value) = &key {
                                    if value == request_secondary_key {
                                        event_senders.push(Arc::clone(event_sender));
                                    }
                                } else {
                                    event_senders.push(Arc::clone(event_sender));
                                }
                            }
                        }
                    }
                    KeyFilter::SecondaryStartWith(key_def, key_prefix) => {
                        for (request_secondary_key_def, request_secondary_key) in
                            &request.secondary_keys_value
                        {
                            if key_def == request_secondary_key_def.as_bytes() {
                                if request_secondary_key.starts_with(key_prefix) {
                                    event_senders.push(Arc::clone(event_sender));
                                }
                            }
                        }
                    }
                }
            }
        }
        event_senders
    }
}