asansio 0.2.0

The Async/await state machine for the Sans I/O design pattern
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
123
124
125
126

mod tlv_pingpong_proto;

use clap::Parser;
use std::io::Read;
use std::io::Write;
use std::net::SocketAddr;
use std::net::TcpStream;
use std::pin::pin;
use std::time::Duration;
use tlv_pingpong_proto::ClientRequest;
use tlv_pingpong_proto::ClientResponse;

#[derive(Parser)]
#[clap(version)]
struct Args {
    #[clap(short, long, default_value = "127.0.0.1:7123")]
    connect: SocketAddr,
}

fn main() {
    client(Args::parse().connect);
}

fn client(connect: SocketAddr) {
    let tcp = TcpStream::connect(connect).unwrap();
    println!("Connected to {}", connect);
    client_process(tcp);
}

struct Cache {
    buf: Vec<u8>,
    msg: String,
    count_read: usize,
    count_send: usize,
    count_recv: usize,
}

impl Cache {
    fn new() -> Self {
        let mut buf = Vec::new();
        buf.reserve(1024);
        Self {
            buf,
            msg: String::new(),
            count_read: 0,
            count_send: 0,
            count_recv: 0,
        }
    }
}

fn client_process_read_payload<'a>(
    cache: &'a mut Cache,
    tcp: &mut TcpStream,
) -> Option<ClientResponse<'a>> {
    cache.buf.resize(1024, 0);
    if let Ok(len) = tcp.read(&mut cache.buf) {
        if len == 0 {
            return None;
        }
        Some(ClientResponse::ReadPayload {
            payload: &cache.buf[0..len],
        })
    } else {
        cache.count_read += 1;
        if cache.count_read % 3 == 0 {
            return Some(ClientResponse::Sleep {
                duration: Duration::from_millis(200),
            });
        }
        if cache.count_send < 100 {
            cache.count_send += 1;
            cache.msg = format!("packet {}", cache.count_send);
            Some(ClientResponse::Message {
                msg: cache.msg.as_ref(),
            })
        } else {
            Some(ClientResponse::ReadPayload { payload: &[] })
        }
    }
}

fn client_process_write_payload<'a>(
    tcp: &mut TcpStream,
    payload: &[u8],
) -> Option<ClientResponse<'a>> {
    tcp.write_all(payload).unwrap();
    Some(ClientResponse::ReadPayload { payload: &[] })
}

fn client_process_message<'a>(cache: &mut Cache, msg: &str) -> Option<ClientResponse<'a>> {
    dbg!(msg);
    cache.count_recv += 1;
    if cache.count_recv == 100 {
        None
    } else {
        Some(ClientResponse::ReadPayload { payload: &[] })
    }
}

fn client_process(mut tcp: TcpStream) {
    let mut cache = Cache::new();

    tcp.set_read_timeout(Some(Duration::from_millis(10)))
        .unwrap();

    let (sans, io) = asansio::new();
    let task = pin!(tlv_pingpong_proto::run_client(sans));

    let mut handle = io.start(task);
    while handle.is_some() {
        let response = match handle.as_ref().unwrap().message() {
            Some(ClientRequest::ReadPayload) => client_process_read_payload(&mut cache, &mut tcp),
            Some(ClientRequest::WritePayload { payload }) => {
                client_process_write_payload(&mut tcp, payload)
            }
            Some(ClientRequest::Message { msg }) => client_process_message(&mut cache, msg),
            Some(ClientRequest::Error) => break,
            None => break,
        };
        let Some(response) = response else {
            break;
        };
        handle = io.handle(handle.take().unwrap(), &response);
    }
}