talksik/client-sdk-rust
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

feat: add ffi datachannel & mute events (#88)

Browse Source
This commit is contained in:
Théo Monnom
2023-06-18 22:45:05 +02:00
committed by GitHub
parent 33dfcb27b0
commit 25f4c9a075
69 changed files with 1934 additions and 1546 deletions
Download Patch File Download Diff File Expand all files Collapse all files
livekit/src/room/id.rs
+8 -14
View File
@@ -3,16 +3,6 @@ use std::fmt;
macro_rules! id_str {
($($name:ident;)*) => {
$(
impl $name {
pub fn new(str: String) -> Self {
Self(str)
}
pub fn as_str(&self) -> &str {
&self.0
}
}
impl From<String> for $name {
fn from(str: String) -> $name {
$name(str)
@@ -41,16 +31,20 @@ macro_rules! id_str {
}
#[derive(Clone, Default, Debug, Eq, Hash, PartialEq, PartialOrd, Ord)]
pub struct ParticipantSid(String);
#[repr(transparent)]
pub struct ParticipantSid(pub String);
#[derive(Clone, Default, Debug, Eq, Hash, PartialEq, PartialOrd, Ord)]
pub struct ParticipantIdentity(String);
#[repr(transparent)]
pub struct ParticipantIdentity(pub String);
#[derive(Clone, Default, Debug, Eq, Hash, PartialEq, PartialOrd, Ord)]
pub struct TrackSid(String);
#[repr(transparent)]
pub struct TrackSid(pub String);
#[derive(Clone, Default, Debug, Eq, Hash, PartialEq, PartialOrd, Ord)]
pub struct RoomSid(String);
#[repr(transparent)]
pub struct RoomSid(pub String);
id_str! {
ParticipantSid;
livekit/src/room/mod.rs
+93 -77
View File
@@ -9,12 +9,10 @@ use livekit_protocol::observer::Dispatcher;
use parking_lot::{Mutex, RwLock, RwLockReadGuard};
use std::collections::HashMap;
use std::fmt::Debug;
use std::sync::atomic::{AtomicU8, Ordering};
use std::sync::Arc;
use thiserror::Error;
use tokio::sync::{mpsc, oneshot};
use tokio::task::JoinHandle;
use tracing::{error, info, instrument, trace, Level};
pub use crate::rtc_engine::SimulateScenario;
@@ -82,7 +80,7 @@ pub enum RoomEvent {
},
DataReceived {
payload: Arc<Vec<u8>>,
kind: proto::data_packet::Kind,
kind: DataPacketKind,
participant: RemoteParticipant,
},
ConnectionStateChanged(ConnectionState),
@@ -100,18 +98,41 @@ pub enum ConnectionState {
Unknown,
}
#[derive(Debug, Clone, Copy, Eq, PartialEq)]
pub enum DataPacketKind {
Lossy,
Reliable,
}
#[derive(Debug, Clone)]
pub struct RoomOptions {
pub auto_subscribe: bool,
pub adaptive_stream: bool,
pub dynacast: bool,
}
impl Default for RoomOptions {
fn default() -> Self {
Self {
auto_subscribe: true,
adaptive_stream: false,
dynacast: false,
}
}
}
struct RoomHandle {
session_task: JoinHandle<()>,
close_emitter: oneshot::Sender<()>,
}
pub struct Room {
inner: Arc<SessionInner>,
inner: Arc<RoomSession>,
handle: Mutex<Option<RoomHandle>>,
}
impl Debug for Room {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
f.debug_struct("Room")
.field("sid", &self.sid())
.field("name", &self.name())
@@ -124,14 +145,21 @@ impl Room {
pub async fn connect(
url: &str,
token: &str,
options: RoomOptions,
) -> RoomResult<(Self, mpsc::UnboundedReceiver<RoomEvent>)> {
let (rtc_engine, engine_events) = RtcEngine::new();
let (rtc_engine, engine_events) = RtcEngine::connect(
url,
token,
SignalOptions {
auto_subscribe: options.auto_subscribe,
adaptive_stream: options.adaptive_stream,
..Default::default()
},
)
.await?;
let rtc_engine = Arc::new(rtc_engine);
rtc_engine
.connect(url, token, SignalOptions::default())
.await?;
let join_response = rtc_engine.join_response().unwrap();
let join_response = rtc_engine.join_response();
let pi = join_response.participant.unwrap().clone();
let local_participant = LocalParticipant::new(
rtc_engine.clone(),
@@ -142,11 +170,13 @@ impl Room {
);
let room_info = join_response.room.unwrap();
let inner = Arc::new(SessionInner {
state: AtomicU8::new(ConnectionState::Disconnected as u8),
sid: Mutex::new(room_info.sid.into()),
name: Mutex::new(room_info.name),
metadata: Mutex::new(room_info.metadata),
let inner = Arc::new(RoomSession {
sid: room_info.sid.into(),
name: room_info.name,
info: RwLock::new(RoomInfo {
state: ConnectionState::Disconnected,
metadata: room_info.metadata,
}),
participants: Default::default(),
participants_tasks: Default::default(),
active_speakers: Default::default(),
@@ -196,15 +226,15 @@ impl Room {
}
pub fn sid(&self) -> RoomSid {
self.inner.sid.lock().clone()
self.inner.sid.clone()
}
pub fn name(&self) -> String {
self.inner.name.lock().clone()
self.inner.name.clone()
}
pub fn metadata(&self) -> String {
self.inner.metadata.lock().clone()
self.inner.info.read().metadata.clone()
}
pub fn local_participant(&self) -> LocalParticipant {
@@ -212,7 +242,7 @@ impl Room {
}
pub fn connection_state(&self) -> ConnectionState {
self.inner.state.load(Ordering::Acquire).try_into().unwrap()
self.inner.info.read().state
}
pub fn participants(&self) -> RwLockReadGuard<HashMap<ParticipantSid, RemoteParticipant>> {
@@ -224,20 +254,24 @@ impl Room {
}
}
struct SessionInner {
state: AtomicU8, // ConnectionState
sid: Mutex<RoomSid>,
name: Mutex<String>,
metadata: Mutex<String>,
participants: RwLock<HashMap<ParticipantSid, RemoteParticipant>>,
participants_tasks: RwLock<HashMap<ParticipantSid, (JoinHandle<()>, oneshot::Sender<()>)>>,
active_speakers: RwLock<Vec<Participant>>,
rtc_engine: Arc<RtcEngine>,
local_participant: LocalParticipant,
dispatcher: Dispatcher<RoomEvent>,
struct RoomInfo {
metadata: String,
state: ConnectionState,
}
impl Debug for SessionInner {
pub(crate) struct RoomSession {
rtc_engine: Arc<RtcEngine>,
sid: RoomSid,
name: String,
info: RwLock<RoomInfo>,
dispatcher: Dispatcher<RoomEvent>,
active_speakers: RwLock<Vec<Participant>>,
local_participant: LocalParticipant,
participants: RwLock<HashMap<ParticipantSid, RemoteParticipant>>,
participants_tasks: RwLock<HashMap<ParticipantSid, (JoinHandle<()>, oneshot::Sender<()>)>>,
}
impl Debug for RoomSession {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("SessionInner")
.field("sid", &self.sid)
@@ -247,8 +281,7 @@ impl Debug for SessionInner {
}
}
impl SessionInner {
#[instrument(level = Level::DEBUG)]
impl RoomSession {
async fn room_task(
self: Arc<Self>,
mut engine_events: EngineEvents,
@@ -259,20 +292,19 @@ impl SessionInner {
res = engine_events.recv() => {
if let Some(event) = res {
if let Err(err) = self.on_engine_event(event).await {
error!("failed to handle engine event: {:?}", err);
log::error!("failed to handle engine event: {:?}", err);
}
}
},
_ = &mut close_receiver => {
trace!("closing room_task");
log::trace!("closing room_task");
break;
}
}
}
}
/// Listen to the Participant events and forward them to the Room Dispatcher
#[instrument(level = Level::DEBUG)]
/// Forward participant events to the room dispatcher
async fn participant_task(
self: Arc<Self>,
participant: Participant,
@@ -284,19 +316,18 @@ impl SessionInner {
res = participant_events.recv() => {
if let Some(event) = res {
if let Err(err) = self.on_participant_event(&participant, event).await {
error!("failed to handle participant event for {:?}: {:?}", participant.sid(), err);
log::error!("failed to handle participant event for {:?}: {:?}", participant.sid(), err);
}
}
},
_ = &mut close_rx => {
trace!("closing participant_task for {:?}", participant.sid());
log::trace!("closing participant_task for {:?}", participant.sid());
break;
},
}
}
}
#[instrument(level = Level::DEBUG)]
async fn on_participant_event(
self: &Arc<Self>,
participant: &Participant,
@@ -337,7 +368,6 @@ impl SessionInner {
Ok(())
}
#[instrument(level = Level::DEBUG)]
async fn on_engine_event(self: &Arc<Self>, event: EngineEvent) -> RoomResult<()> {
match event {
EngineEvent::ParticipantUpdate { updates } => self.handle_participant_update(updates),
@@ -414,21 +444,20 @@ impl SessionInner {
Ok(())
}
#[instrument(level = Level::DEBUG)]
async fn close(&self) {
self.rtc_engine.close().await;
}
/// Change the connection state and emit an event
/// Does nothing if the state is already the same
#[instrument(level = Level::DEBUG)]
/// Returns true if the state changed
fn update_connection_state(&self, state: ConnectionState) -> bool {
let old_state = self.state.load(Ordering::Acquire);
if old_state == state as u8 {
let mut info = self.info.write();
if info.state == state {
return false;
}
self.state.store(state as u8, Ordering::Release);
info.state = state;
self.dispatcher
.dispatch(&RoomEvent::ConnectionStateChanged(state));
return true;
@@ -437,7 +466,6 @@ impl SessionInner {
/// Update the participants inside a Room.
/// It'll create, update or remove a participant
/// It also update the participant tracks.
#[instrument(level = Level::DEBUG)]
fn handle_participant_update(self: &Arc<Self>, updates: Vec<proto::ParticipantInfo>) {
for pi in updates {
if pi.sid == self.local_participant.sid()
@@ -452,7 +480,7 @@ impl SessionInner {
if let Some(remote_participant) = remote_participant {
if pi.state == proto::participant_info::State::Disconnected as i32 {
// Participant disconnected
info!("Participant disconnected: {}", pi.sid);
log::info!("Participant disconnected: {}", pi.sid);
self.clone()
.handle_participant_disconnect(remote_participant)
} else {
@@ -461,7 +489,7 @@ impl SessionInner {
}
} else {
// Create a new participant
info!("Participant connected: {}", pi.sid);
log::info!("Participant connected: {}", pi.sid);
let remote_participant = {
let pi = pi.clone();
self.create_participant(pi.sid.into(), pi.identity.into(), pi.name, pi.metadata)
@@ -478,7 +506,6 @@ impl SessionInner {
/// Active speakers changed
/// Update the participants & sort the active_speakers by audio_level
#[instrument(level = Level::DEBUG)]
fn handle_speakers_changed(&self, speakers_info: Vec<proto::SpeakerInfo>) {
let mut speakers = Vec::new();
@@ -513,7 +540,6 @@ impl SessionInner {
/// Handle a connection quality update
/// Emit ConnectionQualityChanged event for the concerned participants
#[instrument(level = Level::DEBUG)]
fn handle_connection_quality_update(&self, updates: Vec<proto::ConnectionQualityInfo>) {
for update in updates {
let participant = {
@@ -542,7 +568,6 @@ impl SessionInner {
}
}
#[instrument(level = Level::DEBUG)]
fn handle_restarting(self: &Arc<Self>) {
// Remove existing participants/subscriptions on full reconnect
for (_, participant) in self.participants.read().iter() {
@@ -555,10 +580,9 @@ impl SessionInner {
}
}
#[instrument(level = Level::DEBUG)]
fn handle_restarted(self: &Arc<Self>) {
// Full reconnect succeeded!
let join_response = self.rtc_engine.join_response().unwrap();
let join_response = self.rtc_engine.join_response();
self.update_connection_state(ConnectionState::Connected);
self.dispatcher.dispatch(&RoomEvent::Reconnected);
@@ -569,22 +593,19 @@ impl SessionInner {
self.handle_participant_update(join_response.other_participants);
// TODO(theomonnom): Synchronize states
// TODO(theomonnom): Room info changed?
// TODO(theomonnom): unpublish & republish tracks
}
#[instrument(level = Level::DEBUG)]
fn handle_disconnected(&self) {
if self.state.load(Ordering::Acquire) == ConnectionState::Disconnected as u8 {
return;
if self.update_connection_state(ConnectionState::Disconnected) {
self.dispatcher.dispatch(&RoomEvent::Disconnected);
}
self.update_connection_state(ConnectionState::Disconnected);
self.dispatcher.dispatch(&RoomEvent::Disconnected);
}
/// Create a new participant
/// Also add it to the participants list
#[instrument(level = Level::DEBUG)]
fn create_participant(
self: &Arc<Self>,
sid: ParticipantSid,
@@ -592,7 +613,13 @@ impl SessionInner {
name: String,
metadata: String,
) -> RemoteParticipant {
let participant = RemoteParticipant::new(sid.clone(), identity, name, metadata);
let participant = RemoteParticipant::new(
self.rtc_engine.clone(),
sid.clone(),
identity,
name,
metadata,
);
// Create the participant task
let (close_tx, close_rx) = oneshot::channel();
@@ -611,7 +638,6 @@ impl SessionInner {
/// A participant has disconnected
/// Cleanup the participant and emit an event
#[instrument(level = Level::DEBUG)]
fn handle_participant_disconnect(self: Arc<Self>, remote_participant: RemoteParticipant) {
tokio::spawn(async move {
for (sid, _) in &*remote_participant.tracks() {
@@ -619,11 +645,12 @@ impl SessionInner {
}
// Close the participant task
if let Some((task, close_tx)) = self
let ptask = self
.participants_tasks
.write()
.remove(&remote_participant.sid())
{
.remove(&remote_participant.sid());
if let Some((task, close_tx)) = ptask {
let _ = close_tx.send(());
let _ = task.await;
}
@@ -649,14 +676,3 @@ fn unpack_stream_id(stream_id: &str) -> Option<(&str, &str)> {
None
}
}
impl From<u8> for ConnectionState {
fn from(value: u8) -> Self {
match value {
0 => ConnectionState::Disconnected,
1 => ConnectionState::Connected,
2 => ConnectionState::Reconnecting,
_ => ConnectionState::Unknown,
}
}
}
livekit/src/room/options.rs
+1 -32
View File
@@ -58,36 +58,6 @@ impl AudioPreset {
}
}
#[derive(Debug, Clone)]
pub struct AudioCaptureOptions {
pub echo_cancellation: bool,
pub noise_suppression: bool,
pub auto_gain_control: bool,
}
impl Default for AudioCaptureOptions {
fn default() -> Self {
Self {
echo_cancellation: true,
noise_suppression: true,
auto_gain_control: true,
}
}
}
#[derive(Clone, Debug)]
pub struct VideoCaptureOptions {
pub resolution: VideoResolution,
}
impl Default for VideoCaptureOptions {
fn default() -> Self {
Self {
resolution: video::H720.resolution(),
}
}
}
#[derive(Clone, Debug)]
pub struct TrackPublishOptions {
// If the encodings aren't set, LiveKit will compute the most appropriate ones
@@ -97,7 +67,7 @@ pub struct TrackPublishOptions {
pub dtx: bool,
pub red: bool,
pub simulcast: bool,
pub name: String,
// pub name: String,
pub source: TrackSource,
}
@@ -110,7 +80,6 @@ impl Default for TrackPublishOptions {
dtx: true,
red: true,
simulcast: true,
name: "unnamed track".to_owned(),
source: TrackSource::Unknown,
}
}
livekit/src/room/participant/local_participant.rs
+37 -29
View File
@@ -1,10 +1,12 @@
use super::{ConnectionQuality, ParticipantInner};
use super::ConnectionQuality;
use super::ParticipantInternal;
use crate::options;
use crate::options::compute_video_encodings;
use crate::options::video_layers_from_encodings;
use crate::options::TrackPublishOptions;
use crate::prelude::*;
use crate::rtc_engine::RtcEngine;
use crate::DataPacketKind;
use livekit_protocol as proto;
use livekit_webrtc::rtp_parameters::RtpEncodingParameters;
use parking_lot::RwLockReadGuard;
@@ -12,12 +14,10 @@ use std::collections::HashMap;
use std::fmt::Debug;
use std::sync::Arc;
use tokio::sync::mpsc;
use tracing::debug;
#[derive(Clone)]
pub struct LocalParticipant {
inner: Arc<ParticipantInner>,
rtc_engine: Arc<RtcEngine>,
inner: Arc<ParticipantInternal>,
}
impl Debug for LocalParticipant {
@@ -39,8 +39,9 @@ impl LocalParticipant {
metadata: String,
) -> Self {
Self {
inner: Arc::new(ParticipantInner::new(sid, identity, name, metadata)),
rtc_engine,
inner: Arc::new(ParticipantInternal::new(
rtc_engine, sid, identity, name, metadata,
)),
}
}
@@ -51,7 +52,7 @@ impl LocalParticipant {
) -> RoomResult<LocalTrackPublication> {
let mut req = proto::AddTrackRequest {
cid: track.rtc_track().id(),
name: options.name.clone(),
name: track.name().clone(),
r#type: proto::TrackType::from(track.kind()) as i32,
muted: track.is_muted(),
source: proto::TrackSource::from(options.source) as i32,
@@ -65,9 +66,9 @@ impl LocalParticipant {
LocalTrack::Video(video_track) => {
// Get the video dimension
// TODO(theomonnom): Use MediaStreamTrack::getSettings() on web
let capture_options = video_track.capture_options();
req.width = capture_options.resolution.width;
req.height = capture_options.resolution.height;
let resolution = video_track.rtc_source().video_resolution();
req.width = resolution.width;
req.height = resolution.height;
encodings = compute_video_encodings(req.width, req.height, &options);
req.layers = video_layers_from_encodings(req.width, req.height, &encodings);
@@ -85,29 +86,34 @@ impl LocalParticipant {
});
}
}
let track_info = self.inner.rtc_engine.add_track(req).await?;
let publication = LocalTrackPublication::new(
track_info.clone(),
Arc::downgrade(&self.inner),
track.clone(),
);
track.update_info(track_info); // Update sid + source
let track_info = self.rtc_engine.add_track(req).await?;
let publication =
LocalTrackPublication::new(track_info.clone(), track.clone(), options.clone());
track.update_info(track_info); // Update SID + Source
debug!("publishing track with cid {:?}", track.rtc_track().id());
log::debug!("publishing track with cid {:?}", track.rtc_track().id());
let transceiver = self
.inner
.rtc_engine
.create_sender(track.clone(), options, encodings)
.await?;
track.update_transceiver(Some(transceiver));
track.start();
//track.start();
track.enable();
tokio::spawn({
let rtc_engine = self.rtc_engine.clone();
let rtc_engine = self.inner.rtc_engine.clone();
async move {
let _ = rtc_engine.negotiate_publisher().await;
}
});
self.inner
.add_track_publication(TrackPublication::Local(publication.clone()));
.add_publication(TrackPublication::Local(publication.clone()));
self.inner
.dispatcher
@@ -125,9 +131,10 @@ impl LocalParticipant {
) -> RoomResult<LocalTrackPublication> {
let mut tracks = self.inner.tracks.write();
if let Some(TrackPublication::Local(publication)) = tracks.remove(&track) {
let track = publication.track().unwrap();
let track = publication.track();
let sender = track.transceiver().unwrap().sender();
self.rtc_engine.remove_track(sender).await?;
self.inner.rtc_engine.remove_track(sender).await?;
track.update_transceiver(None);
self.inner
@@ -135,10 +142,10 @@ impl LocalParticipant {
.dispatch(&ParticipantEvent::LocalTrackUnpublished {
publication: publication.clone(),
});
publication.update_track(None);
// publication.update_track(None);
tokio::spawn({
let rtc_engine = self.rtc_engine.clone();
let rtc_engine = self.inner.rtc_engine.clone();
async move {
let _ = rtc_engine.negotiate_publisher().await;
}
@@ -152,20 +159,21 @@ impl LocalParticipant {
pub async fn publish_data(
&self,
data: &[u8],
kind: proto::data_packet::Kind,
) -> Result<(), RoomError> {
data: Vec<u8>,
kind: DataPacketKind,
destination_sids: Vec<String>,
) -> RoomResult<()> {
let data = proto::DataPacket {
kind: kind as i32,
value: Some(proto::data_packet::Value::User(proto::UserPacket {
participant_sid: self.sid().to_string(),
payload: data.to_vec(),
destination_sids: vec![],
payload: data,
destination_sids: destination_sids.to_owned(),
..Default::default()
})),
};
self.rtc_engine
self.inner
.rtc_engine
.publish_data(&data, kind)
.await
.map_err(Into::into)
livekit/src/room/participant/mod.rs
+59 -44
View File
@@ -1,14 +1,15 @@
use crate::prelude::*;
use crate::rtc_engine::RtcEngine;
use crate::track::TrackError;
use livekit_protocol as proto;
use livekit_protocol::enum_dispatch;
use livekit_protocol::observer::Dispatcher;
use parking_lot::{Mutex, RwLock, RwLockReadGuard};
use parking_lot::{RwLock, RwLockReadGuard};
use std::collections::HashMap;
use std::fmt::Debug;
use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU8, Ordering};
use std::sync::Arc;
use tokio::sync::mpsc;
use std::thread::JoinHandle;
use tokio::sync::{mpsc, oneshot};
mod local_participant;
mod remote_participant;
@@ -38,7 +39,7 @@ pub enum ParticipantEvent {
},
DataReceived {
payload: Arc<Vec<u8>>,
kind: proto::data_packet::Kind,
kind: DataPacketKind,
},
SpeakingChanged {
speaking: bool,
@@ -109,7 +110,6 @@ impl Participant {
pub fn tracks(self: &Self) -> RwLockReadGuard<HashMap<TrackSid, TrackPublication>>;
pub fn register_observer(self: &Self) -> mpsc::UnboundedReceiver<ParticipantEvent>;
// Internal functions
pub(crate) fn set_speaking(self: &Self, speaking: bool) -> ();
pub(crate) fn set_audio_level(self: &Self, level: f32) -> ();
pub(crate) fn set_connection_quality(self: &Self, quality: ConnectionQuality) -> ();
@@ -118,56 +118,76 @@ impl Participant {
}
#[derive(Debug)]
pub(crate) struct ParticipantInner {
pub sid: Mutex<ParticipantSid>,
pub identity: Mutex<ParticipantIdentity>,
pub name: Mutex<String>,
pub metadata: Mutex<String>,
pub speaking: AtomicBool,
pub tracks: RwLock<HashMap<TrackSid, TrackPublication>>,
pub audio_level: AtomicU32,
pub connection_quality: AtomicU8,
pub dispatcher: Dispatcher<ParticipantEvent>,
pub(crate) struct ParticipantInfo {
pub sid: ParticipantSid,
pub identity: ParticipantIdentity,
pub name: String,
pub metadata: String,
pub speaking: bool,
pub audio_level: f32,
pub connection_quality: ConnectionQuality,
}
impl ParticipantInner {
#[derive(Debug)]
pub(crate) struct ParticipantInternal {
pub(super) rtc_engine: Arc<RtcEngine>,
pub(super) dispatcher: Dispatcher<ParticipantEvent>,
info: RwLock<ParticipantInfo>,
tracks: RwLock<HashMap<TrackSid, TrackPublication>>,
tracks_tasks: RwLock<HashMap<TrackSid, (JoinHandle<()>, oneshot::Sender<()>)>>,
}
impl ParticipantInternal {
pub fn new(
rtc_engine: Arc<RtcEngine>,
sid: ParticipantSid,
identity: ParticipantIdentity,
name: String,
metadata: String,
) -> Self {
Self {
sid: Mutex::new(sid),
identity: Mutex::new(identity),
name: Mutex::new(name),
metadata: Mutex::new(metadata),
tracks: Default::default(),
speaking: Default::default(),
audio_level: Default::default(),
connection_quality: AtomicU8::new(ConnectionQuality::Unknown as u8),
rtc_engine,
info: RwLock::new(ParticipantInfo {
sid,
identity,
name,
metadata,
speaking: false,
audio_level: 0.0,
connection_quality: ConnectionQuality::Unknown,
}),
dispatcher: Default::default(),
tracks: Default::default(),
tracks_tasks: Default::default(),
}
}
pub fn update_info(&self, new_info: proto::ParticipantInfo) {
let mut info = self.info.write();
info.sid = new_info.sid.into();
info.name = new_info.name;
info.identity = new_info.identity.into();
info.metadata = new_info.metadata; // TODO(theomonnom): callback MetadataChanged
}
pub fn sid(&self) -> ParticipantSid {
self.sid.lock().clone()
self.info.read().sid.clone()
}
pub fn identity(&self) -> ParticipantIdentity {
self.identity.lock().clone()
self.info.read().identity.clone()
}
pub fn name(&self) -> String {
self.name.lock().clone()
self.info.read().name.clone()
}
pub fn metadata(&self) -> String {
self.metadata.lock().clone()
self.info.read().metadata.clone()
}
pub fn is_speaking(&self) -> bool {
self.speaking.load(Ordering::SeqCst)
self.info.read().speaking
}
pub fn tracks(&self) -> RwLockReadGuard<HashMap<TrackSid, TrackPublication>> {
@@ -175,39 +195,34 @@ impl ParticipantInner {
}
pub fn audio_level(&self) -> f32 {
f32::from_bits(self.audio_level.load(Ordering::SeqCst))
self.info.read().audio_level
}
pub fn connection_quality(&self) -> ConnectionQuality {
self.connection_quality.load(Ordering::SeqCst).into()
self.info.read().connection_quality
}
pub fn register_observer(&self) -> mpsc::UnboundedReceiver<ParticipantEvent> {
self.dispatcher.register()
}
pub fn update_info(&self, info: proto::ParticipantInfo) {
*self.sid.lock() = info.sid.into();
*self.identity.lock() = info.identity.into();
*self.name.lock() = info.name;
*self.metadata.lock() = info.metadata; // TODO(theomonnom): callback MetadataChanged
}
pub fn set_speaking(&self, speaking: bool) {
self.speaking.store(speaking, Ordering::SeqCst);
self.info.write().speaking = speaking;
}
pub fn set_audio_level(&self, audio_level: f32) {
self.audio_level
.store(audio_level.to_bits(), Ordering::SeqCst)
self.info.write().audio_level = audio_level;
}
pub fn set_connection_quality(&self, quality: ConnectionQuality) {
self.connection_quality
.store(quality as u8, Ordering::SeqCst);
self.info.write().connection_quality = quality;
}
pub fn add_track_publication(&self, publication: TrackPublication) {
pub fn remove_publication(&self, sid: &TrackSid) {
self.tracks.write().remove(sid);
}
pub fn add_publication(&self, publication: TrackPublication) {
self.tracks.write().insert(publication.sid(), publication);
}
}
livekit/src/room/participant/remote_participant.rs
+34 -27
View File
@@ -1,6 +1,8 @@
use super::{ConnectionQuality, ParticipantInner};
use crate::prelude::*;
use super::TrackKind;
use super::{ConnectionQuality, ParticipantInternal};
use crate::rtc_engine::RtcEngine;
use crate::track::TrackError;
use crate::{prelude::*, DataPacketKind};
use livekit_protocol as proto;
use livekit_webrtc::prelude::*;
use parking_lot::RwLockReadGuard;
@@ -10,13 +12,12 @@ use std::sync::Arc;
use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::timeout;
use tracing::{debug, error, instrument, Level};
const ADD_TRACK_TIMEOUT: Duration = Duration::from_secs(5);
#[derive(Clone)]
pub struct RemoteParticipant {
inner: Arc<ParticipantInner>,
inner: Arc<ParticipantInternal>,
}
impl Debug for RemoteParticipant {
@@ -31,29 +32,22 @@ impl Debug for RemoteParticipant {
impl RemoteParticipant {
pub(crate) fn new(
rtc_engine: Arc<RtcEngine>,
sid: ParticipantSid,
identity: ParticipantIdentity,
name: String,
metadata: String,
) -> Self {
Self {
inner: Arc::new(ParticipantInner::new(sid, identity, name, metadata)),
inner: Arc::new(ParticipantInternal::new(
rtc_engine, sid, identity, name, metadata,
)),
}
}
#[inline]
pub fn get_track_publication(&self, sid: &TrackSid) -> Option<RemoteTrackPublication> {
self.inner.tracks.read().get(sid).map(|track| {
if let TrackPublication::Remote(remote) = track {
return remote.clone();
}
unreachable!()
})
}
/// Called by the RoomSession when receiving data from the RrcSession
/// Called by the RoomSession when receiving data from the RtcSession
/// It is just used to emit the Data event on the participant dispatcher.
pub(crate) fn on_data_received(&self, data: Arc<Vec<u8>>, kind: proto::data_packet::Kind) {
pub(crate) fn on_data_received(&self, data: Arc<Vec<u8>>, kind: DataPacketKind) {
self.inner
.dispatcher
.dispatch(&ParticipantEvent::DataReceived {
@@ -62,7 +56,6 @@ impl RemoteParticipant {
});
}
#[instrument(level = Level::DEBUG)]
pub(crate) async fn add_subscribed_media_track(
&self,
sid: TrackSid,
@@ -78,7 +71,7 @@ impl RemoteParticipant {
return publication;
}
tokio::task::yield_now().await; // Remove yield
tokio::time::sleep(Duration::from_millis(50)).await;
}
}
};
@@ -111,10 +104,10 @@ impl RemoteParticipant {
}
};
debug!("starting track: {:?}", sid);
log::debug!("starting track: {:?}", sid);
remote_publication.update_track(Some(track.clone().into()));
track.set_muted(remote_publication.is_muted());
//track.set_muted(remote_publication.is_muted());
track.update_info(proto::TrackInfo {
sid: remote_publication.sid().to_string(),
name: remote_publication.name().to_string(),
@@ -124,8 +117,9 @@ impl RemoteParticipant {
});
self.inner
.add_track_publication(TrackPublication::Remote(remote_publication.clone()));
track.start();
.add_publication(TrackPublication::Remote(remote_publication.clone()));
// track.start();
track.enable();
self.inner
.dispatcher
@@ -134,7 +128,7 @@ impl RemoteParticipant {
publication: remote_publication,
});
} else {
error!("could not find published track with sid: {:?}", sid);
log::error!("could not find published track with sid: {:?}", sid);
self.inner
.dispatcher
@@ -149,7 +143,7 @@ impl RemoteParticipant {
if let Some(publication) = self.get_track_publication(sid) {
// Unsubscribe to the track if needed
if let Some(track) = publication.track() {
track.stop();
track.disable();
self.inner
.dispatcher
@@ -159,6 +153,8 @@ impl RemoteParticipant {
});
}
self.inner.remove_publication(sid);
self.inner
.dispatcher
.dispatch(&ParticipantEvent::TrackUnpublished {
@@ -177,9 +173,10 @@ impl RemoteParticipant {
if let Some(publication) = self.get_track_publication(&track.sid.clone().into()) {
publication.update_info(track.clone());
} else {
let publication = RemoteTrackPublication::new(track.clone(), None);
let publication =
RemoteTrackPublication::new(track.clone(), Arc::downgrade(&self.inner), None);
self.inner
.add_track_publication(TrackPublication::Remote(publication.clone()));
.add_publication(TrackPublication::Remote(publication.clone()));
// This is a new track, dispatch publish event
self.inner
@@ -200,6 +197,16 @@ impl RemoteParticipant {
}
}
#[inline]
pub fn get_track_publication(&self, sid: &TrackSid) -> Option<RemoteTrackPublication> {
self.inner.tracks.read().get(sid).map(|track| {
if let TrackPublication::Remote(remote) = track {
return remote.clone();
}
unreachable!()
})
}
#[inline]
pub fn sid(&self) -> ParticipantSid {
self.inner.sid()
livekit/src/room/publication/local.rs
+42 -14
View File
@@ -1,17 +1,13 @@
use super::TrackPublicationInner;
use crate::id::TrackSid;
use crate::options::TrackPublishOptions;
use crate::track::{LocalTrack, Track, TrackDimension, TrackKind, TrackSource};
use crate::participant::ParticipantInternal;
use crate::track::{LocalTrack, TrackDimension, TrackKind, TrackSource};
use livekit_protocol as proto;
use parking_lot::Mutex;
use std::sync::Arc;
use std::sync::{Arc, Weak};
#[derive(Debug)]
struct LocalTrackPublicationInner {
publication_inner: TrackPublicationInner,
#[allow(unused)] // TODO(theomonnom)
options: Mutex<TrackPublishOptions>,
}
#[derive(Clone, Debug)]
@@ -22,17 +18,47 @@ pub struct LocalTrackPublication {
impl LocalTrackPublication {
pub(crate) fn new(
info: proto::TrackInfo,
participant: Weak<ParticipantInternal>,
track: LocalTrack,
options: TrackPublishOptions,
) -> Self {
Self {
inner: Arc::new(LocalTrackPublicationInner {
publication_inner: TrackPublicationInner::new(info, Some(track.into())),
options: Mutex::new(options),
publication_inner: TrackPublicationInner::new(
info,
participant,
Some(track.into()),
),
}),
}
}
pub async fn mute(&self) {}
pub async fn unmute(&self) {}
pub async fn pause_upstream(&self) {}
pub async fn resume_upstream(&self) {}
/*pub fn set_muted(&self, muted: bool) {
if self.is_muted() == muted {
return;
}
self.track().rtc_track().set_enabled(!muted);
let participant = self.inner.publication_inner.participant().upgrade();
if participant.is_none() {
log::warn!("publication's participant is invalid, set_muted failed");
return;
}
let participant = participant.unwrap();
// Engine update muted
// Participant MUTED/UNMUTED event
}*/
#[inline]
pub fn sid(&self) -> TrackSid {
self.inner.publication_inner.sid()
@@ -64,11 +90,13 @@ impl LocalTrackPublication {
}
#[inline]
pub fn track(&self) -> Option<LocalTrack> {
pub fn track(&self) -> LocalTrack {
self.inner
.publication_inner
.track()
.map(|track| track.try_into().unwrap())
.unwrap()
.try_into()
.unwrap()
}
#[inline]
@@ -86,10 +114,10 @@ impl LocalTrackPublication {
false
}
#[inline]
/*#[inline]
pub(crate) fn update_track(&self, track: Option<Track>) {
self.inner.publication_inner.update_track(track);
}
}*/
#[allow(dead_code)]
#[inline]
livekit/src/room/publication/mod.rs
+204 -124
View File
@@ -1,140 +1,49 @@
use super::track::{TrackDimension, TrackEvent};
use super::track::TrackDimension;
use crate::participant::ParticipantInternal;
use crate::prelude::*;
use crate::track::Track;
use futures_util::stream::StreamExt;
use livekit_protocol as proto;
use livekit_protocol::enum_dispatch;
use livekit_protocol::observer::Dispatcher;
use parking_lot::Mutex;
use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
use parking_lot::{Mutex, RwLock};
use proto::observer::Dispatcher;
use std::sync::Arc;
use std::sync::Weak;
use tokio::sync::Notify;
use tokio_stream::wrappers::UnboundedReceiverStream;
mod local;
pub use local::*;
mod remote;
pub use local::*;
pub use remote::*;
#[derive(Debug)]
pub(crate) struct TrackPublicationInner {
track: Mutex<Option<Track>>,
name: Mutex<String>,
sid: Mutex<TrackSid>,
kind: AtomicU8, // Casted to TrackKind
source: AtomicU8, // Casted to TrackSource
simulcasted: AtomicBool,
dimension: Mutex<TrackDimension>,
mime_type: Mutex<String>,
muted: AtomicBool,
dispatcher: Dispatcher<TrackEvent>,
close_notifier: Arc<Notify>,
#[derive(Debug, Clone)]
pub enum PublicationEvent {
Muted,
Unmuted,
Subscribed,
Unsubscribed,
SubscriptionStatusChanged {
old_state: SubscriptionStatus,
new_state: SubscriptionStatus,
},
SubscriptionPermissionChanged {
old_state: PermissionStatus,
new_state: PermissionStatus,
},
SubscriptionFailed,
}
impl TrackPublicationInner {
pub fn new(info: proto::TrackInfo, track: Option<Track>) -> Self {
Self {
track: Mutex::new(track),
name: Mutex::new(info.name),
sid: Mutex::new(info.sid.into()),
kind: AtomicU8::new(
TrackKind::try_from(proto::TrackType::from_i32(info.r#type).unwrap()).unwrap()
as u8,
),
source: AtomicU8::new(TrackSource::from(
proto::TrackSource::from_i32(info.source).unwrap(),
) as u8),
simulcasted: AtomicBool::new(info.simulcast),
dimension: Mutex::new(TrackDimension(info.width, info.height)),
mime_type: Mutex::new(info.mime_type),
muted: AtomicBool::new(info.muted),
dispatcher: Default::default(),
close_notifier: Default::default(),
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SubscriptionStatus {
Desired,
Subscribed,
Unsubscribed,
}
pub fn update_track(&self, track: Option<Track>) {
let mut old_track = self.track.lock();
*old_track = track.clone();
self.close_notifier.notify_waiters();
if let Some(track) = track.as_ref() {
let track_stream = UnboundedReceiverStream::new(track.register_observer());
tokio::spawn({
let dispatcher = self.dispatcher.clone();
let notifier = self.close_notifier.clone();
async move {
let notified = notifier.notified();
futures_util::pin_mut!(notified);
futures_util::future::select(
track_stream.map(Ok).forward(dispatcher),
notified,
)
.await;
}
});
}
}
pub fn update_info(&self, info: proto::TrackInfo) {
*self.name.lock() = info.name;
*self.sid.lock() = info.sid.into();
*self.dimension.lock() = TrackDimension(info.width, info.height);
*self.mime_type.lock() = info.mime_type;
self.kind.store(
TrackKind::try_from(proto::TrackType::from_i32(info.r#type).unwrap()).unwrap() as u8,
Ordering::SeqCst,
);
self.source.store(
TrackSource::from(proto::TrackSource::from_i32(info.source).unwrap()) as u8,
Ordering::SeqCst,
);
self.simulcasted.store(info.simulcast, Ordering::SeqCst);
self.muted.store(info.muted, Ordering::SeqCst);
if let Some(track) = self.track.lock().as_ref() {
track.set_muted(info.muted);
}
}
pub fn sid(&self) -> TrackSid {
self.sid.lock().clone()
}
pub fn name(&self) -> String {
self.name.lock().clone()
}
pub fn kind(&self) -> TrackKind {
self.kind.load(Ordering::SeqCst).try_into().unwrap()
}
pub fn source(&self) -> TrackSource {
self.source.load(Ordering::SeqCst).into()
}
pub fn simulcasted(&self) -> bool {
self.simulcasted.load(Ordering::Relaxed)
}
pub fn dimension(&self) -> TrackDimension {
self.dimension.lock().clone()
}
pub fn mime_type(&self) -> String {
self.mime_type.lock().clone()
}
pub fn track(&self) -> Option<Track> {
self.track.lock().clone()
}
pub fn is_muted(&self) -> bool {
self.muted.load(Ordering::Relaxed)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PermissionStatus {
Allowed,
NotAllowed,
}
#[derive(Clone, Debug)]
@@ -159,8 +68,179 @@ impl TrackPublication {
pub fn track(&self) -> Option<Track> {
match self {
TrackPublication::Local(p) => p.track().map(Into::into),
TrackPublication::Local(p) => Some(p.track().into()),
TrackPublication::Remote(p) => p.track().map(Into::into),
}
}
}
#[derive(Debug)]
pub(crate) struct PublicationInfo {
track: Option<Track>,
name: String,
sid: TrackSid,
kind: TrackKind,
source: TrackSource,
simulcasted: bool,
dimension: TrackDimension,
mime_type: String,
muted: bool,
}
#[derive(Debug)]
pub(crate) struct TrackPublicationInner {
info: RwLock<PublicationInfo>,
dispatcher: Dispatcher<PublicationEvent>,
participant: Weak<ParticipantInternal>,
//forward_handle: Mutex<Option<tokio::task::JoinHandle<()>>>,
forward_close: Arc<Notify>,
}
impl TrackPublicationInner {
pub fn new(
info: proto::TrackInfo,
participant: Weak<ParticipantInternal>,
track: Option<Track>,
) -> Self {
let info = PublicationInfo {
track,
name: info.name,
sid: info.sid.into(),
kind: proto::TrackType::from_i32(info.r#type)
.unwrap()
.try_into()
.unwrap(),
source: proto::TrackSource::from_i32(info.source)
.unwrap()
.try_into()
.unwrap(),
simulcasted: info.simulcast,
dimension: TrackDimension(info.width, info.height),
mime_type: info.mime_type,
muted: info.muted,
};
Self {
info: RwLock::new(info),
dispatcher: Default::default(),
participant,
//forward_handle: Default::default(),
forward_close: Default::default(),
}
}
// Forward track events to the publication events
// e.g: this also allow us to access the signal_client and notify the server if
// a local track changed mute state
async fn track_forward_task(
close_notifier: Weak<Notify>,
track: Track,
dispatcher: Dispatcher<PublicationEvent>,
) {
let mut track_events = track.register_observer();
loop {
let notifier = close_notifier.upgrade();
if notifier.is_none() {
break;
}
let notified = notifier.as_ref().unwrap().notified();
tokio::select! {
_ = notified => {
break;
}
Some(event) = track_events.recv() => {
match event {
TrackEvent::Muted => {
dispatcher.dispatch(&PublicationEvent::Muted);
}
TrackEvent::Unmuted => {
dispatcher.dispatch(&PublicationEvent::Unmuted);
}
}
}
}
}
}
pub fn update_track(&self, track: Option<Track>) {
//let forward_task = self.forward_handle.lock().take();
//if let Some(task) = forward_task {
// Make sure to close the old forwarder before changing the track
self.forward_close.notify_waiters();
//let _ = task.await;
// }
let mut info = self.info.write();
info.track = track.clone();
if let Some(track) = track {
let _handle = tokio::spawn(Self::track_forward_task(
Arc::downgrade(&self.forward_close),
track,
self.dispatcher.clone(),
));
//let mut forward_handle = self.forward_handle.lock();
//*forward_handle = Some(handle);
}
}
// Called when updating a participant info
pub fn update_info(&self, new_info: proto::TrackInfo) {
let mut info = self.info.write();
info.name = new_info.name;
info.sid = new_info.sid.into();
info.dimension = TrackDimension(new_info.width, new_info.height);
info.mime_type = new_info.mime_type;
info.kind =
TrackKind::try_from(proto::TrackType::from_i32(new_info.r#type).unwrap()).unwrap();
info.source = TrackSource::from(proto::TrackSource::from_i32(new_info.source).unwrap());
info.simulcasted = new_info.simulcast;
// TODO MUTE ?????????????????
// info.muted = new_info.muted;
// if let Some(track) = info.track.as_ref() {
// track.set_muted(info.muted);
// }
}
pub fn participant(&self) -> Weak<ParticipantInternal> {
self.participant.clone()
}
pub fn sid(&self) -> TrackSid {
self.info.read().sid.clone()
}
pub fn name(&self) -> String {
self.info.read().name.clone()
}
pub fn kind(&self) -> TrackKind {
self.info.read().kind
}
pub fn source(&self) -> TrackSource {
self.info.read().source
}
pub fn simulcasted(&self) -> bool {
self.info.read().simulcasted
}
pub fn dimension(&self) -> TrackDimension {
self.info.read().dimension.clone()
}
pub fn mime_type(&self) -> String {
self.info.read().mime_type.clone()
}
pub fn track(&self) -> Option<Track> {
self.info.read().track.clone()
}
pub fn is_muted(&self) -> bool {
self.info.read().muted
}
}
livekit/src/room/publication/remote.rs
+122 -16
View File
@@ -1,64 +1,170 @@
use super::TrackPublicationInner;
use super::{PermissionStatus, SubscriptionStatus, TrackPublicationInner};
use crate::id::TrackSid;
use crate::participant::ParticipantInternal;
use crate::publication::PublicationEvent;
use crate::track::{RemoteTrack, Track, TrackDimension, TrackKind, TrackSource};
use livekit_protocol as proto;
use std::sync::Arc;
use parking_lot::RwLock;
use std::sync::{Arc, Weak};
#[derive(Debug)]
struct RemoteInfo {
subscribed: bool,
allowed: bool,
// TODO(theomonnom): other remote info
}
#[derive(Debug)]
struct RemoteInner {
publication_inner: TrackPublicationInner,
info: RwLock<RemoteInfo>,
}
#[derive(Clone, Debug)]
pub struct RemoteTrackPublication {
inner: Arc<TrackPublicationInner>,
inner: Arc<RemoteInner>,
}
impl RemoteTrackPublication {
pub(crate) fn new(info: proto::TrackInfo, track: Option<RemoteTrack>) -> Self {
pub(crate) fn new(
info: proto::TrackInfo,
participant: Weak<ParticipantInternal>,
track: Option<RemoteTrack>,
) -> Self {
Self {
inner: Arc::new(TrackPublicationInner::new(info, track.map(Into::into))),
inner: Arc::new(RemoteInner {
publication_inner: TrackPublicationInner::new(
info,
participant,
track.map(Into::into),
),
info: RwLock::new(RemoteInfo {
subscribed: false,
allowed: false,
}),
}),
}
}
pub fn set_subscribed(&self, subscribed: bool) {
let old_subscription_state = self.subscription_status();
let old_permission_state = self.permission_status();
let mut info = self.inner.info.write();
info.subscribed = subscribed;
if subscribed {
info.allowed = true;
}
let participant = self.inner.publication_inner.participant.upgrade();
if participant.is_none() {
log::warn!("publication's participant is invalid, set_subscribed failed");
return;
}
let participant = participant.unwrap();
let update_subscription = proto::UpdateSubscription {
track_sids: vec![self.sid().0],
subscribe: subscribed,
participant_tracks: vec![proto::ParticipantTracks {
participant_sid: participant.sid().0,
track_sids: vec![self.sid().0],
}],
};
// Engine update subscription
if old_subscription_state != self.subscription_status() {
self.inner.publication_inner.dispatcher.dispatch(
&PublicationEvent::SubscriptionStatusChanged {
old_state: old_subscription_state,
new_state: self.subscription_status(),
},
)
}
if old_permission_state != self.permission_status() {
self.inner.publication_inner.dispatcher.dispatch(
&PublicationEvent::SubscriptionPermissionChanged {
old_state: old_permission_state,
new_state: self.permission_status(),
},
)
}
}
#[inline]
pub fn subscription_status(&self) -> SubscriptionStatus {
if !self.inner.info.read().subscribed {
return SubscriptionStatus::Unsubscribed;
}
if self.track().is_none() {
return SubscriptionStatus::Desired;
}
SubscriptionStatus::Subscribed
}
#[inline]
pub fn permission_status(&self) -> PermissionStatus {
if self.inner.info.read().allowed {
PermissionStatus::Allowed
} else {
PermissionStatus::NotAllowed
}
}
pub fn is_subscribed(&self) -> bool {
self.inner.info.read().allowed && self.track().is_some()
}
#[inline]
pub fn sid(&self) -> TrackSid {
self.inner.sid()
self.inner.publication_inner.sid()
}
#[inline]
pub fn name(&self) -> String {
self.inner.name()
self.inner.publication_inner.name()
}
#[inline]
pub fn kind(&self) -> TrackKind {
self.inner.kind()
self.inner.publication_inner.kind()
}
#[inline]
pub fn source(&self) -> TrackSource {
self.inner.source()
self.inner.publication_inner.source()
}
#[inline]
pub fn simulcasted(&self) -> bool {
self.inner.simulcasted()
self.inner.publication_inner.simulcasted()
}
#[inline]
pub fn dimension(&self) -> TrackDimension {
self.inner.dimension()
self.inner.publication_inner.dimension()
}
#[inline]
pub fn track(&self) -> Option<RemoteTrack> {
self.inner.track().map(|track| track.try_into().unwrap())
self.inner
.publication_inner
.track()
.map(|track| track.try_into().unwrap())
}
#[inline]
pub fn mime_type(&self) -> String {
self.inner.mime_type()
self.inner.publication_inner.mime_type()
}
#[inline]
pub fn is_muted(&self) -> bool {
self.inner.is_muted()
self.inner.publication_inner.is_muted()
}
#[inline]
@@ -68,11 +174,11 @@ impl RemoteTrackPublication {
#[inline]
pub(crate) fn update_track(&self, track: Option<Track>) {
self.inner.update_track(track);
self.inner.publication_inner.update_track(track);
}
#[inline]
pub(crate) fn update_info(&self, info: proto::TrackInfo) {
self.inner.update_info(info);
self.inner.publication_inner.update_info(info);
}
}
livekit/src/room/track/local_audio_track.rs
+51 -56
View File
@@ -1,24 +1,17 @@
use super::TrackInner;
use crate::options::AudioCaptureOptions;
use crate::prelude::*;
use crate::rtc_engine::lk_runtime::LkRuntime;
use crate::webrtc::peer_connection_factory::native::PeerConnectionFactoryExt;
use core::panic;
use livekit_protocol as proto;
use livekit_webrtc::prelude::*;
use parking_lot::Mutex;
use std::fmt::Debug;
use std::sync::Arc;
use tokio::sync::mpsc;
#[derive(Debug)]
pub struct LocalAudioTrackInner {
track_inner: TrackInner,
capture_options: Mutex<AudioCaptureOptions>,
}
#[derive(Clone)]
pub struct LocalAudioTrack {
inner: Arc<LocalAudioTrackInner>,
inner: Arc<TrackInner>,
source: RtcAudioSource,
}
impl Debug for LocalAudioTrack {
@@ -32,85 +25,84 @@ impl Debug for LocalAudioTrack {
}
impl LocalAudioTrack {
pub(crate) fn new(
name: String,
rtc_track: RtcAudioTrack,
capture_options: AudioCaptureOptions,
) -> Self {
pub(crate) fn new(name: String, rtc_track: RtcAudioTrack, source: RtcAudioSource) -> Self {
Self {
inner: Arc::new(LocalAudioTrackInner {
track_inner: TrackInner::new(
"unknown".to_string().into(), // sid
name,
TrackKind::Audio,
MediaStreamTrack::Audio(rtc_track),
),
capture_options: Mutex::new(capture_options),
}),
inner: Arc::new(TrackInner::new(
"unknown".to_string().into(), // sid
name,
TrackKind::Audio,
MediaStreamTrack::Audio(rtc_track),
)),
source,
}
}
#[inline]
pub fn capture_options(&self) -> AudioCaptureOptions {
self.inner.capture_options.lock().clone()
}
#[inline]
pub fn sid(&self) -> TrackSid {
self.inner.track_inner.sid()
self.inner.sid()
}
#[inline]
pub fn name(&self) -> String {
self.inner.track_inner.name()
self.inner.name()
}
#[inline]
pub fn kind(&self) -> TrackKind {
self.inner.track_inner.kind()
self.inner.kind()
}
#[inline]
pub fn source(&self) -> TrackSource {
self.inner.track_inner.source()
self.inner.source()
}
#[inline]
pub fn stream_state(&self) -> StreamState {
self.inner.track_inner.stream_state()
self.inner.stream_state()
}
#[inline]
pub fn start(&self) {
self.inner.track_inner.start()
pub fn enable(&self) {
self.inner.enable()
}
#[inline]
pub fn stop(&self) {
self.inner.track_inner.stop()
pub fn disable(&self) {
self.inner.disable()
}
#[inline]
pub fn is_muted(&self) -> bool {
self.inner.track_inner.is_muted()
self.inner.is_muted()
}
#[inline]
pub fn set_muted(&self, muted: bool) {
self.inner.track_inner.set_muted(muted)
pub fn mute(&self) {
self.inner.set_muted(true);
}
#[inline]
pub fn unmute(&self) {
self.inner.set_muted(false);
}
#[inline]
pub fn rtc_track(&self) -> RtcAudioTrack {
if let MediaStreamTrack::Audio(audio) = self.inner.track_inner.rtc_track() {
if let MediaStreamTrack::Audio(audio) = self.inner.rtc_track() {
return audio;
}
unreachable!()
}
#[inline]
pub fn rtc_source(&self) -> RtcAudioSource {
self.source.clone()
}
#[inline]
pub fn register_observer(&self) -> mpsc::UnboundedReceiver<TrackEvent> {
self.inner.track_inner.register_observer()
self.inner.register_observer()
}
#[inline]
@@ -120,30 +112,33 @@ impl LocalAudioTrack {
#[inline]
pub(crate) fn transceiver(&self) -> Option<RtpTransceiver> {
self.inner.track_inner.transceiver()
self.inner.transceiver()
}
#[inline]
pub(crate) fn update_transceiver(&self, transceiver: Option<RtpTransceiver>) {
self.inner.track_inner.update_transceiver(transceiver)
self.inner.update_transceiver(transceiver)
}
#[inline]
pub(crate) fn update_info(&self, info: proto::TrackInfo) {
self.inner.track_inner.update_info(info)
self.inner.update_info(info)
}
}
impl LocalAudioTrack {
pub fn create_audio_track(
name: &str,
options: AudioCaptureOptions,
source: livekit_webrtc::audio_source::native::NativeAudioSource,
) -> LocalAudioTrack {
let rtc_track = LkRuntime::instance()
.pc_factory()
.create_audio_track(&livekit_webrtc::native::create_random_uuid(), source);
Self::new(name.to_string(), rtc_track, options)
pub fn create_audio_track(name: &str, source: RtcAudioSource) -> LocalAudioTrack {
let rtc_track = match source.clone() {
#[cfg(not(target_arch = "wasm32"))]
RtcAudioSource::Native(native_source) => {
use livekit_webrtc::peer_connection_factory::native::PeerConnectionFactoryExt;
LkRuntime::instance().pc_factory().create_audio_track(
&livekit_webrtc::native::create_random_uuid(),
native_source,
)
}
_ => panic!("unsupported audio source"),
};
Self::new(name.to_string(), rtc_track, source)
}
}
livekit/src/room/track/local_track.rs
+33
View File
@@ -0,0 +1,33 @@
use super::TrackInner;
use super::{track_dispatch, LocalAudioTrack, LocalVideoTrack};
use crate::prelude::*;
use crate::track::TrackEvent;
use livekit_protocol as proto;
use livekit_protocol::enum_dispatch;
use livekit_webrtc::prelude::*;
use std::sync::Arc;
use tokio::sync::mpsc;
#[derive(Clone, Debug)]
pub enum LocalTrack {
Audio(LocalAudioTrack),
Video(LocalVideoTrack),
}
impl LocalTrack {
track_dispatch!([Audio, Video]);
enum_dispatch!(
[Audio, Video];
pub fn mute(self: &Self) -> ();
pub fn unmute(self: &Self) -> ();
);
#[inline]
pub fn rtc_track(&self) -> MediaStreamTrack {
match self {
Self::Audio(track) => track.rtc_track().into(),
Self::Video(track) => track.rtc_track().into(),
}
}
}
livekit/src/room/track/local_video_track.rs
+51 -55
View File
@@ -1,23 +1,16 @@
use super::TrackInner;
use crate::prelude::*;
use crate::rtc_engine::lk_runtime::LkRuntime;
use crate::{options::VideoCaptureOptions, prelude::*};
use livekit_protocol as proto;
use livekit_webrtc::peer_connection_factory::native::PeerConnectionFactoryExt;
use livekit_webrtc::prelude::*;
use parking_lot::Mutex;
use std::fmt::Debug;
use std::sync::Arc;
use tokio::sync::mpsc;
#[derive(Debug)]
struct LocalVideoTrackInner {
track_inner: TrackInner,
capture_options: Mutex<VideoCaptureOptions>,
}
#[derive(Clone)]
pub struct LocalVideoTrack {
inner: Arc<LocalVideoTrackInner>,
inner: Arc<TrackInner>,
source: RtcVideoSource,
}
impl Debug for LocalVideoTrack {
@@ -31,85 +24,84 @@ impl Debug for LocalVideoTrack {
}
impl LocalVideoTrack {
pub fn new(
name: String,
rtc_track: RtcVideoTrack,
capture_options: VideoCaptureOptions,
) -> Self {
pub fn new(name: String, rtc_track: RtcVideoTrack, source: RtcVideoSource) -> Self {
Self {
inner: Arc::new(LocalVideoTrackInner {
track_inner: TrackInner::new(
"unknown".to_string().into(), // sid
name,
TrackKind::Video,
MediaStreamTrack::Video(rtc_track),
),
capture_options: Mutex::new(capture_options),
}),
inner: Arc::new(TrackInner::new(
"unknown".to_string().into(), // sid
name,
TrackKind::Video,
MediaStreamTrack::Video(rtc_track),
)),
source,
}
}
#[inline]
pub fn capture_options(&self) -> VideoCaptureOptions {
self.inner.capture_options.lock().clone()
}
#[inline]
pub fn sid(&self) -> TrackSid {
self.inner.track_inner.sid()
self.inner.sid()
}
#[inline]
pub fn name(&self) -> String {
self.inner.track_inner.name()
self.inner.name()
}
#[inline]
pub fn kind(&self) -> TrackKind {
self.inner.track_inner.kind()
self.inner.kind()
}
#[inline]
pub fn source(&self) -> TrackSource {
self.inner.track_inner.source()
self.inner.source()
}
#[inline]
pub fn stream_state(&self) -> StreamState {
self.inner.track_inner.stream_state()
self.inner.stream_state()
}
#[inline]
pub fn start(&self) {
self.inner.track_inner.start()
pub fn enable(&self) {
self.inner.enable()
}
#[inline]
pub fn stop(&self) {
self.inner.track_inner.stop()
pub fn disable(&self) {
self.inner.disable()
}
#[inline]
pub fn is_muted(&self) -> bool {
self.inner.track_inner.is_muted()
self.inner.is_muted()
}
#[inline]
pub fn set_muted(&self, muted: bool) {
self.inner.track_inner.set_muted(muted)
pub fn mute(&self) {
self.inner.set_muted(true);
}
#[inline]
pub fn unmute(&self) {
self.inner.set_muted(false);
}
#[inline]
pub fn rtc_track(&self) -> RtcVideoTrack {
if let MediaStreamTrack::Video(video) = self.inner.track_inner.rtc_track() {
if let MediaStreamTrack::Video(video) = self.inner.rtc_track() {
return video;
}
unreachable!()
}
#[inline]
pub fn rtc_source(&self) -> RtcVideoSource {
self.source.clone()
}
#[inline]
pub fn register_observer(&self) -> mpsc::UnboundedReceiver<TrackEvent> {
self.inner.track_inner.register_observer()
self.inner.register_observer()
}
#[inline]
@@ -119,30 +111,34 @@ impl LocalVideoTrack {
#[inline]
pub(crate) fn transceiver(&self) -> Option<RtpTransceiver> {
self.inner.track_inner.transceiver()
self.inner.transceiver()
}
#[inline]
pub(crate) fn update_transceiver(&self, transceiver: Option<RtpTransceiver>) {
self.inner.track_inner.update_transceiver(transceiver)
self.inner.update_transceiver(transceiver)
}
#[inline]
pub(crate) fn update_info(&self, info: proto::TrackInfo) {
self.inner.track_inner.update_info(info)
self.inner.update_info(info)
}
}
impl LocalVideoTrack {
pub fn create_video_track(
name: &str,
options: VideoCaptureOptions,
source: livekit_webrtc::video_source::native::NativeVideoSource,
) -> LocalVideoTrack {
let rtc_track = LkRuntime::instance()
.pc_factory()
.create_video_track(&livekit_webrtc::native::create_random_uuid(), source);
pub fn create_video_track(name: &str, source: RtcVideoSource) -> LocalVideoTrack {
let rtc_track = match source.clone() {
#[cfg(not(target_arch = "wasm32"))]
RtcVideoSource::Native(native_source) => {
use livekit_webrtc::peer_connection_factory::native::PeerConnectionFactoryExt;
LkRuntime::instance().pc_factory().create_video_track(
&livekit_webrtc::native::create_random_uuid(),
native_source,
)
}
_ => panic!("unsupported video source"),
};
Self::new(name.to_string(), rtc_track, options)
Self::new(name.to_string(), rtc_track, source)
}
}
livekit/src/room/track/mod.rs
+72 -144
View File
@@ -3,19 +3,22 @@ use livekit_protocol as proto;
use livekit_protocol::enum_dispatch;
use livekit_protocol::observer::Dispatcher;
use livekit_webrtc::prelude::*;
use parking_lot::Mutex;
use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
use parking_lot::RwLock;
use thiserror::Error;
use tokio::sync::mpsc;
mod local_audio_track;
mod local_track;
mod local_video_track;
mod remote_audio_track;
mod remote_track;
mod remote_video_track;
pub use local_audio_track::*;
pub use local_track::*;
pub use local_video_track::*;
pub use remote_audio_track::*;
pub use remote_track::*;
pub use remote_video_track::*;
#[derive(Error, Debug, Clone)]
@@ -45,10 +48,10 @@ pub enum TrackSource {
ScreenshareAudio,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[derive(Debug, Clone)]
pub enum TrackEvent {
Mute,
Unmute,
Muted,
Unmuted,
}
#[derive(Clone, Copy, Debug)]
@@ -62,18 +65,6 @@ pub enum Track {
RemoteVideo(RemoteVideoTrack),
}
#[derive(Clone, Debug)]
pub enum LocalTrack {
Audio(LocalAudioTrack),
Video(LocalVideoTrack),
}
#[derive(Clone, Debug)]
pub enum RemoteTrack {
Audio(RemoteAudioTrack),
Video(RemoteVideoTrack),
}
#[derive(Clone, Debug)]
pub enum VideoTrack {
Local(LocalVideoTrack),
@@ -95,12 +86,11 @@ macro_rules! track_dispatch {
pub fn kind(self: &Self) -> TrackKind;
pub fn source(self: &Self) -> TrackSource;
pub fn stream_state(self: &Self) -> StreamState;
pub fn start(self: &Self) -> ();
pub fn stop(self: &Self) -> ();
pub fn enable(self: &Self) -> ();
pub fn disable(self: &Self) -> ();
pub fn is_muted(self: &Self) -> bool;
pub fn set_muted(self: &Self, muted: bool) -> ();
pub fn register_observer(self: &Self) -> mpsc::UnboundedReceiver<TrackEvent>;
pub fn is_remote(self: &Self) -> bool;
pub fn register_observer(self: &Self) -> mpsc::UnboundedReceiver<TrackEvent>;
pub(crate) fn transceiver(self: &Self) -> Option<RtpTransceiver>;
pub(crate) fn update_transceiver(self: &Self, transceiver: Option<RtpTransceiver>) -> ();
@@ -109,6 +99,8 @@ macro_rules! track_dispatch {
};
}
pub(crate) use track_dispatch;
impl Track {
track_dispatch!([LocalAudio, LocalVideo, RemoteAudio, RemoteVideo]);
@@ -123,30 +115,6 @@ impl Track {
}
}
impl LocalTrack {
track_dispatch!([Audio, Video]);
#[inline]
pub fn rtc_track(&self) -> MediaStreamTrack {
match self {
Self::Audio(track) => track.rtc_track().into(),
Self::Video(track) => track.rtc_track().into(),
}
}
}
impl RemoteTrack {
track_dispatch!([Audio, Video]);
#[inline]
pub fn rtc_track(&self) -> MediaStreamTrack {
match self {
Self::Audio(track) => track.rtc_track().into(),
Self::Video(track) => track.rtc_track().into(),
}
}
}
impl VideoTrack {
track_dispatch!([Local, Remote]);
@@ -171,90 +139,73 @@ impl AudioTrack {
}
}
#[derive(Debug)]
struct TrackInfo {
sid: TrackSid,
name: String,
kind: TrackKind,
source: TrackSource,
stream_state: StreamState,
muted: bool,
transceiver: Option<RtpTransceiver>,
}
#[derive(Debug)]
pub(crate) struct TrackInner {
pub sid: Mutex<TrackSid>,
pub name: Mutex<String>,
pub kind: AtomicU8, // TrackKind
pub source: AtomicU8, // TrackSource
pub stream_state: AtomicU8, // StreamState
pub muted: AtomicBool,
pub rtc_track: MediaStreamTrack,
pub transceiver: Mutex<Option<RtpTransceiver>>,
pub dispatcher: Dispatcher<TrackEvent>,
info: RwLock<TrackInfo>,
rtc_track: MediaStreamTrack,
dispatcher: Dispatcher<TrackEvent>,
}
impl TrackInner {
pub fn new(sid: TrackSid, name: String, kind: TrackKind, rtc_track: MediaStreamTrack) -> Self {
Self {
sid: Mutex::new(sid),
name: Mutex::new(name),
kind: AtomicU8::new(kind as u8),
source: AtomicU8::new(TrackSource::Unknown as u8),
stream_state: AtomicU8::new(StreamState::Active as u8),
muted: AtomicBool::new(false),
info: RwLock::new(TrackInfo {
sid,
name,
kind,
source: TrackSource::Unknown,
stream_state: StreamState::Active,
muted: false,
transceiver: None,
}),
rtc_track,
transceiver: Default::default(),
dispatcher: Default::default(),
}
}
pub fn sid(&self) -> TrackSid {
self.sid.lock().clone()
self.info.read().sid.clone()
}
pub fn name(&self) -> String {
self.name.lock().clone()
self.info.read().name.clone()
}
pub fn kind(&self) -> TrackKind {
self.kind.load(Ordering::SeqCst).try_into().unwrap()
self.info.read().kind
}
pub fn source(&self) -> TrackSource {
self.source.load(Ordering::SeqCst).into()
self.info.read().source
}
pub fn stream_state(&self) -> StreamState {
self.stream_state.load(Ordering::SeqCst).try_into().unwrap()
self.info.read().stream_state
}
pub fn is_muted(&self) -> bool {
self.muted.load(Ordering::SeqCst)
self.info.read().muted
}
pub fn start(&self) {
pub fn enable(&self) {
self.rtc_track.set_enabled(true);
}
pub fn stop(&self) {
pub fn disable(&self) {
self.rtc_track.set_enabled(false);
}
pub fn set_muted(&self, muted: bool) {
if self
.muted
.compare_exchange(!muted, muted, Ordering::SeqCst, Ordering::SeqCst)
.is_err()
{
return;
}
if !muted {
self.start();
} else {
self.stop();
}
let event = if muted {
TrackEvent::Mute
} else {
TrackEvent::Unmute
};
self.dispatcher.dispatch(&event);
}
pub fn rtc_track(&self) -> MediaStreamTrack {
self.rtc_track.clone()
}
@@ -264,24 +215,39 @@ impl TrackInner {
}
pub fn transceiver(&self) -> Option<RtpTransceiver> {
self.transceiver.lock().clone()
self.info.read().transceiver.clone()
}
pub fn update_transceiver(&self, transceiver: Option<RtpTransceiver>) {
*self.transceiver.lock() = transceiver;
self.info.write().transceiver = transceiver;
}
pub fn update_info(&self, info: proto::TrackInfo) {
*self.name.lock() = info.name;
*self.sid.lock() = info.sid.into();
self.kind.store(
TrackKind::try_from(proto::TrackType::from_i32(info.r#type).unwrap()).unwrap() as u8,
Ordering::SeqCst,
);
self.source.store(
TrackSource::from(proto::TrackSource::from_i32(info.source).unwrap()) as u8,
Ordering::SeqCst,
);
pub fn set_muted(&self, muted: bool) {
log::debug!("set_muted: {} {}", self.sid(), muted);
if self.is_muted() == muted {
return;
}
if muted {
self.disable();
} else {
self.enable();
}
self.dispatcher.dispatch(if muted {
&TrackEvent::Muted
} else {
&TrackEvent::Unmuted
});
}
pub fn update_info(&self, new_info: proto::TrackInfo) {
let mut info = self.info.write();
info.name = new_info.name;
info.sid = new_info.sid.into();
info.kind =
TrackKind::try_from(proto::TrackType::from_i32(new_info.r#type).unwrap()).unwrap();
info.source = TrackSource::from(proto::TrackSource::from_i32(new_info.source).unwrap());
// Muted and StreamState are not handled separately (events)
}
}
@@ -370,44 +336,6 @@ impl TryFrom<Track> for AudioTrack {
}
}
// Conversions from integers (Useful since we're using atomic values to represent our enums)
impl TryFrom<u8> for TrackKind {
type Error = &'static str;
fn try_from(kind: u8) -> Result<Self, Self::Error> {
match kind {
0 => Ok(Self::Audio),
1 => Ok(Self::Video),
_ => Err("invalid track kind"),
}
}
}
impl TryFrom<u8> for StreamState {
type Error = &'static str;
fn try_from(state: u8) -> Result<Self, Self::Error> {
match state {
0 => Ok(Self::Active),
1 => Ok(Self::Paused),
_ => Err("invalid stream state"),
}
}
}
impl From<u8> for TrackSource {
fn from(source: u8) -> Self {
match source {
1 => Self::Camera,
2 => Self::Microphone,
3 => Self::Screenshare,
4 => Self::ScreenshareAudio,
_ => Self::Unknown,
}
}
}
impl From<TrackKind> for MediaType {
fn from(kind: TrackKind) -> Self {
match kind {
livekit/src/room/track/remote_audio_track.rs
+6 -10
View File
@@ -1,3 +1,4 @@
use super::remote_track;
use super::TrackInner;
use crate::prelude::*;
use livekit_protocol as proto;
@@ -59,13 +60,13 @@ impl RemoteAudioTrack {
}
#[inline]
pub fn start(&self) {
self.inner.start()
pub fn enable(&self) {
self.inner.enable()
}
#[inline]
pub fn stop(&self) {
self.inner.stop()
pub fn disable(&self) {
self.inner.disable()
}
#[inline]
@@ -73,11 +74,6 @@ impl RemoteAudioTrack {
self.inner.is_muted()
}
#[inline]
pub fn set_muted(&self, muted: bool) {
self.inner.set_muted(muted)
}
#[inline]
pub fn rtc_track(&self) -> RtcAudioTrack {
if let MediaStreamTrack::Audio(audio) = self.inner.rtc_track() {
@@ -110,6 +106,6 @@ impl RemoteAudioTrack {
#[inline]
pub(crate) fn update_info(&self, info: proto::TrackInfo) {
self.inner.update_info(info)
remote_track::update_info(&self.inner, info);
}
}
livekit/src/room/track/remote_track.rs
+33
View File
@@ -0,0 +1,33 @@
use super::track_dispatch;
use super::TrackInner;
use super::{RemoteAudioTrack, RemoteVideoTrack};
use crate::prelude::*;
use crate::track::TrackEvent;
use livekit_protocol as proto;
use livekit_protocol::enum_dispatch;
use livekit_webrtc::prelude::*;
use std::sync::Arc;
use tokio::sync::mpsc;
#[derive(Clone, Debug)]
pub enum RemoteTrack {
Audio(RemoteAudioTrack),
Video(RemoteVideoTrack),
}
impl RemoteTrack {
track_dispatch!([Audio, Video]);
#[inline]
pub fn rtc_track(&self) -> MediaStreamTrack {
match self {
Self::Audio(track) => track.rtc_track().into(),
Self::Video(track) => track.rtc_track().into(),
}
}
}
pub(crate) fn update_info(track: &Arc<TrackInner>, new_info: proto::TrackInfo) {
track.update_info(new_info.clone());
track.set_muted(new_info.muted);
}
livekit/src/room/track/remote_video_track.rs
+6 -11
View File
@@ -1,4 +1,4 @@
use super::TrackInner;
use super::{remote_track, TrackInner};
use crate::prelude::*;
use livekit_protocol as proto;
use livekit_webrtc::prelude::*;
@@ -59,13 +59,13 @@ impl RemoteVideoTrack {
}
#[inline]
pub fn start(&self) {
self.inner.start()
pub fn enable(&self) {
self.inner.enable()
}
#[inline]
pub fn stop(&self) {
self.inner.stop()
pub fn disable(&self) {
self.inner.disable()
}
#[inline]
@@ -73,11 +73,6 @@ impl RemoteVideoTrack {
self.inner.is_muted()
}
#[inline]
pub fn set_muted(&self, muted: bool) {
self.inner.set_muted(muted)
}
#[inline]
pub fn rtc_track(&self) -> RtcVideoTrack {
if let MediaStreamTrack::Video(video) = self.inner.rtc_track() {
@@ -110,6 +105,6 @@ impl RemoteVideoTrack {
#[inline]
pub(crate) fn update_info(&self, info: proto::TrackInfo) {
self.inner.update_info(info);
remote_track::update_info(&self.inner, info);
}
}