tubes 0.6.2

Host/Client protocol based on pipenet
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

use crate::client_id::ClientId;
use crate::{core::to_pipenet, data::MessageDataInternal, prelude::*};

use pipenet::NonBlockStream;
use std::{
    collections::HashMap,
    net::{TcpListener, TcpStream},
    sync::{Arc, Mutex},
    time::{Duration, Instant},
};

type Result<T> = std::result::Result<T, Box<dyn std::error::Error>>;

pub(crate) fn accept_loop<const MS: usize>(
    server_uuid: ClientId,
    config: Config,
    clients: Arc<Mutex<HashMap<ClientId, NonBlockStream::<MS>>>>,
    listener: TcpListener,
    accept_timeout: Duration,
) {
    while let Ok((stream, _)) = listener.accept() {
        let Ok(mut map) = clients.lock() else {
            return;
        };
        accept_one::<MS>(server_uuid, &config, &mut map, accept_timeout, stream);
    }
}

fn accept_one<const MS: usize>(
    server_uuid: ClientId,
    config: &Config,
    map: &mut HashMap<ClientId, NonBlockStream::<MS>>,
    accept_timeout: Duration,
    stream: TcpStream,
) {
    if stream.set_nonblocking(true).is_err() {
        // client will be dropped here: cannot set nonblocking
        eprintln!("[{}] accept: Cannot set client to nonblocking", server_uuid);
        return;
    }
    let mut client = to_pipenet::<MS>(stream, config);
    let Ok(Some(accepted_uuid)) = wait_for_client_joined_message(accept_timeout, &mut client)
    else {
        // client will be dropped here: timed out waiting for join
        eprintln!(
            "[{}] accept: Timed out while accepting unknown client",
            server_uuid
        );
        return;
    };
    // Notify the client of the uuid of the server.
    let Ok(server_uuid_msg) = (MessageDataInternal::ServerUuid(server_uuid)).try_into() else {
        eprintln!("[{}] accept: Cannot build message", server_uuid);
        return;
    };
    let Ok(accepted_client_joined_msg): Result<Vec<u8>> =
        (MessageDataInternal::ClientJoined(accepted_uuid)).try_into()
    else {
        eprintln!("[{}] accept: Cannot build message", server_uuid);
        return;
    };
    let Ok(_) = client.write(server_uuid_msg) else {
        // If can't send uuid to client, drop and continue.
        eprintln!(
            "[{}] accept: Cannot send message to client {}",
            server_uuid, accepted_uuid
        );
        return;
    };
    for (map_uuid, map_client) in map.iter_mut() {
        // Tell to each existing client that this new client joined
        let Ok(_) = map_client.write(accepted_client_joined_msg.clone()) else {
            // If can't send uuids to client, drop and continue.
            eprintln!(
                "[{}] accept: Cannot send message to client {}",
                server_uuid, map_uuid
            );
            return;
        };
        // Also tell the new joiner about all the previously existing clients.
        let Ok(map_client_joined_msg): Result<Vec<u8>> =
            (MessageDataInternal::ClientJoined(*map_uuid)).try_into()
        else {
            eprintln!("[{}] accept: Cannot build message", server_uuid);
            return;
        };
        let Ok(_) = client.write(map_client_joined_msg.clone()) else {
            // If can't send uuids to client, drop and continue.
            eprintln!(
                "[{}] accept: Cannot send message to client {}",
                server_uuid, accepted_uuid
            );
            return;
        };
    }

    // Only now insert it into the map
    map.insert(accepted_uuid, client);
}

fn wait_for_client_joined_message<const MS: usize>(
    timeout: Duration,
    client: &mut NonBlockStream<MS>,
) -> Result<Option<ClientId>> {
    let now = Instant::now();
    loop {
        let Some(msg) = client.read()? else {
            continue;
        };
        let msg = MessageDataInternal::try_from(msg.as_slice())?;
        if let MessageDataInternal::ClientJoined(uuid) = msg {
            return Ok(Some(uuid));
        }
        if now.elapsed() > timeout {
            return Ok(None);
        }
    }
}