sharedstate 0.5.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

use message_encoding::MessageEncoding;
use std::{fmt::Debug, future::Future, hash::Hash};
use tokio::{
    io::{AsyncRead, AsyncWrite},
    sync::mpsc::{channel, Receiver, Sender},
};

use crate::{
    net::message_channel::{NetIoSettings, ReadChannel, WriteChannel},
    shared::messages::{SyncRequest, SyncResponse},
    state::determinstic_state::DeterministicState,
};

pub trait SyncIOAddress:
    Debug + Clone + Copy + Send + Sync + PartialEq + Eq + Ord + Hash + MessageEncoding + 'static
{
}

impl<T: Debug + Clone + Copy + Send + Sync + PartialEq + Eq + Ord + Hash + MessageEncoding + 'static> SyncIOAddress
    for T
{
}

pub trait SyncIO: Sized + Send + Sync + 'static {
    type Address: SyncIOAddress;
    type Read: AsyncRead + Send + Unpin + 'static;
    type Write: AsyncWrite + Send + Unpin + 'static;

    fn connect(&self, remote: &Self::Address) -> impl Future<Output = std::io::Result<SyncConnection<Self>>> + Send;
}

pub trait SyncIOListener: SyncIO {
    fn next_client(&self) -> impl Future<Output = std::io::Result<SyncConnection<Self>>> + Send;
}

pub struct SyncConnection<I: SyncIO> {
    pub remote: I::Address,
    pub read: I::Read,
    pub write: I::Write,
}

impl<I: SyncIO> SyncConnection<I> {
    pub fn client_channels<D: DeterministicState>(
        self,
        settings: NetIoSettings,
    ) -> (I::Address, Sender<SyncRequest<I::Address, D>>, Receiver<SyncResponse<I::Address, D>>)
    where
        D: MessageEncoding,
        D::Action: MessageEncoding,
        D::AuthorityAction: MessageEncoding,
    {
        self.channels(settings)
    }

    pub fn server_channels<D: DeterministicState>(
        self,
        settings: NetIoSettings,
    ) -> (I::Address, Sender<SyncResponse<I::Address, D>>, Receiver<SyncRequest<I::Address, D>>)
    where
        D: MessageEncoding,
        D::Action: MessageEncoding,
        D::AuthorityAction: MessageEncoding,
    {
        self.channels(settings)
    }

    fn channels<W: MessageEncoding + Send + Sync + 'static, R: MessageEncoding + Send + Sync + 'static>(
        self,
        settings: NetIoSettings,
    ) -> (I::Address, Sender<W>, Receiver<R>) {
        let (write_tx, write_rx) = channel(512);
        let (read_tx, read_rx) = channel(512);

        tokio::spawn(
            WriteChannel::<I, W> {
                input: write_rx,
                output: self.write,
                settings: settings.clone(),
            }
            .start(),
        );

        tokio::spawn(
            ReadChannel::<I, R> {
                output: read_tx,
                input: self.read,
                settings,
            }
            .start(),
        );

        (self.remote, write_tx, read_rx)
    }
}