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
}
}