webrtc 0.17.1

A pure Rust implementation of WebRTC API
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

use std::io;
use std::net::SocketAddr;
use std::sync::atomic::Ordering;

use async_trait::async_trait;
use portable_atomic::AtomicUsize;
use util::conn::conn_pipe::pipe;

use super::*;
use crate::mux::mux_func::{match_all, match_srtp};

const TEST_PIPE_BUFFER_SIZE: usize = 8192;

#[tokio::test]
async fn test_no_endpoints() -> crate::error::Result<()> {
    // In memory pipe
    let (ca, _) = pipe();

    let mut m = Mux::new(Config {
        conn: Arc::new(ca),
        buffer_size: TEST_PIPE_BUFFER_SIZE,
    });

    Mux::dispatch(&[0], &m.endpoints).await?;
    m.close().await;

    Ok(())
}

struct MuxErrorConn {
    idx: AtomicUsize,
    data: Vec<Vec<u8>>,
}

type Result<T> = std::result::Result<T, util::Error>;

#[async_trait]
impl Conn for MuxErrorConn {
    async fn connect(&self, _addr: SocketAddr) -> Result<()> {
        Err(io::Error::other("Not applicable").into())
    }

    async fn recv(&self, buf: &mut [u8]) -> Result<usize> {
        let idx = self.idx.fetch_add(1, Ordering::SeqCst);
        if idx < self.data.len() {
            let n = std::cmp::min(buf.len(), self.data[idx].len());
            buf[..n].copy_from_slice(&self.data[idx][..n]);
            Ok(n)
        } else {
            Err(io::Error::other(format!("idx {} >= data.len {}", idx, self.data.len())).into())
        }
    }

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

    async fn send(&self, _buf: &[u8]) -> Result<usize> {
        Err(io::Error::other("Not applicable").into())
    }

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

    fn local_addr(&self) -> Result<SocketAddr> {
        Err(io::Error::other("Not applicable").into())
    }

    fn remote_addr(&self) -> Option<SocketAddr> {
        None
    }

    async fn close(&self) -> Result<()> {
        Ok(())
    }

    fn as_any(&self) -> &(dyn std::any::Any + Send + Sync) {
        self
    }
}

#[tokio::test]
async fn test_non_fatal_read() -> Result<()> {
    let expected_data = b"expected_data".to_vec();

    let conn = Arc::new(MuxErrorConn {
        idx: AtomicUsize::new(0),
        data: vec![
            expected_data.clone(),
            expected_data.clone(),
            expected_data.clone(),
        ],
    });

    let mut m = Mux::new(Config {
        conn,
        buffer_size: TEST_PIPE_BUFFER_SIZE,
    });

    let e = m.new_endpoint(Box::new(match_all)).await;
    let mut buff = vec![0u8; TEST_PIPE_BUFFER_SIZE];

    let n = e.recv(&mut buff).await?;
    assert_eq!(&buff[..n], expected_data);

    let n = e.recv(&mut buff).await?;
    assert_eq!(&buff[..n], expected_data);

    let n = e.recv(&mut buff).await?;
    assert_eq!(&buff[..n], expected_data);

    m.close().await;

    Ok(())
}

#[tokio::test]
async fn test_non_fatal_dispatch() -> Result<()> {
    let (ca, cb) = pipe();

    let mut m = Mux::new(Config {
        conn: Arc::new(ca),
        buffer_size: TEST_PIPE_BUFFER_SIZE,
    });

    let e = m.new_endpoint(Box::new(match_srtp)).await;
    e.buffer.set_limit_size(1).await;

    for _ in 0..25 {
        let srtp_packet = [128, 1, 2, 3, 4].to_vec();
        cb.send(&srtp_packet).await?;
    }

    m.close().await;

    Ok(())
}