twamp-rs 0.2.0

Implementation of TWAMP (RFC 5357)
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
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222

use crate::twamp_control::Accept;
use crate::twamp_control::AcceptSession;
use crate::twamp_control::RequestTwSession;
use crate::twamp_control::SecurityMode;
use crate::twamp_control::ServerGreeting;
use crate::twamp_control::ServerStart;
use crate::twamp_control::SetUpResponse;
use crate::twamp_control::StartAck;
use crate::twamp_control::StartSessions;
use crate::twamp_control::StopSessions;
use anyhow::{Result, anyhow};
use deku::prelude::*;
use std::mem::size_of;
use std::net::IpAddr;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;
use tokio::sync::oneshot;
use tracing::*;

/// Control-Client is responsible for initiating and handling TWAMP-Control with a
/// [Server](crate::server::Server).
///
/// Responsibilites of Control-Client on TWAMP-Control, in sequence, are:
/// -   [Read Server Greeting](Self::read_server_greeting)
/// -   [Send Set-Up-Response](Self::send_set_up_response)
/// -   [Read Server-Start](Self::read_server_start)
/// -   [Send Request-TW-Session](Self::send_request_tw_session)
/// -   [Read Accept-Session](Self::read_accept_session)
/// -   [Send Start-Sessions](Self::send_start_sessions)
/// -   [Read Start-Ack](Self::read_start_ack)
/// -   [Send Stop-Sessions](Self::send_stop_sessions)
#[derive(Debug)]
pub struct ControlClient {
    /// TCP stream on which TWAMP-Control is being used.
    pub stream: Option<TcpStream>,
}

impl ControlClient {
    pub fn new() -> Self {
        Self { stream: None }
    }
    /// Initiates TCP connection and starts the [TWAMP-Control](crate::twamp_control) protocol with
    /// Server, handling communication until the test ends or connection is killed/stopped.
    pub async fn do_twamp_control(
        &mut self,
        twamp_control: TcpStream,
        start_session_tx: oneshot::Sender<()>,
        reflector_port_tx: oneshot::Sender<u16>,
        responder_reflect_port: u16,
        controller_port: u16,
        reflector_timeout: u64,
        twamp_test_complete_rx: oneshot::Receiver<()>,
    ) -> Result<()> {
        self.stream = Some(twamp_control);
        self.read_server_greeting().await?;
        self.send_set_up_response().await?;
        self.read_server_start().await?;
        self.send_request_tw_session(responder_reflect_port, controller_port, reflector_timeout)
            .await?;
        let accept_session = self.read_accept_session().await?;
        if accept_session.accept != Accept::Ok {
            return Err(anyhow!("Did not receive Ok in Accept-Session"));
        };

        debug!("Responder provided port: {}", accept_session.port);
        reflector_port_tx.send(accept_session.port).unwrap();
        self.send_start_sessions().await?;
        let start_ack = self.read_start_ack().await?;
        if start_ack.accept != Accept::Ok {
            return Err(anyhow!("Start-Ack should be zero"));
        }
        start_session_tx.send(()).unwrap();
        // testing
        debug!(
            "Waiting for Session-Sender to complete, Control-Client will then send Stop-Sessions."
        );
        let _ = twamp_test_complete_rx.await;
        debug!("Received confirmation that TWAMP-Test is complete. Sending Stop-Sessions");
        self.send_stop_sessions().await?;
        Ok(())
    }

    /// Reads [`ServerGreeting`] from TWAMP-Control stream.
    pub async fn read_server_greeting(&mut self) -> Result<ServerGreeting> {
        let mut buf = [0; size_of::<ServerGreeting>()];
        info!("Reading ServerGreeting");
        self.stream.as_mut().unwrap().read_exact(&mut buf).await?;
        let (_rest, server_greeting) = ServerGreeting::from_bytes((&buf, 0)).expect("should have received valid ServerGreeting");
        debug!("Server greeting: {:?}", server_greeting);
        info!("Done reading ServerGreeting");
        Ok(server_greeting)
    }

    /// Creates a `SetUpResponse`, converts to bytes and sends it out on `TWAMP-Control`.
    pub async fn send_set_up_response(&mut self) -> Result<()> {
        info!("Preparing to send Set-Up-Response");
        let set_up_response = SetUpResponse::new(SecurityMode::Unauthenticated);
        debug!("Set-Up-Response: {:?}", set_up_response);
        let encoded = set_up_response.unwrap().to_bytes().unwrap();
        self.stream
            .as_mut()
            .unwrap()
            .write_all(&encoded[..])
            .await?;
        info!("Set-Up-Response sent");
        Ok(())
    }

    /// Reads from `TWAMP-Control` stream assuming the bytes to be received will be of a
    /// `ServerStart`. Converts those bytes into a `ServerStart` struct and returns it.
    pub async fn read_server_start(&mut self) -> Result<ServerStart> {
        let mut buf = [0; size_of::<ServerStart>()];
        info!("Reading Server-Start");
        self.stream.as_mut().unwrap().read_exact(&mut buf).await?;
        let (_rest, server_start) = ServerStart::from_bytes((&buf, 0)).unwrap();
        debug!("Server-Start: {:?}", server_start);
        info!("Done reading Server-Start");
        Ok(server_start)
    }

    /// Creates a `Request-Tw-Session`, converts to bytes and sends it out on `TWAMP-Control`.
    pub async fn send_request_tw_session(
        &mut self,
        session_reflector_port: u16,
        controller_port: u16,
        timeout: u64,
    ) -> Result<RequestTwSession> {
        info!("Preparing to send Request-TW-Session");
        let stream = self.stream.as_ref().unwrap();
        let sender_address = match stream.local_addr().unwrap().ip() {
            IpAddr::V4(ip) => ip,
            IpAddr::V6(ip) => panic!("da hail did v6 come from: {ip}"),
        };
        let receiver_address = match stream.peer_addr().unwrap().ip() {
            IpAddr::V4(ip) => ip,
            IpAddr::V6(ip) => panic!("da hail did v6 come from: {ip}"),
        };
        debug!(
            "Request-TW-Session reflector port: {}",
            session_reflector_port
        );
        let request_tw_session = RequestTwSession::new(
            sender_address,
            controller_port,
            receiver_address,
            session_reflector_port,
            None,
            timeout,
        );
        debug!("request-tw-session: {:?}", request_tw_session);
        let encoded = request_tw_session.to_bytes().unwrap();
        self.stream
            .as_mut()
            .unwrap()
            .write_all(&encoded[..])
            .await?;
        info!("Request-TW-Session sent");
        Ok(request_tw_session)
    }

    /// Reads from `TWAMP-Control` stream assuming the bytes to be received will be of a
    /// `AcceptSession`. Converts those bytes into a `AcceptSession` struct and returns it.
    pub async fn read_accept_session(&mut self) -> Result<AcceptSession> {
        let mut buf = [0; size_of::<AcceptSession>()];
        info!("Reading Accept-Session");
        self.stream.as_mut().unwrap().read_exact(&mut buf).await?;
        let (_rest, accept_session) = AcceptSession::from_bytes((&buf, 0)).unwrap();
        debug!("Accept-Session: {:?}", accept_session);
        info!("Read Accept-Session");

        Ok(accept_session)
    }

    /// Creates a `Start-Sessions`, converts to bytes and sends it out on `TWAMP-Control`.
    pub async fn send_start_sessions(&mut self) -> Result<()> {
        info!("Preparing to send Start-Sessions");
        let start_sessions = StartSessions::new();
        debug!("Start-Sessions: {:?}", start_sessions);
        let encoded = start_sessions.to_bytes().unwrap();
        self.stream
            .as_mut()
            .unwrap()
            .write_all(&encoded[..])
            .await?;
        info!("Start-Sessions sent");
        Ok(())
    }

    /// Reads from `TWAMP-Control` stream assuming the bytes to be received will be of a
    /// `Start-Ack`. Converts those bytes into a `Start-Ack` struct and returns it.
    pub async fn read_start_ack(&mut self) -> Result<StartAck> {
        let mut buf = [0; size_of::<StartAck>()];
        info!("Reading Start-Ack");
        self.stream.as_mut().unwrap().read_exact(&mut buf).await?;
        let (_rest, start_ack) = StartAck::from_bytes((&buf, 0)).unwrap();
        debug!("Start-Ack: {:?}", start_ack);
        info!("Done reading Start-Ack");
        Ok(start_ack)
    }

    /// Creates a `Stop-Sessions`, converts to bytes and sends it out on `TWAMP-Control`.
    pub async fn send_stop_sessions(&mut self) -> Result<()> {
        info!("Preparing to send Stop-Sessions");
        let stop_sessions = StopSessions::new(Accept::Ok);
        debug!("Stop-Sessions: {:?}", stop_sessions);
        let encoded = stop_sessions.to_bytes().unwrap();
        self.stream
            .as_mut()
            .unwrap()
            .write_all(&encoded[..])
            .await?;
        info!("Stop-Sessions sent");
        Ok(())
    }
}

impl Default for ControlClient {
    /// Construct an empty `ControlClient` with no context.
    fn default() -> Self {
        ControlClient { stream: None }
    }
}