cold-io 0.2.0

Event driven network io
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
119
120
121
122

// Copyright 2021 Vladislav Melnik
// SPDX-License-Identifier: MIT

use cold_io::{
    Proposal, ProposalKind, Proposer, Request, ConnectionSource, State, ReadOnce, WriteOnce,
    IoResult,
};

#[derive(Clone, Copy)]
enum ExampleState<const INITIATOR: bool> {
    Empty,
    Done,
}

impl<const INITIATOR: bool> ExampleState<INITIATOR> {
    fn can_terminate(&self) -> bool {
        matches!(self, &ExampleState::Done)
    }
}

impl<R, W, const INITIATOR: bool> State<R, W> for ExampleState<INITIATOR>
where
    R: ReadOnce,
    W: WriteOnce,
{
    type Ext = &'static str;

    type Rng = ();

    fn accept(&mut self, proposal: Proposal<R, W, Self::Ext, Self::Rng>) -> Request {
        use self::ExampleState::*;
        use std::str;

        const ADDRESS: ([u8; 4], u16) = ([127, 0, 0, 1], 8224);

        log::info!(
            "{}, {}",
            if INITIATOR { "initiator" } else { "responder" },
            proposal
        );

        match (*self, proposal.kind) {
            (Empty, ProposalKind::Wake) => {
                if INITIATOR {
                    Request::default().add_connect(ADDRESS)
                } else {
                    Request::default().set_source(ConnectionSource::Port(8224))
                }
            },
            (Empty, ProposalKind::Idle) => Request::default(),
            (Empty, ProposalKind::Connection { .. }) => Request::default(),
            (Empty, ProposalKind::OnReadable(id, once)) => {
                if !INITIATOR {
                    let mut buf = [0; 13];
                    if let IoResult::Done { length, .. } = once.read(&mut buf) {
                        let msg = str::from_utf8(&buf[..length]).unwrap();
                        log::info!("{} -> {:?}", id, msg);
                        *self = Done;
                        Request::default()
                    } else {
                        // we did not done the task, and connection is closed
                        // let's keep the same state
                        Request::default()
                    }
                } else {
                    // just to be explicit in real code, can omit this `drop`
                    // drop the `once` object, will close the reading half of the connection
                    drop(once);
                    Request::default()
                }
            },
            (Empty, ProposalKind::OnWritable(id, once)) => {
                if INITIATOR {
                    let msg = "hello, world!";
                    if let IoResult::Done { .. } = once.write(msg.as_bytes()) {
                        log::info!("{} <- {:?}", id, msg);
                        *self = Done;
                        Request::default()
                    } else {
                        // we did not done the task, and connection is closed
                        // let's try connect again
                        Request::default().add_connect(ADDRESS)
                    }
                } else {
                    drop(once);
                    Request::default()
                }
            },
            (Empty, ProposalKind::Custom(_)) => Request::default(),
            (Done, _) => Request::default(),
        }
    }
}

fn main() {
    use std::{thread, time::Duration, iter};

    env_logger::init();

    let r_thread = thread::spawn(move || {
        let mut rngs = iter::repeat(());
        let mut responder = ExampleState::<false>::Empty;
        let mut proposer = Proposer::new(1, 8).unwrap();
        while !responder.can_terminate() {
            proposer
                .run(&mut rngs, &mut responder, Duration::from_secs(1))
                .unwrap();
        }
    });
    thread::sleep(Duration::from_millis(100));

    let mut rngs = iter::repeat(());
    let mut initiator = ExampleState::<true>::Empty;
    let mut proposer = Proposer::new(0, 8).unwrap();
    while !initiator.can_terminate() {
        proposer
            .run(&mut rngs, &mut initiator, Duration::from_secs(1))
            .unwrap();
    }

    r_thread.join().unwrap();
}