sharedstate 2.0.0

Sync heavily read state across many servers
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

use std::{io, net::SocketAddr, sync::Arc};

use tokio::net::{
    tcp::{OwnedReadHalf, OwnedWriteHalf},
    TcpListener, TcpStream,
};

use crate::transport::traits::{SyncConnection, SyncIO, SyncIOListener};

#[derive(Clone)]
pub struct TcpIo {
    address: SocketAddr,
    listener: Arc<TcpListener>,
}

impl TcpIo {
    pub async fn bind(address: SocketAddr) -> io::Result<Self> {
        Self::from_listener(TcpListener::bind(address).await?)
    }

    pub fn from_listener(listener: TcpListener) -> io::Result<Self> {
        let address = listener.local_addr()?;
        Ok(Self {
            address,
            listener: Arc::new(listener),
        })
    }

    pub fn address(&self) -> SocketAddr {
        self.address
    }
}

impl SyncIO for TcpIo {
    type Address = SocketAddr;
    type Read = OwnedReadHalf;
    type Write = OwnedWriteHalf;

    async fn connect(&self, remote: &Self::Address) -> io::Result<SyncConnection<Self>> {
        let stream = TcpStream::connect(remote).await?;
        let (read, write) = stream.into_split();
        Ok(SyncConnection {
            remote: *remote,
            read,
            write,
        })
    }
}

impl SyncIOListener for TcpIo {
    async fn next_client(&self) -> io::Result<SyncConnection<Self>> {
        let (stream, remote) = self.listener.accept().await?;
        let (read, write) = stream.into_split();
        Ok(SyncConnection { remote, read, write })
    }
}

#[cfg(test)]
mod tests {
    use std::net::{Ipv4Addr, SocketAddr};

    use tokio::io::{AsyncReadExt, AsyncWriteExt};

    use super::*;

    fn localhost_ephemeral() -> SocketAddr {
        SocketAddr::from((Ipv4Addr::LOCALHOST, 0))
    }

    #[tokio::test]
    async fn bind_updates_ephemeral_port() {
        let io = TcpIo::bind(localhost_ephemeral()).await.unwrap();

        assert_ne!(io.address().port(), 0);
        assert_eq!(io.address(), io.listener.local_addr().unwrap());
    }

    #[tokio::test]
    async fn connect_and_accept_return_socket_addresses() {
        let server = TcpIo::bind(localhost_ephemeral()).await.unwrap();
        let client = TcpIo::bind(localhost_ephemeral()).await.unwrap();

        let accept = tokio::spawn({
            let server = server.clone();
            async move { server.next_client().await }
        });

        let mut client_conn = client.connect(&server.address()).await.unwrap();
        let mut server_conn = accept.await.unwrap().unwrap();

        assert_eq!(client_conn.remote, server.address());
        assert_eq!(server_conn.remote, client_conn.write.local_addr().unwrap());

        client_conn.write.write_all(b"ping").await.unwrap();

        let mut received = [0; 4];
        server_conn.read.read_exact(&mut received).await.unwrap();
        assert_eq!(&received, b"ping");
    }
}