webrtc-connection 0.2.0

WebRTC connection.
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

use futures::Future;
use std::{sync::Arc, pin::Pin}; 
use cs_trace::{Tracer, child};
use connection_utils::Channel;
use tokio::{sync::{mpsc::{self, Sender}, Mutex}};
use webrtc::data_channel::{RTCDataChannel, data_channel_message::DataChannelMessage};

mod async_read;
mod async_write;

pub struct RtcChannel {
    trace: Box<dyn Tracer>,
    id: u16,
    label: String,
    channel: Arc<Mutex<Arc<RTCDataChannel>>>,
    current_read_future: Option<Pin<Box<dyn Future<Output = Result<DataChannelMessage, std::io::Error>>>>>,
    current_write_future: Option<Pin<Box<dyn Future<Output = Result<usize, std::io::Error>>>>>,
    read_message_queue: Arc<Mutex<mpsc::Receiver<DataChannelMessage>>>,
}

// TODO: fix
unsafe impl Send for RtcChannel {}
unsafe impl Sync for RtcChannel {}

async fn setup_listeners(
    channel: Arc<RTCDataChannel>,
    sender: Arc<Sender<DataChannelMessage>>,
) {
    let on_message_channel = Arc::clone(&channel);
    channel.on_open(Box::new(move || {
        return Box::pin(async move {
            on_message_channel.on_message(Box::new(move |message: DataChannelMessage| {
                let read_sender = Arc::clone(&sender);
                return Box::pin(async move {
                    read_sender
                        .send(message)
                        .await
                        .expect("Cannot send data channel message to the queue.");
                });
            })).await;
        });
    })).await;
}

impl RtcChannel {
    pub async fn new(
        trace: &Box<dyn Tracer>,
        channel: Arc<RTCDataChannel>,
    ) -> RtcChannel {
        let (
            read_queue_sender,
            read_message_queue,
        ) = mpsc::channel(100);

        setup_listeners(
            Arc::clone(&channel),
            Arc::new(read_queue_sender),
        ).await;

        return RtcChannel {
            id: channel.id(),
            label: channel.label().to_string(),
            trace: child!(trace, "rtc-channel"),
            channel: Arc::new(Mutex::new(channel)),
            read_message_queue: Arc::new(Mutex::new(read_message_queue)),
            current_read_future: None,
            current_write_future: None,
        };
    }

    pub fn id(&self) -> u16 {
        return self.id;
    }

    pub fn label(&self) -> &String {
        return &self.label;
    }
}

impl Channel for RtcChannel {
    fn id(&self) -> u16 {
        return self.id;
    }


    fn label(&self) ->  &String {
        return &self.label;
    }
}