infomgr 0.2.2

Show system information.
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

#[allow(unused)]
use tracing::{debug, error, info, warn};

use error_stack::{IntoReport, Result, ResultExt};
use serde::{de::DeserializeOwned, Serialize};
use {
    super::{InfoCon, InfomgrError},
    ipcon_sys::ipcon::{
        Ipcon, IPCON_KERNEL_GROUP_NAME, IPCON_KERNEL_NAME, IPF_DISABLE_KEVENT_FILTER, IPF_RCV_IF,
        IPF_SND_IF,
    },
    ipcon_sys::ipcon_msg::{IpconMsg, IpconMsgType},
    std::sync::mpsc::{Receiver, Sender},
    std::sync::Arc,
    std::thread::JoinHandle,
};

#[derive(Default)]
pub struct IpconInfomgr {
    handler: Option<JoinHandle<Result<(), InfomgrError>>>,
}

impl InfoCon for IpconInfomgr {
    /* source : group@peer */
    fn sender<T>(&mut self, source: Arc<String>, receiver: Receiver<T>) -> Result<(), InfomgrError>
    where
        T: 'static + Serialize + Send,
    {
        let handler = std::thread::spawn(move || -> Result<(), InfomgrError> {
            let (group, peer) = source
                .split_once('@')
                .ok_or(InfomgrError::InvalidData)
                .into_report()
                .attach_printable(format!("Invalid group {}", source))?;

            let ih = Ipcon::new(Some(peer), Some(IPF_SND_IF))
                .change_context(InfomgrError::IpconError)?;
            ih.register_group(group)
                .change_context(InfomgrError::IpconError)?;

            loop {
                let report = receiver
                    .recv()
                    .into_report()
                    .change_context(InfomgrError::IOError)?;

                let buf = serde_json::to_string(&report)
                    .into_report()
                    .change_context(InfomgrError::InvalidData)?;

                ih.send_multicast(group, buf.as_bytes(), false)
                    .change_context(InfomgrError::IpconError)?;
            }
        });

        self.handler = Some(handler);
        Ok(())
    }

    /* source : group@peer */
    fn receiver<T>(
        &mut self,
        source: Arc<String>,
        sender: Sender<Option<T>>,
    ) -> Result<(), InfomgrError>
    where
        T: 'static + DeserializeOwned + Send,
    {
        let handler = std::thread::spawn(move || -> Result<(), InfomgrError> {
            let (group, peer) = source
                .split_once('@')
                .ok_or(InfomgrError::InvalidData)
                .into_report()
                .attach_printable(format!("Invalid group {}", source))?;

            let matched = |p: &str, g: &str| -> bool { p == peer && g == group };

            let ih = Ipcon::new(None, Some(IPF_RCV_IF | IPF_DISABLE_KEVENT_FILTER))
                .change_context(InfomgrError::IpconError)?;

            ih.join_group(IPCON_KERNEL_NAME, IPCON_KERNEL_GROUP_NAME)
                .change_context(InfomgrError::IpconError)?;
            if ih.join_group(peer, group).is_ok() {
                debug!("Infomgr: {}@{} connected", group, peer);
            }

            loop {
                let msg = ih.receive_msg().change_context(InfomgrError::IOError)?;

                match msg {
                    IpconMsg::IpconMsgKevent(kevent) => {
                        if let Some((p, g)) = kevent.group_added() {
                            if matched(&p, &g) && ih.join_group(&p, &g).is_ok() {
                                debug!("Infomgr: {}@{} connected", g, p);
                            }

                            continue;
                        }

                        if let Some((p, g)) = kevent.group_removed() {
                            if matched(&p, &g) {
                                debug!("Infomgr {}@{} lost.", g, p);
                                sender
                                    .send(None)
                                    .map_err(|_| InfomgrError::IOError)
                                    .into_report()?;
                                continue;
                            }
                        }
                    }

                    IpconMsg::IpconMsgUser(m)
                        if matches!(m.msg_type, IpconMsgType::IpconMsgTypeGroup) =>
                    {
                        let group = m.group.unwrap();
                        if matched(&m.peer, &group) {
                            if let Ok(s) = std::str::from_utf8(&m.buf) {
                                if let Ok(ir) = serde_json::from_str(s) {
                                    debug!("Infomgr report received from {}@{}.", group, m.peer);
                                    sender
                                        .send(Some(ir))
                                        .map_err(|_| InfomgrError::IOError)
                                        .into_report()?;
                                }
                            }
                        }
                    }
                    _ => {}
                }
            }
        });
        self.handler = Some(handler);
        Ok(())
    }
}