use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;
use tracing::debug;
use crate::api::common::error::MediaTransportError;
use crate::api::server::transport::core::connection::ClientConnection;
use crate::{RtpSsrc};
pub async fn is_ssrc_demultiplexing_enabled(
ssrc_demultiplexing_enabled: &Arc<RwLock<bool>>,
) -> Result<bool, MediaTransportError> {
Ok(*ssrc_demultiplexing_enabled.read().await)
}
pub async fn enable_ssrc_demultiplexing(
ssrc_demultiplexing_enabled: &Arc<RwLock<bool>>,
) -> Result<bool, MediaTransportError> {
if *ssrc_demultiplexing_enabled.read().await {
return Ok(true);
}
*ssrc_demultiplexing_enabled.write().await = true;
debug!("Enabled SSRC demultiplexing on server");
Ok(true)
}
pub async fn register_client_ssrc(
client_id: &str,
ssrc: RtpSsrc,
ssrc_demultiplexing_enabled: &Arc<RwLock<bool>>,
clients: &Arc<RwLock<HashMap<String, ClientConnection>>>,
) -> Result<bool, MediaTransportError> {
if !*ssrc_demultiplexing_enabled.read().await {
return Err(MediaTransportError::ConfigError("SSRC demultiplexing is not enabled".to_string()));
}
let clients_guard = clients.read().await;
let client = clients_guard.get(client_id)
.ok_or_else(|| MediaTransportError::ClientNotFound(client_id.to_string()))?;
if !client.connected {
return Err(MediaTransportError::ClientNotConnected(client_id.to_string()));
}
let mut session = client.session.lock().await;
let created = session.create_stream_for_ssrc(ssrc).await;
if created {
debug!("Pre-registered SSRC {:08x} for client {}", ssrc, client_id);
} else {
debug!("SSRC {:08x} was already registered for client {}", ssrc, client_id);
}
Ok(created)
}
pub async fn get_client_ssrcs(
client_id: &str,
clients: &Arc<RwLock<HashMap<String, ClientConnection>>>,
) -> Result<Vec<u32>, MediaTransportError> {
let clients_guard = clients.read().await;
let client = clients_guard.get(client_id)
.ok_or_else(|| MediaTransportError::ClientNotFound(client_id.to_string()))?;
if !client.connected {
return Err(MediaTransportError::ClientNotConnected(client_id.to_string()));
}
let session = client.session.lock().await;
let ssrcs = session.get_all_ssrcs().await;
Ok(ssrcs)
}
pub async fn find_clients_by_ssrc(
ssrc: RtpSsrc,
clients: &Arc<RwLock<HashMap<String, ClientConnection>>>,
) -> Result<Vec<String>, MediaTransportError> {
let mut result = Vec::new();
let clients_guard = clients.read().await;
for (client_id, client) in clients_guard.iter() {
if !client.connected {
continue;
}
let session = client.session.lock().await;
let ssrcs = session.get_all_ssrcs().await;
if ssrcs.contains(&ssrc) {
result.push(client_id.clone());
}
}
Ok(result)
}
pub async fn map_ssrc_to_client_id(
ssrc: RtpSsrc,
clients: &Arc<RwLock<HashMap<String, ClientConnection>>>,
) -> Result<Option<String>, MediaTransportError> {
let matches = find_clients_by_ssrc(ssrc, clients).await?;
Ok(matches.into_iter().next())
}