cig 0.1.2

Simplify TCP/IP applications with a transparential, persistent-mode and data-driven protocol
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

use std::net::ToSocketAddrs;
use std::{io, time::SystemTime};

use serde::{de::DeserializeOwned, Serialize};

use crate::{
    pkg::{Pkg, PkgData},
    stog,
};
use crate::{PeerID, IO_RESTORE_CONNECT_SYNC_SECS};

use self::{client::Client, host::Host};

pub mod client;
pub mod host;

pub enum Conn<T: DeserializeOwned + Serialize> {
    Host(Host<T>),
    Client(Client<T>),
}

impl<T: DeserializeOwned + Serialize> Conn<T> {
    pub fn connect<A>(addr: A) -> io::Result<Self>
    where
        A: ToSocketAddrs,
    {
        Ok(Self::Client(Client::connect(addr)?))
    }

    pub fn is_host(&self) -> bool {
        matches!(self, Self::Host(_))
    }

    pub fn host<A>(addr: A) -> io::Result<Self>
    where
        A: ToSocketAddrs,
    {
        Ok(Self::Host(Host::bind(addr)?))
    }

    pub fn poll(&mut self) -> stog::Result<Option<Pkg<T>>> {
        let r = match self {
            Self::Host(ref mut h) => h.poll(),
            Self::Client(ref mut c) => c.poll(),
        };
        if matches!(r, Err(stog::Error::ConnClosed)) {
            self.restore_connection()?;
            Ok(None)
        } else {
            r
        }
    }

    pub fn get_id(&self) -> PeerID {
        match self {
            Self::Host(h) => h.id,
            Self::Client(c) => c.id,
        }
    }

    pub fn send_to(&mut self, id: PeerID, data: T) -> stog::Result<Pkg<T>> {
        let pkg = Pkg::new(self.get_id(), PkgData::Data(data, id));
        match self {
            Self::Host(ref mut h) => h.send_to(id, &pkg)?,
            Self::Client(ref mut c) => stog::send(&pkg, &mut c.stream)?,
        };
        Ok(pkg)
    }

    pub fn send_to_all(&mut self, data: T) -> stog::Result<Pkg<T>> {
        let pkg = Pkg::new(self.get_id(), PkgData::GlobalData(data));
        match self {
            Self::Host(ref mut h) => h.send_to_all(&pkg)?,
            Self::Client(ref mut c) => stog::send(&pkg, &mut c.stream)?,
        };
        Ok(pkg)
    }

    pub fn restore_connection(&mut self) -> stog::Result<()> {
        // works by giving a interval for each connected peer to the old server to stabilish a new
        // connection. If that one can't host, the next one tries.
        let client = match self {
            Conn::Client(c) => c,
            Conn::Host(_) => unreachable!(),
        };
        let connected_peers: Vec<_> = client.get_active_peers().collect();
        loop {
            let now = (SystemTime::now().duration_since(std::time::UNIX_EPOCH)).unwrap();
            let i = (now - client.last_host_timestamp).as_secs()
                / IO_RESTORE_CONNECT_SYNC_SECS as u64
                % connected_peers.len() as u64;
            let target = client.peers[connected_peers[i as usize] as usize]
                .as_ref()
                .unwrap();
            let new_conn = if connected_peers[i as usize] == client.id {
                Conn::host(target)
            } else {
                Conn::connect(target)
            };
            match new_conn {
                Ok(s) => {
                    *self = s;
                    break;
                }
                Err(ref e) if e.kind() == std::io::ErrorKind::ConnectionRefused => (),
                Err(e) => panic!("error restoring: {e:?}"),
            }
        }
        Ok(())
    }
}