webrtc-connection 0.2.0

use std::sync::Arc;
use cs_trace::{Tracer, create_trace};
use cs_utils::futures::GenericCodec;
use tokio::{io::{AsyncRead, AsyncWrite}, sync::{Mutex, mpsc::{Sender, Receiver}}};
use tokio_util::codec::Framed;
use futures::{StreamExt, stream::{SplitStream, SplitSink}};
use webrtc::{peer_connection::{configuration::RTCConfiguration, RTCPeerConnection}, api::{APIBuilder, media_engine::MediaEngine, interceptor_registry::register_default_interceptors}, interceptor::registry::Registry};
use connection_utils::{Channel, Disconnected};

use crate::types::RTCSignalingMessage;

mod event_handlers;

mod disconnected;
mod connected;

pub struct RtcConnection<TAsyncDuplexStream: AsyncRead + AsyncWrite + Unpin + Send + 'static> {
    trace: Box<dyn Tracer>,
    signaling_source: Arc<Mutex<SplitStream<Framed<Box<TAsyncDuplexStream>, GenericCodec<RTCSignalingMessage>>>>>,
    signaling_sink: Arc<Mutex<SplitSink<Framed<Box<TAsyncDuplexStream>, GenericCodec<RTCSignalingMessage>>, RTCSignalingMessage>>>,
    peer_connection: Arc<RTCPeerConnection>,
    on_data_channel_sink: Arc<Mutex<Sender<Box<dyn Channel>>>>,
    on_data_channel_source: Option<Receiver<Box<dyn Channel>>>,
}

impl<TAsyncDuplexStream: AsyncRead + AsyncWrite + Unpin + Send + 'static> RtcConnection<TAsyncDuplexStream> {
    pub async fn new(
        stream: Box<TAsyncDuplexStream>,
        config: RTCConfiguration,
    ) -> anyhow::Result<Box<dyn Disconnected>> {
        let trace = create_trace!("rtc-connection");
        let codec = GenericCodec::new();
        let framed_stream = Framed::new(stream, codec);
        let (
            signaling_sink,
            signaling_source
        ) = framed_stream.split();

        // create a MediaEngine object to configure the supported codec
        let mut media_engine = MediaEngine::default();
        // register default codecs
        media_engine.register_default_codecs()?;

        // create a InterceptorRegistry. this is the user configurable RTP/RTCP Pipeline.
        // this provides NACKs, RTCP Reports and other features. if you use `webrtc.NewPeerConnection`
        // this is enabled by default. if you are manually managing You MUST create a interceptorRegistry
        // for each peerConnection.
        let mut registry = Registry::new();
        // use the default set of Interceptors
        registry = register_default_interceptors(registry, &mut media_engine).await?;

        // create the API object with the MediaEngine
        let api = APIBuilder::new()
            .with_media_engine(media_engine)
            .with_interceptor_registry(registry)
            .build();

        // create a new RTCPeerConnection
        let peer_connection = Arc::new(
            api.new_peer_connection(config).await?
        );

        let (
            on_data_channel_sink,
            on_data_channel_source,
        ) = tokio::sync::mpsc::channel(10);
        
        return Ok(
            Box::new(
                RtcConnection {
                    trace,
                    signaling_source: Arc::new(Mutex::new(signaling_source)),
                    signaling_sink: Arc::new(Mutex::new(signaling_sink)),
                    peer_connection,
                    on_data_channel_sink: Arc::new(Mutex::new(on_data_channel_sink)),
                    on_data_channel_source: Some(on_data_channel_source),
                },
            ),
        );
    }
}