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

use super::*;
use crate::errors::*;

use async_trait::async_trait;
use std::io;
use std::sync::atomic::Ordering;
use util::Conn;

impl Agent {
    // Dial connects to the remote agent, acting as the controlling ice agent.
    // Dial blocks until at least one ice candidate pair has successfully connected.
    pub async fn dial(
        &self,
        remote_ufrag: String,
        remote_pwd: String,
    ) -> Result<Arc<Mutex<impl Conn>>, Error> {
        {
            let mut ai = self.agent_internal.lock().await;
            ai.start_connectivity_checks(true, remote_ufrag, remote_pwd)
                .await?;

            // block until pair selected
            tokio::select! {
                _ = ai.on_connected_rx.recv() => {},
            }
        }
        Ok(Arc::clone(&self.agent_internal))
    }

    // Accept connects to the remote agent, acting as the controlled ice agent.
    // Accept blocks until at least one ice candidate pair has successfully connected.
    pub async fn accept(
        &self,
        remote_ufrag: String,
        remote_pwd: String,
    ) -> Result<Arc<Mutex<impl Conn>>, Error> {
        {
            let mut ai = self.agent_internal.lock().await;
            ai.start_connectivity_checks(false, remote_ufrag, remote_pwd)
                .await?;

            // block until pair selected
            tokio::select! {
                _ = ai.on_connected_rx.recv() => {},
            }
        }

        Ok(Arc::clone(&self.agent_internal))
    }
}

impl AgentInternal {
    // bytes_sent returns the number of bytes sent
    pub fn bytes_sent(&self) -> usize {
        self.bytes_sent.load(Ordering::SeqCst)
    }

    // bytes_received returns the number of bytes received
    pub fn bytes_received(&self) -> usize {
        self.bytes_received.load(Ordering::SeqCst)
    }
}

#[async_trait]
impl Conn for AgentInternal {
    async fn connect(&self, _addr: SocketAddr) -> io::Result<()> {
        Err(io::Error::new(io::ErrorKind::Other, "Not applicable"))
    }

    async fn recv(&self, buf: &mut [u8]) -> io::Result<usize> {
        if self.done_tx.is_none() {
            return Err(io::Error::new(io::ErrorKind::Other, "Conn is closed"));
        }

        let mut n = 0;
        if let Some(buffer) = &self.buffer {
            n = match buffer.read(buf, None).await {
                Ok(n) => n,
                Err(err) => return Err(io::Error::new(io::ErrorKind::Other, err.to_string())),
            };
            self.bytes_received.fetch_add(n, Ordering::SeqCst);
        }

        Ok(n)
    }

    async fn recv_from(&self, _buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> {
        Err(io::Error::new(io::ErrorKind::Other, "Not applicable"))
    }

    async fn send(&self, buf: &[u8]) -> io::Result<usize> {
        if self.done_tx.is_none() {
            return Err(io::Error::new(io::ErrorKind::Other, "Conn is closed"));
        }

        if is_message(buf) {
            return Err(io::Error::new(
                io::ErrorKind::Other,
                ERR_ICE_WRITE_STUN_MESSAGE.to_string(),
            ));
        }

        let result = if let Some(pair) = self.get_selected_pair() {
            pair.write(buf).await
        } else if let Some(pair) = self.get_best_available_candidate_pair() {
            pair.write(buf).await
        } else {
            Err(ERR_NO_CANDIDATE_PAIRS.to_owned())
        };

        match result {
            Ok(n) => {
                self.bytes_sent.fetch_add(buf.len(), Ordering::SeqCst);
                Ok(n)
            }
            Err(err) => Err(io::Error::new(io::ErrorKind::Other, err.to_string())),
        }
    }

    async fn send_to(&self, _buf: &[u8], _target: SocketAddr) -> io::Result<usize> {
        Err(io::Error::new(io::ErrorKind::Other, "Not applicable"))
    }

    fn local_addr(&self) -> io::Result<SocketAddr> {
        if let Some(pair) = self.get_selected_pair() {
            Ok(pair.local.addr())
        } else {
            Err(io::Error::new(
                io::ErrorKind::AddrNotAvailable,
                "Addr Not Available",
            ))
        }
    }
}