Struct ControlClient 

pub struct ControlClient {
    pub stream: Option<TcpStream>,
}

Control-Client is responsible for initiating and handling TWAMP-Control with a Server.

Responsibilites of Control-Client on TWAMP-Control are:

• Read Server Greeting
• Send Set-Up-Response
• Read Server-Start
• Send Request-TW-Session

Fields

stream: Option<TcpStream>

TCP stream on which TWAMP-Control is being used.

Implementations

impl ControlClient

pub fn new() -> Self

pub async fn do_twamp_control( &mut self, twamp_control: TcpStream, start_session_tx: Sender<()>, reflector_port_tx: Sender<u16>, responder_reflect_port: u16, controller_port: u16, reflector_timeout: u64, twamp_test_complete_rx: Receiver<()>, ) -> Result<()>

Initiates TCP connection and starts the TWAMP-Control protocol with Server, handling communication until the test ends or connection is killed/stopped.

Examples found in repository

examples/controller/controller.rs (lines 62-70)

    pub async fn do_twamp(
        mut self,
        responder_addr: Ipv4Addr,
        responder_port: u16,
        controller_addr: Ipv4Addr,
        mut controller_port: u16,
        responder_reflect_port: u16,
        number_of_test_packets: u32,
        reflector_timeout: u64,
        stop_session_sleep: u64,
    ) -> Result<()> {
        let twamp_control =
            TcpStream::connect(SocketAddrV4::new(responder_addr, responder_port)).await?;
        let udp_socket =
            UdpSocket::bind(SocketAddrV4::new(controller_addr, controller_port)).await?;
        controller_port = udp_socket.local_addr().unwrap().port();

        let (start_session_tx, start_session_rx) = oneshot::channel::<()>();
        let (twamp_test_complete_tx, twamp_test_complete_rx) = oneshot::channel::<()>();
        let (reflector_port_tx, reflector_port_rx) = oneshot::channel::<u16>();
        let control_client_handle = spawn(async move {
            self.control_client
                .do_twamp_control(
                    twamp_control,
                    start_session_tx,
                    reflector_port_tx,
                    responder_reflect_port,
                    controller_port,
                    reflector_timeout,
                    twamp_test_complete_rx,
                )
                .await
                .unwrap();
        });
        let reflected_pkts_vec: Arc<Mutex<Vec<(TwampTestPacketUnauthReflected, TimeStamp)>>> =
            Arc::new(Mutex::new(Vec::new()));
        let reflected_pkts_vec_cloned = Arc::clone(&reflected_pkts_vec);
        let session_sender_handle = spawn(async move {
            // Wait until we get the Accept-Session's port.
            let final_port = reflector_port_rx.await.unwrap();
            debug!("Received reflector port: {}", final_port);
            udp_socket
                .connect(SocketAddrV4::new(responder_addr, final_port))
                .await
                .unwrap();
            // Wait until start-sessions is received
            start_session_rx.await.unwrap();
            debug!("Start-Session identified. Start Session-Sender.");
            self.session_sender = Some(Arc::new(
                SessionSender::new(
                    Arc::new(udp_socket),
                    SocketAddrV4::new(responder_addr, final_port),
                )
                .await,
            ));
            let session_sender_send = Arc::clone(self.session_sender.as_ref().unwrap());
            let session_sender_recv = Arc::clone(self.session_sender.as_ref().unwrap());
            let send_task = spawn(async move {
                let _ = session_sender_send.send_it(number_of_test_packets).await;
                info!("Sent all test packets");
            });
            let recv_task = spawn(async move {
                let _ = session_sender_recv
                    .recv(number_of_test_packets, reflected_pkts_vec_cloned)
                    .await;
                info!("Got back all test packets");
            });
            // wait for all test pkts to be sent.
            send_task.await.unwrap();

            select! {
                // If stop-session-sleep duration finishes before all pkts are received, drop
                // recv task and conclude.
                _ = sleep(Duration::from_secs(stop_session_sleep)) => (),
                // Ignore stop-session-sleep duration if session-sender got all test pkts before
                // duration.
                _ = recv_task => ()
            }
            // Inform Control-Client to send Stop-Sessions
            twamp_test_complete_tx.send(()).unwrap();
        });
        try_join!(control_client_handle, session_sender_handle).unwrap();
        debug!("Control-Client & Session-Sender tasks completed.");
        let acquired_vec = reflected_pkts_vec.lock().await;
        debug!("Reflected pkts len: {}", acquired_vec.len());
        get_metrics(&acquired_vec, number_of_test_packets as f64);
        Ok(())
    }

pub async fn read_server_greeting(&mut self) -> Result<ServerGreeting>

Reads from TWAMP-Control stream assuming the bytes to be received will be of a ServerGreeting. Converts those bytes into a ServerGreeting struct and returns it.

pub async fn send_set_up_response(&mut self) -> Result<()>

Creates a SetUpResponse, converts to bytes and sends it out on TWAMP-Control.

pub async fn read_server_start(&mut self) -> Result<ServerStart>

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 send_request_tw_session( &mut self, session_reflector_port: u16, controller_port: u16, timeout: u64, ) -> Result<RequestTwSession>

Creates a Request-Tw-Session, converts to bytes and sends it out on TWAMP-Control.

pub async fn read_accept_session(&mut self) -> Result<AcceptSession>

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 send_start_sessions(&mut self) -> Result<()>

Creates a Start-Sessions, converts to bytes and sends it out on TWAMP-Control.

pub async fn read_start_ack(&mut self) -> Result<StartAck>

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 send_stop_sessions(&mut self) -> Result<()>

Creates a Stop-Sessions, converts to bytes and sends it out on TWAMP-Control.

Trait Implementations

impl Debug for ControlClient

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Default for ControlClient

fn default() -> Self

Construct an empty ControlClient with no context.