enigma_rtc/

session.rs

use std::sync::{Arc, Mutex};

use async_trait::async_trait;
use tokio::sync::{mpsc, Mutex as AsyncMutex};

use crate::config::RtcConfig;
use crate::error::{EnigmaRtcError, RtcResult};
use crate::signaling::decode_signaling;
use crate::types::{CallState, RtcEvent, SignalingMessage};
use crate::webrtc::{RtcEngine, WebRtcEngine};

pub struct CallSession {
    state: Mutex<CallState>,
    engine: Arc<dyn RtcEngine>,
    event_tx: mpsc::UnboundedSender<RtcEvent>,
    event_rx: AsyncMutex<mpsc::UnboundedReceiver<RtcEvent>>,
    config: RtcConfig,
}

impl CallSession {
    pub fn new(config: RtcConfig) -> RtcResult<Self> {
        let (event_tx, event_rx) = mpsc::unbounded_channel();
        let engine = Arc::new(WebRtcEngine::new(config.clone())?);
        engine.set_event_sender(event_tx.clone())?;
        Ok(Self::from_parts(config, engine, event_tx, event_rx))
    }

    pub fn with_engine<E>(config: RtcConfig, engine: Arc<E>) -> RtcResult<Self>
    where
        E: RtcEngine + 'static,
    {
        let (event_tx, event_rx) = mpsc::unbounded_channel();
        engine.set_event_sender(event_tx.clone())?;
        let engine: Arc<dyn RtcEngine> = engine;
        Ok(Self::from_parts(config, engine, event_tx, event_rx))
    }

    pub fn state(&self) -> CallState {
        match self.state.lock() {
            Ok(guard) => *guard,
            Err(poisoned) => *poisoned.into_inner(),
        }
    }

    pub fn config(&self) -> &RtcConfig {
        &self.config
    }

    pub fn create_offer(&self) -> RtcResult<SignalingMessage> {
        {
            let mut state = self
                .state
                .lock()
                .map_err(|_| EnigmaRtcError::InvalidState)?;
            if *state != CallState::Idle {
                return Err(EnigmaRtcError::InvalidState);
            }
            *state = CallState::CreatingOffer;
        }
        self.emit_state(CallState::CreatingOffer)?;
        let sdp = self.engine.create_offer()?;
        if sdp.trim().is_empty() {
            return Err(EnigmaRtcError::InvalidSdp);
        }
        {
            let mut state = self
                .state
                .lock()
                .map_err(|_| EnigmaRtcError::InvalidState)?;
            *state = CallState::WaitingAnswer;
        }
        self.emit_state(CallState::WaitingAnswer)?;
        Ok(SignalingMessage::Offer { sdp })
    }

    pub fn accept_offer(&self, offer_sdp: &str) -> RtcResult<SignalingMessage> {
        if offer_sdp.trim().is_empty() {
            return Err(EnigmaRtcError::InvalidSdp);
        }
        {
            let mut state = self
                .state
                .lock()
                .map_err(|_| EnigmaRtcError::InvalidState)?;
            if *state != CallState::Idle {
                return Err(EnigmaRtcError::InvalidState);
            }
            *state = CallState::IncomingOffer;
        }
        self.emit_state(CallState::IncomingOffer)?;
        let sdp = self.engine.create_answer(offer_sdp)?;
        if sdp.trim().is_empty() {
            return Err(EnigmaRtcError::InvalidSdp);
        }
        {
            let mut state = self
                .state
                .lock()
                .map_err(|_| EnigmaRtcError::InvalidState)?;
            *state = CallState::Connected;
        }
        self.emit_state(CallState::Connected)?;
        self.emit_event(RtcEvent::ConnectionEstablished)?;
        Ok(SignalingMessage::Answer { sdp })
    }

    pub fn accept_answer(&self, answer_sdp: &str) -> RtcResult<()> {
        if answer_sdp.trim().is_empty() {
            return Err(EnigmaRtcError::InvalidSdp);
        }
        {
            let state = self
                .state
                .lock()
                .map_err(|_| EnigmaRtcError::InvalidState)?;
            if *state != CallState::WaitingAnswer {
                return Err(EnigmaRtcError::InvalidState);
            }
        }
        self.engine.apply_answer(answer_sdp)?;
        {
            let mut state = self
                .state
                .lock()
                .map_err(|_| EnigmaRtcError::InvalidState)?;
            *state = CallState::Connected;
        }
        self.emit_state(CallState::Connected)?;
        self.emit_event(RtcEvent::ConnectionEstablished)
    }

    pub fn add_ice_candidate(&self, candidate_json: &str) -> RtcResult<()> {
        {
            let state = self
                .state
                .lock()
                .map_err(|_| EnigmaRtcError::InvalidState)?;
            match *state {
                CallState::WaitingAnswer | CallState::IncomingOffer | CallState::Connected => {}
                _ => return Err(EnigmaRtcError::InvalidState),
            }
        }
        let message = decode_signaling(candidate_json)?;
        match message {
            SignalingMessage::IceCandidate { .. } => {
                self.engine.add_remote_candidate(candidate_json)
            }
            _ => Err(EnigmaRtcError::InvalidCandidate),
        }
    }

    pub fn hangup(&self, reason: Option<&str>) -> RtcResult<SignalingMessage> {
        {
            let mut state = self
                .state
                .lock()
                .map_err(|_| EnigmaRtcError::InvalidState)?;
            match *state {
                CallState::Idle
                | CallState::CreatingOffer
                | CallState::WaitingAnswer
                | CallState::IncomingOffer
                | CallState::Connected => {
                    *state = CallState::Ending;
                }
                CallState::Ending | CallState::Ended => return Err(EnigmaRtcError::InvalidState),
            }
        }
        self.emit_state(CallState::Ending)?;
        {
            let mut state = self
                .state
                .lock()
                .map_err(|_| EnigmaRtcError::InvalidState)?;
            *state = CallState::Ended;
        }
        self.emit_state(CallState::Ended)?;
        self.emit_event(RtcEvent::ConnectionClosed)?;
        Ok(SignalingMessage::Hangup {
            reason: reason
                .filter(|s| !s.trim().is_empty())
                .map(|s| s.to_string()),
        })
    }

    pub fn set_microphone_enabled(&self, enabled: bool) -> RtcResult<()> {
        {
            let state = self
                .state
                .lock()
                .map_err(|_| EnigmaRtcError::InvalidState)?;
            if matches!(*state, CallState::Ended) {
                return Err(EnigmaRtcError::InvalidState);
            }
        }
        self.engine.set_microphone_enabled(enabled)
    }

    pub fn set_camera_enabled(&self, enabled: bool) -> RtcResult<()> {
        {
            let state = self
                .state
                .lock()
                .map_err(|_| EnigmaRtcError::InvalidState)?;
            if matches!(*state, CallState::Ended) {
                return Err(EnigmaRtcError::InvalidState);
            }
        }
        self.engine.set_camera_enabled(enabled)
    }

    fn emit_state(&self, state: CallState) -> RtcResult<()> {
        self.emit_event(RtcEvent::StateChanged(state))
    }

    fn emit_event(&self, event: RtcEvent) -> RtcResult<()> {
        self.event_tx
            .send(event)
            .map_err(|_| EnigmaRtcError::ChannelClosed)
    }

    fn from_parts(
        config: RtcConfig,
        engine: Arc<dyn RtcEngine>,
        event_tx: mpsc::UnboundedSender<RtcEvent>,
        event_rx: mpsc::UnboundedReceiver<RtcEvent>,
    ) -> Self {
        Self {
            state: Mutex::new(CallState::Idle),
            engine,
            event_tx,
            event_rx: AsyncMutex::new(event_rx),
            config,
        }
    }
}

#[async_trait]
pub(crate) trait SessionEventReceiver: Send + Sync {
    async fn recv_event(&self) -> RtcResult<RtcEvent>;
}

#[async_trait]
impl SessionEventReceiver for CallSession {
    async fn recv_event(&self) -> RtcResult<RtcEvent> {
        let mut rx = self.event_rx.lock().await;
        match rx.recv().await {
            Some(event) => Ok(event),
            None => Err(EnigmaRtcError::ChannelClosed),
        }
    }
}

impl CallSession {
    pub async fn next_event(&self) -> RtcResult<RtcEvent> {
        self.recv_event().await
    }
}