embassy-socket 0.1.2

tcp client/server callback
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
127
128
129
130
131
132
133
134
135
136
137
138
139

use embassy_net::{IpEndpoint, Stack};
use embassy_net::tcp::State;
use embassy_time::{Duration, Timer};
use crate::channel::WriteChannel;
use crate::connection::socket_state::SocketState;
use crate::connection::TcpConnection;
use crate::err::{SocketErr, SocketResult};
use crate::tcp_server::callback::TcpServerCallBack;

/// single tcp server runner
pub struct TcpServerRunner<'d, const N: usize, const TX_SZ: usize, const RX_SZ: usize, const BUF_SIZE: usize, CB: TcpServerCallBack> {
    /// tcp stack
    stack: Stack<'d>,
    /// socket state, memory pool
    state: &'d SocketState<N, TX_SZ, RX_SZ, BUF_SIZE>,
    /// listener port
    port: u16,
    /// socket timeout
    socket_timeout: Option<Duration>,
    /// read data timeout, default is 100 milliseconds
    read_timeout: Duration,
    /// tcp callback
    pub cb: &'d CB,
}

/// runner support Clone
impl<'d, const N: usize, const TX_SZ: usize, const RX_SZ: usize, const BUF_SIZE: usize, CB: TcpServerCallBack> Clone for
TcpServerRunner<'d, N, TX_SZ, RX_SZ, BUF_SIZE, CB> {
    #[inline]
    fn clone(&self) -> Self {
        Self {
            stack: self.stack,
            state: self.state,
            port: self.port,
            socket_timeout: self.socket_timeout,
            read_timeout: self.read_timeout,
            cb: self.cb,
        }
    }
}

/// runner support Copy
impl<'d, const N: usize, const TX_SZ: usize, const RX_SZ: usize, const BUF_SIZE: usize, CB: TcpServerCallBack> Copy for
TcpServerRunner<'d, N, TX_SZ, RX_SZ, BUF_SIZE, CB> {}

/// custom method
impl<'d, const N: usize, const TX_SZ: usize, const RX_SZ: usize, const BUF_SIZE: usize, CB: TcpServerCallBack>
TcpServerRunner<'d, N, TX_SZ, RX_SZ, BUF_SIZE, CB> {
    /// create one runner
    #[inline]
    pub fn new(stack: Stack<'d>, state: &'d SocketState<N, TX_SZ, RX_SZ, BUF_SIZE>, port: u16, cb: &'d CB) -> Self {
        Self { stack, state, port, socket_timeout: None, read_timeout: Duration::from_millis(100), cb }
    }

    /// set socket timeout<br />
    /// connection timeout and etc. <br />
    /// tt is recommended not to set or set to None
    #[inline]
    pub fn socket_timeout(&mut self, timeout: Option<Duration>) {
        self.socket_timeout = timeout;
    }

    /// set read timeout
    #[inline]
    pub fn read_timeout(&mut self, timeout: Duration) {
        self.read_timeout = timeout;
    }

    /// run tcp server
    pub async fn run<const CN: usize>(&self, wch: &WriteChannel<'_, CN>, t: &mut CB::T) {
        loop {
            if let Err(e) = self.run_logic(wch, t).await {
                self.cb.err(e, t).await;
            }
        }
    }

    /// run logic
    async fn run_logic<const CN: usize>(&self, wch: &WriteChannel<'_, CN>, t: &mut CB::T) -> SocketResult<()> {
        let mut conn = self.try_accept().await?;
        let endpoint = conn.socket.remote_endpoint().ok_or_else(SocketErr::no_route)?;

        wch.enable().await;
        self.cb.conn(endpoint, wch, t).await;
        while !self.read_logic(&mut conn, endpoint, wch, t).await {}
        wch.disable().await;
        self.cb.dis_conn(endpoint, t).await;
        Ok(())
    }

    /// read logic
    async fn read_logic<const CN: usize>(
        &self,
        conn: &mut TcpConnection<'d, N, TX_SZ, RX_SZ, BUF_SIZE>,
        endpoint: IpEndpoint,
        wch: &WriteChannel<'_, CN>,
        t: &mut CB::T) -> bool {
        if !conn.socket.can_recv() {
            if let Err(e) = self.write_logic(conn, wch).await { self.cb.err(e, t).await; }
            return matches!(conn.socket.state(), State::CloseWait|State::Closed);
        }

        match conn.try_read().await {
            Ok(bytes) => {
                self.cb.recv(endpoint, bytes, wch, t).await;
                false
            }
            Err(e) => {
                self.cb.err(e.into(), t).await;
                true
            }
        }
    }

    /// write logic
    async fn write_logic<const CN: usize>(
        &self,
        conn: &mut TcpConnection<'d, N, TX_SZ, RX_SZ, BUF_SIZE>,
        wch: &WriteChannel<'_, CN>) -> SocketResult<()> {
        if wch.is_empty().await {
            Timer::after(self.read_timeout).await;
            return Ok(());
        }

        wch.try_tcp_write(conn).await?;
        Ok(())
    }

    /// try accept connection
    async fn try_accept(&self) -> SocketResult<TcpConnection<'d, N, TX_SZ, RX_SZ, BUF_SIZE>> {
        self.stack.wait_link_up().await;
        self.stack.wait_config_up().await;

        let mut conn = TcpConnection::new(self.stack, self.state)?;
        conn.socket.set_timeout(self.socket_timeout);
        conn.socket.accept(self.port).await?;
        Ok(conn)
    }
}