pollination 0.1.0

An experimental library exposing hybrid Raft and CRDT primitives.
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

use crate::constants;
use crate::message::PollinationMessage;
use std::time::Instant;
use tokio::sync::mpsc::{Sender, error::SendError};
use tracing::debug;
use treeclocks::{EventTree, IdTree};

#[derive(Debug)]
pub struct Connection {
    pub(crate) peer_id: Option<IdTree>,
    pub(crate) peer_ts: Option<EventTree>,
    pub(crate) prev_msg: Option<(PollinationMessage, Instant)>,
    tx: Sender<PollinationMessage>,
}

impl Connection {
    pub fn new(tx: Sender<PollinationMessage>) -> Self {
        Self {
            tx,
            prev_msg: None,
            peer_id: None,
            peer_ts: None,
        }
    }

    pub async fn send(
        &mut self,
        msg: PollinationMessage,
    ) -> Result<(), SendError<PollinationMessage>> {
        if self.debounce(&msg) {
            return Ok(());
        }

        self.prev_msg = Some((msg.light_clone(), Instant::now()));
        self.tx.send(msg).await
    }

    fn debounce(&self, msg: &PollinationMessage) -> bool {
        if let Some((prev_msg, timeout)) = &self.prev_msg {
            if timeout.elapsed() > constants::DEBOUNCE_TIMEOUT {
                return false;
            }

            use PollinationMessage::*;
            match (&prev_msg, msg) {
                (
                    Heartbeat {
                        timestamp: ts_old, ..
                    }
                    | Update {
                        timestamp: ts_old, ..
                    },
                    Heartbeat {
                        timestamp: ts_new, ..
                    },
                ) if ts_new <= ts_old => {
                    debug!("Skipping heartbeat since one already sent");
                    true
                }
                (
                    Update {
                        timestamp: ts_old, ..
                    },
                    Update {
                        timestamp: ts_new, ..
                    },
                ) if ts_new <= ts_old => {
                    debug!("Skipping update since one already sent");
                    true
                }
                _ => false,
            }
        } else {
            false
        }
    }
}