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 129 130 131 132 133 134 135 136 137 138
use fxhash::FxHashMap; use multimap::MultiMap; use super::event::*; use super::header::*; use super::meta::*; use super::sink::*; use super::error::*; pub trait EventIndexer where Self: EventSink + Send + Sync + std::fmt::Debug, { fn rebuild(&mut self, _data: &Vec<EventHeader>) -> Result<(), SinkError> { Ok(()) } fn clone_indexer(&self) -> Box<dyn EventIndexer>; } #[derive(Debug, Copy, Clone)] pub struct EventLeaf { pub record: super::crypto::Hash, pub created: u64, pub updated: u64, } #[derive(Default, Debug)] pub(crate) struct BinaryTreeIndexer { primary: FxHashMap<PrimaryKey, EventLeaf>, secondary: MultiMap<MetaCollection, PrimaryKey>, } impl BinaryTreeIndexer { #[allow(dead_code)] pub(crate) fn contains_key(&self, key: &PrimaryKey) -> bool { self.primary.contains_key(key) } #[allow(dead_code)] pub(crate) fn count(&self) -> usize { self.primary.iter().count() } #[allow(dead_code)] pub(crate) fn feed(&mut self, entry: &EventHeader) { for core in entry.meta.core.iter() { match core { CoreMetadata::Tombstone(key) => { self.primary.remove(&key); if let Some(tree) = entry.meta.get_tree() { if let Some(vec) = self.secondary.get_vec_mut(&tree.vec) { vec.retain(|x| *x != *key); } } return; }, _ => { }, } } for core in entry.meta.core.iter() { match core { CoreMetadata::Data(key) => { if entry.raw.data_hash.is_none() { continue; } let when = entry.meta.get_timestamp(); let v = self.primary.entry(key.clone()).or_insert(EventLeaf { record: crate::crypto::Hash { val: [0; 16] }, created: match when { Some(t) => t.time_since_epoch_ms, None => 0 }, updated: 0, }); v.record = entry.raw.event_hash.clone(); v.updated = match when { Some(t) => t.time_since_epoch_ms, None => 0 }; }, CoreMetadata::Tree(tree) => { if let Some(key) = entry.meta.get_data_key() { let vec = tree.vec.clone(); let exists = match self.secondary.get_vec(&vec) { Some(a) => a.contains(&key), None => false, }; if exists == false { self.secondary.insert(vec, key); } } } _ => { }, } } } pub(crate) fn lookup_primary(&self, key: &PrimaryKey) -> Option<EventLeaf> { match self.primary.get(key) { None => None, Some(a) => Some(a.clone()) } } pub(crate) fn lookup_secondary(&self, key: &MetaCollection) -> Option<Vec<EventLeaf>> { match self.secondary.get_vec(key) { Some(vec) => { Some(vec.iter() .map(|a| a.clone()) .filter_map(|a| self.primary.get(&a)) .map(|a| a.clone()) .collect::<Vec<_>>()) }, None => None, } } } #[derive(Default, Debug)] pub struct UselessIndexer { } impl EventSink for UselessIndexer { } impl EventIndexer for UselessIndexer { fn clone_indexer(&self) -> Box<dyn EventIndexer> { Box::new(UselessIndexer::default()) } fn rebuild(&mut self, _headers: &Vec<EventHeader>) -> Result<(), SinkError> { Ok(()) } }