titanrt 0.7.0

Typed reactive runtime for real-time systems
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
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147

use std::{collections::HashSet, marker::PhantomData, sync::Arc};

use ahash::AHashMap;
use anyhow::{Result, anyhow};
use parking_lot::RwLock;
use serde::{Deserialize, Serialize};
use tokio::runtime::Runtime;

#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct SpecificClient<Spec: Clone> {
    pub id: usize,
    pub ip: Option<String>,
    pub spec: Spec,
}

#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct ClientConfig<Spec: Clone> {
    pub default: Option<Spec>,
    pub specific: Vec<SpecificClient<Spec>>,
    #[serde(default)]
    pub fail_on_empty: bool,
}

pub trait ClientInitializer<Spec: Clone>
where
    Self: Sized,
{
    fn init(spec: &SpecificClient<Spec>, rt: Option<Arc<Runtime>>) -> Result<Self>;
}

#[derive(Clone)]
pub struct ClientsMap<Client: ClientInitializer<Spec>, Spec: Clone> {
    inner: Arc<RwLock<AHashMap<usize, Arc<Client>>>>,
    init_rt: Option<Arc<Runtime>>,
    _spec: PhantomData<Spec>,
}

impl<Client: ClientInitializer<Config>, Config: Clone> ClientsMap<Client, Config> {
    pub fn new(config: &ClientConfig<Config>, rt: Option<Arc<Runtime>>) -> Result<Self> {
        let map = Self {
            inner: Arc::new(RwLock::new(AHashMap::new())),
            init_rt: rt,
            _spec: PhantomData,
        };

        map.rebuild(config)?;

        Ok(map)
    }

    /// Rebuild the map from a full configuration snapshot.
    pub fn rebuild(&self, config: &ClientConfig<Config>) -> Result<()> {
        let mut id_seen: HashSet<usize> = HashSet::default();
        let mut prepared: Vec<(usize, Arc<Client>)> = Vec::with_capacity(config.specific.len() + 1);

        for spec in config.specific.iter() {
            if !id_seen.insert(spec.id) {
                tracing::warn!(
                    "[CientsPool] ID <{}> of IP <{:?}> addr is not unique",
                    spec.id,
                    spec.ip
                );
                continue;
            }

            let client = self.init_client(spec)?;
            prepared.push((spec.id, client));
        }

        if let Some(spec) = &config.default {
            let default_id = id_seen.len();
            let default_client = SpecificClient {
                id: default_id,
                ip: None,
                spec: spec.clone(),
            };
            let client = self.init_client(&default_client)?;
            prepared.push((default_id, client));
        }

        if config.fail_on_empty && prepared.is_empty() {
            return Err(anyhow!("Clients map is empty"));
        }

        let mut guard = self.inner.write();
        guard.clear();
        for (id, client) in prepared {
            guard.insert(id, client);
        }

        Ok(())
    }

    fn init_client(&self, spec: &SpecificClient<Config>) -> Result<Arc<Client>> {
        Client::init(spec, self.init_rt.clone()).map(Arc::new)
    }

    pub fn get(&self, id: &usize) -> Option<Arc<Client>> {
        self.inner.read().get(id).cloned()
    }

    pub fn len(&self) -> usize {
        self.inner.read().len()
    }

    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    pub fn contains(&self, id: &usize) -> bool {
        self.inner.read().contains_key(id)
    }

    pub fn ids(&self) -> Vec<usize> {
        self.inner.read().keys().copied().collect()
    }

    pub fn upsert(&self, client: SpecificClient<Config>) -> Result<Arc<Client>> {
        let id = client.id;
        let client = self.init_client(&client)?;
        let mut guard = self.inner.write();
        guard.insert(id, client.clone());
        Ok(client)
    }

    pub fn remove(&self, id: usize) -> Option<Arc<Client>> {
        self.inner.write().remove(&id)
    }

    pub fn next_vacant_id(&self) -> usize {
        let guard = self.inner.read();
        let mut candidate = guard.len();
        while guard.contains_key(&candidate) {
            candidate += 1;
        }
        candidate
    }

    pub fn sole_entry_id(&self) -> Option<usize> {
        let guard = self.inner.read();
        if guard.len() == 1 {
            guard.keys().next().copied()
        } else {
            None
        }
    }
}