apeirogond 0.21.0

Daemon for Apeirogon, a cross-platform software KVM
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

// SPDX-License-Identifier: GPL-3.0-or-later

use std::net::Ipv4Addr;
use std::time::Duration;

use tempfile::TempDir;
use tokio::sync::oneshot;
use tokio::task::JoinHandle;
use tokio::time::timeout;

use super::*;
use crate::command::Status;

pub(super) const TEST_TIMEOUT: Duration = Duration::from_secs(15);

pub(crate) struct Peer {
    pub(crate) sender: Option<mpsc::Sender<Command>>,
    pub(super) discoveries: mpsc::Sender<crate::discovery::Event>,
    task: JoinHandle<Result<(), DaemonError>>,
    pub(crate) machine: MachineId,
    pub(crate) addr: SocketAddr,
    // Keep persistence alive until the daemon task is stopped.
    pub(crate) dir: TempDir,
}

impl Peer {
    pub(crate) fn start() -> Self {
        let dir = tempfile::tempdir().unwrap();
        let daemon = Daemon::load(dir.path(), SocketAddr::from((Ipv4Addr::LOCALHOST, 0))).unwrap();
        Self::run(daemon, dir)
    }

    pub(crate) fn run(daemon: Daemon, dir: TempDir) -> Self {
        let machine = daemon.identity.machine_id;
        let addr = daemon.endpoint.local_addr().unwrap();
        let (sender, receiver) = mpsc::channel(MAILBOX_CAPACITY);
        let (discoveries, discovery_rx) = mpsc::channel(MAILBOX_CAPACITY);
        let task = tokio::spawn(daemon.run(receiver, discovery_rx));
        Self {
            sender: Some(sender),
            discoveries,
            task,
            machine,
            addr,
            dir,
        }
    }

    pub(crate) async fn command<T>(
        &self,
        command: impl FnOnce(Reply<T>) -> Command,
    ) -> Result<T, DaemonError> {
        let (reply, receiver) = oneshot::channel();
        self.sender
            .as_ref()
            .unwrap()
            .send(command(reply))
            .await
            .unwrap();
        receiver.await.unwrap()
    }

    pub(crate) async fn status(&self) -> Status {
        self.command(Command::Status).await.unwrap()
    }

    pub(crate) async fn wait(&self, predicate: impl Fn(&Status) -> bool) -> Status {
        timeout(TEST_TIMEOUT, async {
            loop {
                let status = self.status().await;
                if predicate(&status) {
                    return status;
                }
                tokio::task::yield_now().await;
            }
        })
        .await
        .expect("daemon did not reach expected state")
    }

    pub(crate) async fn pair(&self, requester: &Self) -> Entry {
        let code = self.command(Command::PairHost).await.unwrap();
        let entry = timeout(
            TEST_TIMEOUT,
            requester.command(|reply| Command::PairJoin {
                addr: self.addr,
                code,
                reply,
            }),
        )
        .await
        .unwrap()
        .unwrap();
        self.wait(|status| linked(status, requester.machine)).await;
        requester.wait(|status| linked(status, self.machine)).await;
        entry
    }

    pub(crate) async fn stop(mut self) {
        self.sender.take();
        timeout(TEST_TIMEOUT, &mut self.task)
            .await
            .unwrap()
            .unwrap()
            .unwrap();
    }
}

impl Drop for Peer {
    fn drop(&mut self) {
        self.task.abort();
    }
}

pub(super) fn linked(status: &Status, machine: MachineId) -> bool {
    status
        .peers
        .iter()
        .any(|peer| peer.id == machine && peer.connected)
}

pub(super) fn paired(status: &Status, machine: MachineId) -> bool {
    status.peers.iter().any(|peer| peer.id == machine)
}