flnode 0.10.0

Glue code for flmodules
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

use std::collections::HashMap;

use flarch::{
    broker::{Broker, SubsystemHandler},
    nodeids::{NodeID, U256},
    platform_async_trait,
};
use flmodules::{
    network::broker::{BrokerNetwork, NetworkConnectionState, NetworkIn, NetworkOut},
    nodeconfig::NodeInfo,
};
use tokio::sync::watch;

#[derive(Default, Debug, Clone)]
pub struct NetStats {
    pub states: HashMap<U256, NetworkConnectionState>,
    pub connected: Vec<NodeID>,
    pub online: Vec<NodeID>,
    pub node_infos: HashMap<NodeID, NodeInfo>,
}

/// Collects the statistics of the connections sent by the network broker.
pub struct NetworkStats {
    stats: NetStats,
    tx: Option<watch::Sender<NetStats>>,
}

impl NetworkStats {
    pub async fn start(mut broker_net: BrokerNetwork) -> anyhow::Result<watch::Receiver<NetStats>> {
        let stats = NetStats::default();
        let (tx, rx) = watch::channel(stats.clone());
        let mut broker = Broker::new();
        broker
            .add_handler(Box::new(Self {
                stats,
                tx: Some(tx),
            }))
            .await?;
        broker_net
            .add_translator_o_ti(broker, Box::new(|msg| Some(msg)))
            .await?;
        Ok(rx)
    }
}

#[platform_async_trait()]
impl SubsystemHandler<NetworkOut, NetworkIn> for NetworkStats {
    async fn messages(&mut self, msgs: Vec<NetworkOut>) -> Vec<NetworkIn> {
        for msg in msgs {
            match msg {
                NetworkOut::NodeListFromWS(node_infos) => {
                    self.stats.online = node_infos.iter().map(|ni| ni.get_id()).collect();
                    for info in node_infos {
                        self.stats.node_infos.insert(info.get_id(), info);
                    }
                }
                NetworkOut::ConnectionState(state) => {
                    self.stats.states.insert(state.id, state);
                }
                NetworkOut::Connected(id) => {
                    self.stats.connected.push(id);
                }
                NetworkOut::Disconnected(id) => {
                    self.stats.connected.retain(|i| i != &id);
                }
                _ => {}
            }
        }
        self.tx
            .clone()
            .map(|tx| tx.send(self.stats.clone()).is_err().then(|| self.tx = None));
        vec![]
    }
}

impl NetStats {
    pub fn node_infos_connected(&self) -> Vec<NodeInfo> {
        self.connected
            .iter()
            .filter_map(|id| self.node_infos.get(id))
            .cloned()
            .collect()
    }
}