tropocol 0.1.2

Send and receive serde-compatible objects over TCP (async)
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

use serde_json::{to_vec, from_slice};
use std::net::SocketAddr;

use super::*;

/// Handles encoding/decoding and transmission/reception of TCP messages
pub async fn session<I, O, R, T>(
    stream: TcpStream,
    handle_incoming: R,
    await_outgoing: T,
) -> Result<(), Error>
where
    I: Incoming,
    O: Outgoing,
    R: HandleIncoming<I>,
    T: GetOutgoing<O>,
{
    let peer_addr = stream.peer_addr().ok();
    let tx_tcp = stream.clone();
    let rx_tcp = stream;

    let tx_task = transmit(tx_tcp, await_outgoing, peer_addr);
    let rx_task = receive(rx_tcp, handle_incoming, peer_addr);

    let result = or(tx_task, rx_task).await;

    #[cfg(feature = "log")]
    if let Err(error) = &result {
        match peer_addr {
            Some(addr) => log::error!("{}: {}", addr, error),
            None => log::error!("<noaddr>: {}", error),
        }
    } else {
        match peer_addr {
            Some(addr) => log::debug!("{}: Connection Closed", addr),
            None => log::debug!("<noaddr>: Connection Closed"),
        }
    }

    result
}

#[allow(unused_variables)]
async fn receive<I, R>(
    mut rx_tcp: TcpStream,
    mut handler: R,
    peer_addr: Option<SocketAddr>,
) -> Result<(), Error>
where
    I: Incoming,
    R: HandleIncoming<I>,
{
    loop {
        let mut len = [0u8; 8];

        if rx_tcp.read_exact(&mut len).await.is_err() {
            break Ok((/* Connection Closed (Clean) */));
        }

        let len = u64::from_be_bytes(len) as usize;
        let mut buffer = vec![0u8; len];

        if rx_tcp.read_exact(&mut buffer).await.is_err() {
            break Err(Error::UnexpectedEof);
        }

        let incoming = from_slice(&buffer).map_err(Error::Codec)?;

        #[cfg(feature = "log")]
        match peer_addr {
            Some(addr) => log::debug!("{}: Received {:?}", addr, incoming),
            None => log::debug!("<noaddr>: Received {:?}", incoming),
        }

        handler.handle_incoming(incoming).await;
    }
}

#[allow(unused_variables)]
async fn transmit<O, T>(
    mut tx_tcp: TcpStream,
    mut source: T,
    peer_addr: Option<SocketAddr>,
) -> Result<(), Error>
where
    O: Outgoing,
    T: GetOutgoing<O>,
{
    while let Some(outgoing) = source.get_outgoing().await {
        #[cfg(feature = "log")]
        match peer_addr {
            Some(addr) => log::debug!("{}: Sending {:?}", addr, outgoing),
            None => log::debug!("<noaddr>: Sending {:?}", outgoing),
        }

        let bytes = to_vec(&outgoing).map_err(Error::Codec)?;
        let len = (bytes.len() as u64).to_be_bytes();

        let mut tx_fail = false;
        tx_fail |= tx_tcp.write_all(&len).await.is_err();
        tx_fail |= tx_tcp.write_all(&bytes).await.is_err();

        if tx_fail {
            return Err(Error::UnexpectedEof);
        }
    }

    Ok(())
}