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

//! Conflict-free replicated datatypes.

use std::fmt::Debug;
use std::iter;
use {Frame, FrameOrd, Heap, Op, Terminator, UUID};

mod lww;
pub use self::lww::LWW;

mod set;
pub use self::set::Set;

/// Operations common to all Conflict-free Replicated Datatypes modeled by RON.
pub trait CRDT {
    /// Rust type this CRDT can be mapped to.
    type T;

    /// Returns the state Frame of a new, empty CRDT instance with UUID `obj`.
    fn new<'a>(obj: UUID) -> Frame<'a>;
    /// Reduce state Frame `state` and update Frames `updates` to a new state Frame.
    fn reduce<'a>(
        state: Frame<'a>, updates: Vec<Frame<'a>>,
    ) -> Option<Frame<'a>>;
    /// Maps the state Frame `state` to a Rust type.
    fn map<'a>(state: Frame<'a>) -> Option<Self::T>;
}

fn merge<'a, F>(state: Frame<'a>, updates: Vec<Frame<'a>>) -> Option<Frame<'a>>
where
    F: FrameOrd<'a> + Debug,
{
    let (ty, obj, full_state) = match state.peek() {
        Some(&Op { ref ty, ref object, ref location, .. }) => {
            (ty.clone(), object.clone(), location.is_zero())
        }
        None => {
            return None;
        }
    };
    let (min, max) = {
        let mut events = iter::once(&state)
            .chain(updates.iter())
            .filter_map(|frm| frm.peek().as_ref().map(|op| op.event.clone()));
        let mut min = events
            .clone()
            .min_by(|a, b| UUID::weak_cmp(a, b))
            .unwrap_or_else(UUID::zero);
        let max = events
            .max_by(|a, b| UUID::weak_cmp(a, b))
            .unwrap_or_else(UUID::zero);

        (min, max)
    };
    let mut heap = Heap::<F>::new(
        iter::once(state).chain(updates.into_iter()).collect::<Vec<_>>(),
    );
    let mut out = vec![Op {
        ty: ty,
        object: obj,
        event: max,
        location: if full_state { UUID::zero() } else { min },
        term: Terminator::Header,
        atoms: Default::default(),
    }];

    while let Some(mut op) = heap.pop() {
        op.term = Terminator::Reduced;
        out.push(op);
    }

    Some(Frame::compress(out))
}