Operation-based Counter

cmrdt.rs

use std::borrow::BorrowMut;
use std::cell::{Ref, RefCell};
use std::ops::Add;
use std::rc::Rc;

use crate::cmrdts::CmRDTOp::ADD;

#[derive(Copy, Clone)]
pub enum CmRDTOp<T> {
    ADD(T)
}

type SiteId = i32;

pub trait CmRDT {
    type Item;

    fn value(&self) -> Self::Item;

    fn op(&self, op: CmRDTOp<Self::Item>);
}

pub trait CmRDTSite {
    type Item;

    fn site_id(&self) -> SiteId;

    fn consume(&self, op: CmRDTOp<Self::Item>);

    fn produce_to(instance: &Rc<GCounter>, router: &Rc<OpsRouter>);
}


pub trait Effector<S, T> {
    fn eval(&self, prev_state: &S, input: &T) -> S;
}

struct AddEffector {}

impl Effector<i32, i32> for AddEffector {
    fn eval(&self, prev_state: &i32, input: &i32) -> i32 {
        prev_state + input
    }
}


pub struct GCounter {
    id: SiteId,
    ops: RefCell<Vec<CmRDTOp<i32>>>,
    router: RefCell<Option<Rc<OpsRouter>>>,
}

impl GCounter {
    pub fn new(id: i32) -> GCounter {
        GCounter {
            id,
            ops: RefCell::new(vec![]),
            router: RefCell::new(None),
        }
    }
}

impl CmRDT for GCounter {
    type Item = i32;

    fn value(&self) -> Self::Item {
        self.ops.borrow().iter().fold(0, |acc, op|
            match op {
                ADD(value) => AddEffector {}.eval(&acc, value)
            },
        )
    }

    fn op(&self, op: CmRDTOp<Self::Item>) {
        self.ops.borrow_mut().push(op);

        if let Some(router) = self.router.borrow().as_ref() {
            router.notify_all(self.site_id(), op);
        }
    }
}

impl CmRDTSite for GCounter {
    type Item = i32;

    fn site_id(&self) -> SiteId {
        self.id
    }

    fn consume(&self, op: CmRDTOp<Self::Item>) {
        self.ops.borrow_mut().push(op);
    }

    fn produce_to(instance: &Rc<GCounter>, router: &Rc<OpsRouter>) {
        instance.router.replace(Some(Rc::clone(router)));

        if let Some(router) = instance.router.borrow().as_ref() {
            router.subscribe_g_counter(instance);
        }
    }
}


pub struct OpsRouter {
    g_counters: RefCell<Vec<Rc<GCounter>>>,
}

impl OpsRouter {
    pub fn new() -> OpsRouter {
        OpsRouter {
            g_counters: RefCell::new(vec![])
        }
    }

    pub fn subscribe_g_counter(&self, g_counter: &Rc<GCounter>) {
        self.g_counters.borrow_mut().push(Rc::clone(g_counter));
    }

    fn notify_all(&self, producer_site_id: SiteId, op: CmRDTOp<i32>) {
        self.g_counters.borrow()
            .iter()
            .filter(|g_counter| g_counter.id != producer_site_id)
            .for_each(|ggis_counter| g_counter.consume(op))
    }
}