msrt 0.1.7

Portable MSRT protocol implementation.
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

//! Passive single-peer endpoint manager.

use crate::core::Result;
use crate::endpoint::{EndpointPoll, EngineConfig, MessageId, PeerSlot, ReceiveReport};

/// Passive single-peer endpoint.
///
/// This endpoint owns at most one protocol session. It does not actively send hello on
/// startup; instead, it creates the session lazily when bytes arrive from the
/// peer. Disconnect drops the session so the next peer connection starts with a
/// fresh protocol session.
#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct PassiveEndpoint {
    peer: PeerSlot,
}

impl PassiveEndpoint {
    /// Creates a passive endpoint using `config` for each new peer session.
    #[must_use]
    pub const fn new(config: EngineConfig) -> Self {
        Self {
            peer: PeerSlot::new(config),
        }
    }

    /// Returns the single peer slot.
    #[must_use]
    pub const fn peer(&self) -> &PeerSlot {
        &self.peer
    }

    /// Returns the mutable single peer slot.
    pub fn peer_mut(&mut self) -> &mut PeerSlot {
        &mut self.peer
    }

    /// Drops the active host session if one exists.
    pub fn disconnect(&mut self) {
        self.peer.disconnect();
    }

    /// Queues an application message if a host session exists.
    pub fn send(&mut self, message: &[u8]) -> Result<Option<MessageId>> {
        self.peer.send(message)
    }

    /// Creates a passive session if needed, then feeds received bytes into it.
    pub fn receive(&mut self, now_ms: u64, bytes: &[u8]) -> ReceiveReport {
        if let Err(error) = self.peer.engine_or_accept_passive(now_ms) {
            return ReceiveReport::Error(error);
        }

        self.peer.receive(now_ms, bytes)
    }

    /// Polls one endpoint action from the active host engine.
    pub fn poll<'a>(&mut self, now_ms: u64, tx_buf: &'a mut [u8]) -> Result<EndpointPoll<'a>> {
        self.peer.poll(now_ms, tx_buf)
    }

    /// Drops the active host session if it has been idle for at least `timeout_ms`.
    pub fn disconnect_if_idle(&mut self, now_ms: u64, timeout_ms: u64) -> bool {
        self.peer.disconnect_if_idle(now_ms, timeout_ms)
    }
}

#[cfg(test)]
mod tests {
    use crate::endpoint::{ClientEndpoint, EndpointPoll, PeerState};

    use super::PassiveEndpoint;

    #[test]
    fn passive_endpoint_starts_without_engine_and_accepts_first_peer_bytes() {
        let mut host = ClientEndpoint::default();
        let mut passive = PassiveEndpoint::default();
        let mut host_tx = [0; 128];
        let mut passive_tx = [0; 128];

        assert!(!passive.peer().has_session());

        host.connect(1).unwrap();
        let EndpointPoll::Transmit {
            bytes: hello_bytes, ..
        } = host.poll(1, &mut host_tx).unwrap()
        else {
            panic!("host should transmit hello");
        };

        passive.receive(2, hello_bytes);
        assert!(passive.peer().has_session());
        assert_eq!(passive.peer().state(), PeerState::Connected);

        let EndpointPoll::Transmit {
            bytes: ack_bytes, ..
        } = passive.poll(2, &mut passive_tx).unwrap()
        else {
            panic!("passive endpoint should transmit ack");
        };

        host.receive(3, ack_bytes);
        assert_eq!(host.peer().state(), PeerState::Connected);
    }

    #[test]
    fn passive_disconnect_drops_engine_until_next_receive() {
        let mut passive = PassiveEndpoint::default();

        passive.receive(1, &[]);
        assert!(passive.peer().has_session());

        passive.disconnect();
        assert!(!passive.peer().has_session());
        assert_eq!(passive.peer().state(), PeerState::Disconnected);
    }
}