1use std::{fmt::Debug, ops::Add};
2use sum_tree::{ContextLessSummary, Dimension, Edit, Item, KeyedItem, SumTree};
3
4pub trait Operation: Clone + Debug {
5 fn lamport_timestamp(&self) -> clock::Lamport;
6}
7
8#[derive(Clone, Debug)]
9struct OperationItem<T>(T);
10
11#[derive(Clone, Debug)]
12pub struct OperationQueue<T: Operation>(SumTree<OperationItem<T>>);
13
14#[derive(Clone, Copy, Debug, Default, Eq, Ord, PartialEq, PartialOrd)]
15pub struct OperationKey(clock::Lamport);
16
17#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
18pub struct OperationSummary {
19 pub key: OperationKey,
20 pub len: usize,
21}
22
23impl OperationKey {
24 pub fn new(timestamp: clock::Lamport) -> Self {
25 Self(timestamp)
26 }
27}
28
29impl<T: Operation> Default for OperationQueue<T> {
30 fn default() -> Self {
31 OperationQueue::new()
32 }
33}
34
35impl<T: Operation> OperationQueue<T> {
36 pub fn new() -> Self {
37 OperationQueue(SumTree::default())
38 }
39
40 pub fn len(&self) -> usize {
41 self.0.summary().len
42 }
43
44 pub fn is_empty(&self) -> bool {
45 self.len() == 0
46 }
47
48 pub fn insert(&mut self, mut ops: Vec<T>) {
49 ops.sort_by_key(|op| op.lamport_timestamp());
50 ops.dedup_by_key(|op| op.lamport_timestamp());
51 self.0.edit(
52 ops.into_iter()
53 .map(|op| Edit::Insert(OperationItem(op)))
54 .collect(),
55 (),
56 );
57 }
58
59 pub fn drain(&mut self) -> Self {
60 let clone = self.clone();
61 self.0 = SumTree::default();
62 clone
63 }
64
65 pub fn iter(&self) -> impl Iterator<Item = &T> {
66 self.0.iter().map(|i| &i.0)
67 }
68}
69
70impl ContextLessSummary for OperationSummary {
71 fn zero() -> Self {
72 Default::default()
73 }
74
75 fn add_summary(&mut self, other: &Self) {
76 assert!(self.key < other.key);
77 self.key = other.key;
78 self.len += other.len;
79 }
80}
81
82impl Add<&Self> for OperationSummary {
83 type Output = Self;
84
85 fn add(self, other: &Self) -> Self {
86 assert!(self.key < other.key);
87 OperationSummary {
88 key: other.key,
89 len: self.len + other.len,
90 }
91 }
92}
93
94impl Dimension<'_, OperationSummary> for OperationKey {
95 fn zero(_cx: ()) -> Self {
96 Default::default()
97 }
98
99 fn add_summary(&mut self, summary: &OperationSummary, _: ()) {
100 assert!(*self <= summary.key);
101 *self = summary.key;
102 }
103}
104
105impl<T: Operation> Item for OperationItem<T> {
106 type Summary = OperationSummary;
107
108 fn summary(&self, _cx: ()) -> Self::Summary {
109 OperationSummary {
110 key: OperationKey::new(self.0.lamport_timestamp()),
111 len: 1,
112 }
113 }
114}
115
116impl<T: Operation> KeyedItem for OperationItem<T> {
117 type Key = OperationKey;
118
119 fn key(&self) -> Self::Key {
120 OperationKey::new(self.0.lamport_timestamp())
121 }
122}
123
124#[cfg(test)]
125mod tests {
126 use super::*;
127
128 #[test]
129 fn test_len() {
130 let mut clock = clock::Lamport::new(0);
131
132 let mut queue = OperationQueue::new();
133 assert_eq!(queue.len(), 0);
134
135 queue.insert(vec![
136 TestOperation(clock.tick()),
137 TestOperation(clock.tick()),
138 ]);
139 assert_eq!(queue.len(), 2);
140
141 queue.insert(vec![TestOperation(clock.tick())]);
142 assert_eq!(queue.len(), 3);
143
144 drop(queue.drain());
145 assert_eq!(queue.len(), 0);
146
147 queue.insert(vec![TestOperation(clock.tick())]);
148 assert_eq!(queue.len(), 1);
149 }
150
151 #[derive(Clone, Debug, Eq, PartialEq)]
152 struct TestOperation(clock::Lamport);
153
154 impl Operation for TestOperation {
155 fn lamport_timestamp(&self) -> clock::Lamport {
156 self.0
157 }
158 }
159}