datacake-cluster 0.1.0

Eventually consistent state replication as a library (consensus, RPC and conflict resolution) for building your own eventually consistent databases.
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

use std::collections::HashMap;
use std::net::SocketAddr;
use std::str::FromStr;
use std::sync::Arc;
use std::time::Duration;

use parking_lot::RwLock;
use tonic::transport::{Channel, Endpoint, Error};

pub const TIMEOUT_LIMIT: Duration = Duration::from_secs(2);
pub const CONNECT_TIMEOUT_LIMIT: Duration = Duration::from_secs(2);

#[derive(Clone, Default)]
/// A collection of RPC client connections which can be reused and multiplexed.
pub struct RpcNetwork {
    clients: Arc<RwLock<HashMap<SocketAddr, Channel>>>,
}

impl RpcNetwork {
    /// Attempts to get an already existing connection or creates a new connection.
    pub async fn get_or_connect(&self, addr: SocketAddr) -> Result<Channel, Error> {
        {
            let guard = self.clients.read();
            if let Some(channel) = guard.get(&addr) {
                return Ok(channel.clone());
            }
        }

        trace!(addr = %addr, "Connect client to network.");
        self.connect(addr).await
    }

    /// Connects to a given address and adds it to the clients.
    pub async fn connect(&self, addr: SocketAddr) -> Result<Channel, Error> {
        let uri = format!("http://{}", addr);
        let channel = Endpoint::from_str(&uri)
            .unwrap()
            .timeout(TIMEOUT_LIMIT)
            .connect_timeout(CONNECT_TIMEOUT_LIMIT)
            .connect()
            .await?;

        {
            let mut guard = self.clients.write();
            guard.insert(addr, channel.clone());
        }

        Ok(channel)
    }
    /// Attempts to get an already existing connection or creates a new connection.
    pub fn get_or_connect_lazy(&self, addr: SocketAddr) -> Channel {
        {
            let guard = self.clients.read();
            if let Some(channel) = guard.get(&addr) {
                return channel.clone();
            }
        }

        self.connect_lazy(addr)
    }

    /// Creates a new endpoint channel which connects lazily to the node.
    pub fn connect_lazy(&self, addr: SocketAddr) -> Channel {
        let uri = format!("http://{}", addr);
        let channel = Endpoint::from_str(&uri)
            .unwrap()
            .timeout(TIMEOUT_LIMIT)
            .connect_timeout(CONNECT_TIMEOUT_LIMIT)
            .connect_lazy();

        {
            let mut guard = self.clients.write();
            guard.insert(addr, channel.clone());
        }

        channel
    }

    /// Removes a client from the network.
    pub fn disconnect(&self, addr: SocketAddr) {
        let mut guard = self.clients.write();
        guard.remove(&addr);
    }
}