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

use callbacks on internal events (#100)

Browse Source
This commit is contained in:
Théo Monnom
2023-06-29 22:52:43 +02:00
committed by GitHub
parent efc334af8e
commit c7c82cc693
33 changed files with 1522 additions and 1447 deletions
Download Patch File Download Diff File Expand all files Collapse all files
livekit/src/lib.rs
-2
View File
@@ -1,5 +1,3 @@
extern crate core;
pub mod proto;
mod room;
mod rtc_engine;
livekit/src/prelude.rs
+2 -2
View File
@@ -1,4 +1,4 @@
pub use crate::participant::{LocalParticipant, Participant, ParticipantEvent, RemoteParticipant};
pub use crate::participant::{LocalParticipant, Participant, RemoteParticipant};
pub use crate::{
ConnectionState, DataPacketKind, Room, RoomError, RoomEvent, RoomOptions, RoomResult,
@@ -8,7 +8,7 @@ pub use crate::publication::{LocalTrackPublication, RemoteTrackPublication, Trac
pub use crate::track::{
AudioTrack, LocalAudioTrack, LocalTrack, LocalVideoTrack, RemoteAudioTrack, RemoteTrack,
RemoteVideoTrack, StreamState, Track, TrackEvent, TrackKind, TrackSource, VideoTrack,
RemoteVideoTrack, StreamState, Track, TrackDimension, TrackKind, TrackSource, VideoTrack,
};
pub use crate::id::*;
livekit/src/proto.rs
+11 -1
View File
@@ -1,7 +1,17 @@
use crate::{track, DataPacketKind};
use crate::{participant, track, DataPacketKind};
use livekit_protocol::*;
// Conversions
impl From<ConnectionQuality> for participant::ConnectionQuality {
fn from(value: ConnectionQuality) -> Self {
match value {
ConnectionQuality::Excellent => Self::Excellent,
ConnectionQuality::Good => Self::Good,
ConnectionQuality::Poor => Self::Poor,
}
}
}
impl TryFrom<TrackType> for track::TrackKind {
type Error = &'static str;
livekit/src/room/mod.rs
+74 -111
View File
@@ -6,11 +6,12 @@ use crate::rtc_engine::{EngineEvent, EngineEvents, EngineResult, RtcEngine};
use livekit_api::signal_client::SignalOptions;
use livekit_protocol as proto;
use livekit_protocol::observer::Dispatcher;
use parking_lot::{Mutex, RwLock, RwLockReadGuard};
use parking_lot::RwLock;
use std::collections::HashMap;
use std::fmt::Debug;
use std::sync::Arc;
use thiserror::Error;
use tokio::sync::Mutex as AsyncMutex;
use tokio::sync::{mpsc, oneshot};
use tokio::task::JoinHandle;
@@ -37,14 +38,26 @@ pub enum RoomError {
}
#[derive(Clone, Debug)]
#[non_exhaustive]
pub enum RoomEvent {
ParticipantConnected(RemoteParticipant),
ParticipantDisconnected(RemoteParticipant),
LocalTrackPublished {
publication: LocalTrackPublication,
},
LocalTrackUnpublished {
publication: LocalTrackPublication,
},
TrackSubscribed {
track: RemoteTrack,
publication: RemoteTrackPublication,
participant: RemoteParticipant,
},
TrackUnsubscribed {
track: RemoteTrack,
publication: RemoteTrackPublication,
participant: RemoteParticipant,
},
TrackPublished {
publication: RemoteTrackPublication,
participant: RemoteParticipant,
@@ -53,15 +66,10 @@ pub enum RoomEvent {
publication: RemoteTrackPublication,
participant: RemoteParticipant,
},
TrackUnsubscribed {
track: RemoteTrack,
publication: RemoteTrackPublication,
participant: RemoteParticipant,
},
TrackSubscriptionFailed {
participant: RemoteParticipant,
error: track::TrackError,
sid: TrackSid,
participant: RemoteParticipant,
},
TrackMuted {
participant: Participant,
@@ -128,7 +136,7 @@ struct RoomHandle {
pub struct Room {
inner: Arc<RoomSession>,
handle: Mutex<Option<RoomHandle>>,
handle: AsyncMutex<Option<RoomHandle>>,
}
impl Debug for Room {
@@ -178,7 +186,6 @@ impl Room {
metadata: room_info.metadata,
}),
participants: Default::default(),
participants_tasks: Default::default(),
active_speakers: Default::default(),
rtc_engine,
local_participant,
@@ -200,7 +207,7 @@ impl Room {
let session = Self {
inner,
handle: Mutex::new(Some(RoomHandle {
handle: AsyncMutex::new(Some(RoomHandle {
session_task,
close_emitter,
})),
@@ -211,10 +218,10 @@ impl Room {
}
pub async fn close(&self) -> RoomResult<()> {
if let Some(handle) = self.handle.lock().take() {
if let Some(handle) = self.handle.lock().await.take() {
self.inner.close().await;
handle.close_emitter.send(()).ok();
handle.session_task.await.ok();
let _ = handle.close_emitter.send(());
let _ = handle.session_task.await;
Ok(())
} else {
Err(RoomError::AlreadyClosed)
@@ -245,8 +252,8 @@ impl Room {
self.inner.info.read().state
}
pub fn participants(&self) -> RwLockReadGuard<HashMap<ParticipantSid, RemoteParticipant>> {
self.inner.participants.read()
pub fn participants(&self) -> HashMap<ParticipantSid, RemoteParticipant> {
self.inner.participants.read().clone()
}
pub async fn simulate_scenario(&self, scenario: SimulateScenario) -> EngineResult<()> {
@@ -268,7 +275,6 @@ pub(crate) struct RoomSession {
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 {
@@ -304,70 +310,6 @@ impl RoomSession {
}
}
/// Forward participant events to the room dispatcher
async fn participant_task(
self: Arc<Self>,
participant: Participant,
mut participant_events: mpsc::UnboundedReceiver<ParticipantEvent>,
mut close_rx: oneshot::Receiver<()>,
) {
loop {
tokio::select! {
res = participant_events.recv() => {
if let Some(event) = res {
if let Err(err) = self.on_participant_event(&participant, event).await {
log::error!("failed to handle participant event for {:?}: {:?}", participant.sid(), err);
}
}
},
_ = &mut close_rx => {
log::trace!("closing participant_task for {:?}", participant.sid());
break;
},
}
}
}
async fn on_participant_event(
self: &Arc<Self>,
participant: &Participant,
event: ParticipantEvent,
) -> RoomResult<()> {
if let Participant::Remote(remote_participant) = participant {
match event {
ParticipantEvent::TrackPublished { publication } => {
self.dispatcher.dispatch(&RoomEvent::TrackPublished {
participant: remote_participant.clone(),
publication,
});
}
ParticipantEvent::TrackUnpublished { publication } => {
self.dispatcher.dispatch(&RoomEvent::TrackUnpublished {
participant: remote_participant.clone(),
publication,
});
}
ParticipantEvent::TrackSubscribed { track, publication } => {
self.dispatcher.dispatch(&RoomEvent::TrackSubscribed {
participant: remote_participant.clone(),
track,
publication,
});
}
ParticipantEvent::TrackUnsubscribed { track, publication } => {
self.dispatcher.dispatch(&RoomEvent::TrackUnsubscribed {
participant: remote_participant.clone(),
track,
publication,
});
}
_ => {}
};
}
Ok(())
}
async fn on_engine_event(self: &Arc<Self>, event: EngineEvent) -> RoomResult<()> {
match event {
EngineEvent::ParticipantUpdate { updates } => self.handle_participant_update(updates),
@@ -432,7 +374,7 @@ impl RoomSession {
participant: participant.clone(),
});
participant.on_data_received(payload, kind);
//participant.on_data_received(payload, kind);
}
}
EngineEvent::SpeakersChanged { speakers } => self.handle_speakers_changed(speakers),
@@ -570,7 +512,8 @@ impl RoomSession {
fn handle_restarting(self: &Arc<Self>) {
// Remove existing participants/subscriptions on full reconnect
for (_, participant) in self.participants.read().iter() {
let participants = self.participants.read().clone();
for (_, participant) in participants.iter() {
self.clone()
.handle_participant_disconnect(participant.clone());
}
@@ -621,44 +564,64 @@ impl RoomSession {
metadata,
);
// Create the participant task
let (close_tx, close_rx) = oneshot::channel();
let participant_task = tokio::spawn(self.clone().participant_task(
Participant::Remote(participant.clone()),
participant.register_observer(),
close_rx,
));
self.participants_tasks
.write()
.insert(sid.clone(), (participant_task, close_tx));
let dispatcher = self.dispatcher.clone();
participant.on_track_published(move |participant, publication| {
dispatcher.dispatch(&RoomEvent::TrackPublished {
participant,
publication,
});
});
let dispatcher = self.dispatcher.clone();
participant.on_track_unpublished(move |participant, publication| {
dispatcher.dispatch(&RoomEvent::TrackUnpublished {
participant,
publication,
});
});
let dispatcher = self.dispatcher.clone();
participant.on_track_subscribed(move |participant, publication, track| {
dispatcher.dispatch(&RoomEvent::TrackSubscribed {
participant,
track,
publication,
});
});
let dispatcher = self.dispatcher.clone();
participant.on_track_unsubscribed(move |participant, publication, track| {
dispatcher.dispatch(&RoomEvent::TrackUnsubscribed {
participant,
track,
publication,
});
});
let dispatcher = self.dispatcher.clone();
participant.on_track_subscription_failed(move |participant, sid, error| {
dispatcher.dispatch(&RoomEvent::TrackSubscriptionFailed {
participant,
sid,
error,
});
});
self.participants.write().insert(sid, participant.clone());
participant
}
/// A participant has disconnected
/// Cleanup the participant and emit an event
fn handle_participant_disconnect(self: Arc<Self>, remote_participant: RemoteParticipant) {
tokio::spawn(async move {
for (sid, _) in &*remote_participant.tracks() {
remote_participant.unpublish_track(&sid);
}
for (sid, _) in remote_participant.tracks() {
remote_participant.unpublish_track(&sid);
}
// Close the participant task
let ptask = self
.participants_tasks
.write()
.remove(&remote_participant.sid());
if let Some((task, close_tx)) = ptask {
let _ = close_tx.send(());
let _ = task.await;
}
self.participants.write().remove(&remote_participant.sid());
self.dispatcher
.dispatch(&RoomEvent::ParticipantDisconnected(remote_participant));
});
self.participants.write().remove(&remote_participant.sid());
self.dispatcher
.dispatch(&RoomEvent::ParticipantDisconnected(remote_participant));
}
fn get_participant(&self, sid: &ParticipantSid) -> Option<RemoteParticipant> {
livekit/src/room/participant/local_participant.rs
+87 -73
View File
@@ -1,5 +1,5 @@
use super::ConnectionQuality;
use super::ParticipantInternal;
use super::ParticipantInner;
use crate::options;
use crate::options::compute_video_encodings;
use crate::options::video_layers_from_encodings;
@@ -9,15 +9,27 @@ use crate::rtc_engine::RtcEngine;
use crate::DataPacketKind;
use livekit_protocol as proto;
use livekit_webrtc::rtp_parameters::RtpEncodingParameters;
use parking_lot::RwLockReadGuard;
use parking_lot::Mutex;
use std::collections::HashMap;
use std::fmt::Debug;
use std::sync::Arc;
use tokio::sync::mpsc;
#[derive(Default)]
struct LocalEvents {
local_track_published:
Mutex<Option<Box<dyn Fn(LocalParticipant, LocalTrackPublication) + Send>>>,
local_track_unpublished:
Mutex<Option<Box<dyn Fn(LocalParticipant, LocalTrackPublication) + Send>>>,
}
struct LocalInfo {
events: LocalEvents,
}
#[derive(Clone)]
pub struct LocalParticipant {
inner: Arc<ParticipantInternal>,
inner: Arc<ParticipantInner>,
local: Arc<LocalInfo>,
}
impl Debug for LocalParticipant {
@@ -39,12 +51,54 @@ impl LocalParticipant {
metadata: String,
) -> Self {
Self {
inner: Arc::new(ParticipantInternal::new(
rtc_engine, sid, identity, name, metadata,
)),
inner: super::new_inner(rtc_engine, sid, identity, name, metadata),
local: Arc::new(LocalInfo {
events: LocalEvents::default(),
}),
}
}
pub(crate) fn update_info(self: &Self, info: proto::ParticipantInfo) {
super::update_info(&self.inner, &Participant::Local(self.clone()), info);
}
pub(crate) fn set_speaking(&self, speaking: bool) {
super::set_speaking(&self.inner, &Participant::Local(self.clone()), speaking);
}
pub(crate) fn set_audio_level(&self, level: f32) {
super::set_audio_level(&self.inner, &Participant::Local(self.clone()), level);
}
pub(crate) fn set_connection_quality(&self, quality: ConnectionQuality) {
super::set_connection_quality(&self.inner, &Participant::Local(self.clone()), quality);
}
#[allow(dead_code)]
pub(crate) fn on_local_track_published(
&self,
handler: impl Fn(LocalParticipant, LocalTrackPublication) + Send + 'static,
) {
*self.local.events.local_track_published.lock() = Some(Box::new(handler));
}
#[allow(dead_code)]
pub(crate) fn on_local_track_unpublished(
&self,
handler: impl Fn(LocalParticipant, LocalTrackPublication) + Send + 'static,
) {
*self.local.events.local_track_unpublished.lock() = Some(Box::new(handler));
}
pub(crate) fn add_publication(&self, publication: TrackPublication) {
super::add_publication(&self.inner, &Participant::Local(self.clone()), publication);
}
#[allow(dead_code)]
pub(crate) fn remove_publication(&self, sid: &TrackSid) {
super::remove_publication(&self.inner, &Participant::Local(self.clone()), sid);
}
pub async fn publish_track(
&self,
track: LocalTrack,
@@ -87,11 +141,7 @@ 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(),
);
let publication = LocalTrackPublication::new(track_info.clone(), track.clone());
track.update_info(track_info); // Update sid + source
log::debug!("publishing track with cid {:?}", track.rtc_track().id());
@@ -101,8 +151,7 @@ impl LocalParticipant {
.create_sender(track.clone(), options, encodings)
.await?;
track.update_transceiver(Some(transceiver));
//track.start();
track.set_transceiver(Some(transceiver));
track.enable();
tokio::spawn({
@@ -112,14 +161,12 @@ impl LocalParticipant {
}
});
self.inner
.add_publication(TrackPublication::Local(publication.clone()));
self.add_publication(TrackPublication::Local(publication.clone()));
self.inner
.dispatcher
.dispatch(&ParticipantEvent::LocalTrackPublished {
publication: publication.clone(),
});
if let Some(local_track_published) = self.local.events.local_track_published.lock().as_ref()
{
local_track_published(self.clone(), publication.clone());
}
Ok(publication)
}
@@ -129,20 +176,21 @@ impl LocalParticipant {
track: TrackSid,
_stop_on_unpublish: bool,
) -> RoomResult<LocalTrackPublication> {
let mut tracks = self.inner.tracks.write();
if let Some(TrackPublication::Local(publication)) = tracks.remove(&track) {
let publication = self.inner.tracks.write().remove(&track);
if let Some(TrackPublication::Local(publication)) = publication {
let track = publication.track();
let sender = track.transceiver().unwrap().sender();
self.inner.rtc_engine.remove_track(sender).await?;
track.update_transceiver(None);
track.set_transceiver(None);
self.inner
.dispatcher
.dispatch(&ParticipantEvent::LocalTrackUnpublished {
publication: publication.clone(),
});
// publication.update_track(None);
if let Some(local_track_unpublished) =
self.local.events.local_track_unpublished.lock().as_ref()
{
local_track_unpublished(self.clone(), publication.clone());
}
publication.set_track(None);
tokio::spawn({
let rtc_engine = self.inner.rtc_engine.clone();
@@ -179,7 +227,6 @@ impl LocalParticipant {
.map_err(Into::into)
}
#[inline]
pub fn get_track_publication(&self, sid: &TrackSid) -> Option<LocalTrackPublication> {
self.inner.tracks.read().get(sid).map(|track| {
if let TrackPublication::Local(local) = track {
@@ -190,68 +237,35 @@ impl LocalParticipant {
})
}
#[inline]
pub fn sid(&self) -> ParticipantSid {
self.inner.sid()
self.inner.info.read().sid.clone()
}
#[inline]
pub fn identity(&self) -> ParticipantIdentity {
self.inner.identity()
self.inner.info.read().identity.clone()
}
#[inline]
pub fn name(&self) -> String {
self.inner.name()
self.inner.info.read().name.clone()
}
#[inline]
pub fn metadata(&self) -> String {
self.inner.metadata()
self.inner.info.read().metadata.clone()
}
#[inline]
pub fn is_speaking(&self) -> bool {
self.inner.is_speaking()
self.inner.info.read().speaking
}
#[inline]
pub fn tracks(&self) -> RwLockReadGuard<HashMap<TrackSid, TrackPublication>> {
self.inner.tracks()
pub fn tracks(&self) -> HashMap<TrackSid, TrackPublication> {
self.inner.tracks.read().clone()
}
#[inline]
pub fn audio_level(&self) -> f32 {
self.inner.audio_level()
self.inner.info.read().audio_level
}
#[inline]
pub fn connection_quality(&self) -> ConnectionQuality {
self.inner.connection_quality()
}
#[inline]
pub fn register_observer(&self) -> mpsc::UnboundedReceiver<ParticipantEvent> {
self.inner.register_observer()
}
#[inline]
pub(crate) fn update_info(self: &Self, info: proto::ParticipantInfo) {
self.inner.update_info(info);
}
#[inline]
pub(crate) fn set_speaking(&self, speaking: bool) {
self.inner.set_speaking(speaking);
}
#[inline]
pub(crate) fn set_audio_level(&self, level: f32) {
self.inner.set_audio_level(level);
}
#[inline]
pub(crate) fn set_connection_quality(&self, quality: ConnectionQuality) {
self.inner.set_connection_quality(quality);
self.inner.info.read().connection_quality
}
}
livekit/src/room/participant/mod.rs
+116 -167
View File
@@ -1,15 +1,11 @@
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::{RwLock, RwLockReadGuard};
use parking_lot::{Mutex, RwLock};
use std::collections::HashMap;
use std::fmt::Debug;
use std::sync::Arc;
use std::thread::JoinHandle;
use tokio::sync::{mpsc, oneshot};
mod local_participant;
mod remote_participant;
@@ -17,52 +13,7 @@ mod remote_participant;
pub use local_participant::*;
pub use remote_participant::*;
#[derive(Debug, Clone)]
pub enum ParticipantEvent {
TrackPublished {
publication: RemoteTrackPublication,
},
TrackUnpublished {
publication: RemoteTrackPublication,
},
TrackSubscribed {
track: RemoteTrack,
publication: RemoteTrackPublication,
},
TrackUnsubscribed {
track: RemoteTrack,
publication: RemoteTrackPublication,
},
TrackSubscriptionFailed {
error: TrackError,
sid: TrackSid,
},
DataReceived {
payload: Arc<Vec<u8>>,
kind: DataPacketKind,
},
SpeakingChanged {
speaking: bool,
},
TrackMuted {
publication: TrackPublication,
},
TrackUnmuted {
publication: TrackPublication,
},
ConnectionQualityChanged {
quality: ConnectionQuality,
},
LocalTrackPublished {
publication: LocalTrackPublication,
},
LocalTrackUnpublished {
publication: LocalTrackPublication,
},
}
#[derive(Debug, Clone, Copy, Eq, PartialEq)]
#[repr(u8)]
pub enum ConnectionQuality {
Unknown,
Excellent,
@@ -70,27 +21,6 @@ pub enum ConnectionQuality {
Poor,
}
impl From<u8> for ConnectionQuality {
fn from(value: u8) -> Self {
match value {
1 => Self::Excellent,
2 => Self::Good,
3 => Self::Poor,
_ => Self::Unknown,
}
}
}
impl From<proto::ConnectionQuality> for ConnectionQuality {
fn from(value: proto::ConnectionQuality) -> Self {
match value {
proto::ConnectionQuality::Excellent => Self::Excellent,
proto::ConnectionQuality::Good => Self::Good,
proto::ConnectionQuality::Poor => Self::Poor,
}
}
}
#[derive(Debug, Clone)]
pub enum Participant {
Local(LocalParticipant),
@@ -107,18 +37,20 @@ impl Participant {
pub fn is_speaking(self: &Self) -> bool;
pub fn audio_level(self: &Self) -> f32;
pub fn connection_quality(self: &Self) -> ConnectionQuality;
pub fn tracks(self: &Self) -> RwLockReadGuard<HashMap<TrackSid, TrackPublication>>;
pub fn register_observer(self: &Self) -> mpsc::UnboundedReceiver<ParticipantEvent>;
pub fn tracks(self: &Self) -> HashMap<TrackSid, TrackPublication>;
pub(crate) fn update_info(self: &Self, info: proto::ParticipantInfo) -> ();
// Internal functions called by the Room when receiving the associated signal messages
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) -> ();
pub(crate) fn update_info(self: &Self, info: proto::ParticipantInfo) -> ();
pub(crate) fn add_publication(self: &Self, publication: TrackPublication) -> ();
pub(crate) fn remove_publication(self: &Self, sid: &TrackSid) -> ();
);
}
#[derive(Debug)]
pub(crate) struct ParticipantInfo {
struct ParticipantInfo {
pub sid: ParticipantSid,
pub identity: ParticipantIdentity,
pub name: String,
@@ -128,101 +60,118 @@ pub(crate) struct ParticipantInfo {
pub connection_quality: ConnectionQuality,
}
#[derive(Debug)]
pub(crate) struct ParticipantInternal {
pub(super) rtc_engine: Arc<RtcEngine>,
pub(super) dispatcher: Dispatcher<ParticipantEvent>,
#[derive(Default)]
struct ParticipantEvents {
track_muted: Mutex<Option<Box<dyn Fn(Participant, TrackPublication, Track) + Send>>>,
track_unmuted: Mutex<Option<Box<dyn Fn(Participant, TrackPublication, Track) + Send>>>,
}
pub(super) struct ParticipantInner {
rtc_engine: Arc<RtcEngine>,
info: RwLock<ParticipantInfo>,
tracks: RwLock<HashMap<TrackSid, TrackPublication>>,
tracks_tasks: RwLock<HashMap<TrackSid, (JoinHandle<()>, oneshot::Sender<()>)>>,
events: Arc<ParticipantEvents>,
}
impl ParticipantInternal {
pub fn new(
rtc_engine: Arc<RtcEngine>,
sid: ParticipantSid,
identity: ParticipantIdentity,
name: String,
metadata: String,
) -> Self {
Self {
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(super) fn new_inner(
rtc_engine: Arc<RtcEngine>,
sid: ParticipantSid,
identity: ParticipantIdentity,
name: String,
metadata: String,
) -> Arc<ParticipantInner> {
Arc::new(ParticipantInner {
rtc_engine,
info: RwLock::new(ParticipantInfo {
sid,
identity,
name,
metadata,
speaking: false,
audio_level: 0.0,
connection_quality: ConnectionQuality::Unknown,
}),
tracks: Default::default(),
events: Default::default(),
})
}
pub(super) fn update_info(
inner: &Arc<ParticipantInner>,
_participant: &Participant,
new_info: proto::ParticipantInfo,
) {
let mut info = inner.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(super) fn set_speaking(
inner: &Arc<ParticipantInner>,
_participant: &Participant,
speaking: bool,
) {
inner.info.write().speaking = speaking;
}
pub(super) fn set_audio_level(
inner: &Arc<ParticipantInner>,
_participant: &Participant,
audio_level: f32,
) {
inner.info.write().audio_level = audio_level;
}
pub(super) fn set_connection_quality(
inner: &Arc<ParticipantInner>,
_participant: &Participant,
quality: ConnectionQuality,
) {
inner.info.write().connection_quality = quality;
}
pub(super) fn remove_publication(
inner: &Arc<ParticipantInner>,
_participant: &Participant,
sid: &TrackSid,
) -> Option<TrackPublication> {
let mut tracks = inner.tracks.write();
let publication = tracks.remove(sid);
if let Some(publication) = publication.clone() {
// remove events
publication.on_muted(|_, _| {});
publication.on_unmuted(|_, _| {});
} else {
// shouldn't happen (internal)
log::warn!("could not find publication to remove: {}", sid);
}
publication
}
pub(super) fn add_publication(
inner: &Arc<ParticipantInner>,
participant: &Participant,
publication: TrackPublication,
) {
let mut tracks = inner.tracks.write();
tracks.insert(publication.sid(), publication.clone());
let events = inner.events.clone();
let particiant = participant.clone();
publication.on_muted(move |publication, track| {
if let Some(cb) = events.track_muted.lock().as_ref() {
cb(particiant.clone(), publication, track);
}
}
});
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.info.read().sid.clone()
}
pub fn identity(&self) -> ParticipantIdentity {
self.info.read().identity.clone()
}
pub fn name(&self) -> String {
self.info.read().name.clone()
}
pub fn metadata(&self) -> String {
self.info.read().metadata.clone()
}
pub fn is_speaking(&self) -> bool {
self.info.read().speaking
}
pub fn tracks(&self) -> RwLockReadGuard<HashMap<TrackSid, TrackPublication>> {
self.tracks.read()
}
pub fn audio_level(&self) -> f32 {
self.info.read().audio_level
}
pub fn connection_quality(&self) -> ConnectionQuality {
self.info.read().connection_quality
}
pub fn register_observer(&self) -> mpsc::UnboundedReceiver<ParticipantEvent> {
self.dispatcher.register()
}
pub fn set_speaking(&self, speaking: bool) {
self.info.write().speaking = speaking;
}
pub fn set_audio_level(&self, audio_level: f32) {
self.info.write().audio_level = audio_level;
}
pub fn set_connection_quality(&self, quality: ConnectionQuality) {
self.info.write().connection_quality = quality;
}
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);
}
let events = inner.events.clone();
let participant = participant.clone();
publication.on_unmuted(move |publication, track| {
if let Some(cb) = events.track_unmuted.lock().as_ref() {
cb(participant.clone(), publication, track);
}
});
}
livekit/src/room/participant/remote_participant.rs
+186 -96
View File
@@ -1,23 +1,39 @@
use super::TrackKind;
use super::{ConnectionQuality, ParticipantInternal};
use super::{ConnectionQuality, ParticipantInner};
use crate::prelude::*;
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;
use parking_lot::Mutex;
use std::collections::{HashMap, HashSet};
use std::fmt::Debug;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::timeout;
const ADD_TRACK_TIMEOUT: Duration = Duration::from_secs(5);
#[derive(Default)]
struct RemoteEvents {
track_published: Mutex<Option<Box<dyn Fn(RemoteParticipant, RemoteTrackPublication) + Send>>>,
track_unpublished: Mutex<Option<Box<dyn Fn(RemoteParticipant, RemoteTrackPublication) + Send>>>,
track_subscribed:
Mutex<Option<Box<dyn Fn(RemoteParticipant, RemoteTrackPublication, RemoteTrack) + Send>>>,
track_unsubscribed:
Mutex<Option<Box<dyn Fn(RemoteParticipant, RemoteTrackPublication, RemoteTrack) + Send>>>,
track_subscription_failed:
Mutex<Option<Box<dyn Fn(RemoteParticipant, TrackSid, TrackError) + Send>>>,
}
struct RemoteInfo {
events: Arc<RemoteEvents>,
}
#[derive(Clone)]
pub struct RemoteParticipant {
inner: Arc<ParticipantInternal>,
inner: Arc<ParticipantInner>,
remote: Arc<RemoteInfo>,
}
impl Debug for RemoteParticipant {
@@ -39,23 +55,13 @@ impl RemoteParticipant {
metadata: String,
) -> Self {
Self {
inner: Arc::new(ParticipantInternal::new(
rtc_engine, sid, identity, name, metadata,
)),
inner: super::new_inner(rtc_engine, sid, identity, name, metadata),
remote: Arc::new(RemoteInfo {
events: Default::default(),
}),
}
}
/// 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: DataPacketKind) {
self.inner
.dispatcher
.dispatch(&ParticipantEvent::DataReceived {
payload: data,
kind,
});
}
pub(crate) async fn add_subscribed_media_track(
&self,
sid: TrackSid,
@@ -106,7 +112,6 @@ impl RemoteParticipant {
log::debug!("starting track: {:?}", sid);
remote_publication.update_track(Some(track.clone().into()));
//track.set_muted(remote_publication.is_muted());
track.update_info(proto::TrackInfo {
sid: remote_publication.sid().to_string(),
@@ -116,26 +121,22 @@ impl RemoteParticipant {
..Default::default()
});
self.inner
.add_publication(TrackPublication::Remote(remote_publication.clone()));
// track.start();
self.add_publication(TrackPublication::Remote(remote_publication.clone()));
track.enable();
self.inner
.dispatcher
.dispatch(&ParticipantEvent::TrackSubscribed {
track,
publication: remote_publication,
});
remote_publication.set_track(Some(track.into())); // This will fire TrackSubscribed on the publication
} else {
log::error!("could not find published track with sid: {:?}", sid);
self.inner
.dispatcher
.dispatch(&ParticipantEvent::TrackSubscriptionFailed {
sid: sid.clone(),
error: TrackError::TrackNotFound(sid.clone().to_string()),
});
if let Some(track_subscription_failed) =
self.remote.events.track_subscription_failed.lock().as_ref()
{
track_subscription_failed(
self.clone(),
sid.clone(),
TrackError::TrackNotFound(sid.0),
);
}
}
}
@@ -144,44 +145,37 @@ impl RemoteParticipant {
// Unsubscribe to the track if needed
if let Some(track) = publication.track() {
track.disable();
self.inner
.dispatcher
.dispatch(&ParticipantEvent::TrackUnsubscribed {
track: track.clone(),
publication: publication.clone(),
});
publication.set_track(None); // This will fire TrackUnsubscribed on the publication
}
self.inner.remove_publication(sid);
self.remove_publication(sid);
self.inner
.dispatcher
.dispatch(&ParticipantEvent::TrackUnpublished {
publication: publication.clone(),
});
publication.update_track(None);
if let Some(track_unpublished) = self.remote.events.track_unpublished.lock().as_ref() {
track_unpublished(self.clone(), publication.clone());
}
}
}
pub(crate) fn update_info(&self, info: proto::ParticipantInfo) {
self.inner.update_info(info.clone());
super::update_info(
&self.inner,
&Participant::Remote(self.clone()),
info.clone(),
);
let mut valid_tracks = HashSet::<TrackSid>::new();
for track in info.tracks {
if let Some(publication) = self.get_track_publication(&track.sid.clone().into()) {
publication.update_info(track.clone());
} else {
let publication =
RemoteTrackPublication::new(track.clone(), Arc::downgrade(&self.inner), None);
self.inner
.add_publication(TrackPublication::Remote(publication.clone()));
let publication = RemoteTrackPublication::new(track.clone(), None);
self.add_publication(TrackPublication::Remote(publication.clone()));
// This is a new track, dispatch publish event
self.inner
.dispatcher
.dispatch(&ParticipantEvent::TrackPublished { publication });
if let Some(track_published) = self.remote.events.track_published.lock().as_ref() {
track_published(self.clone(), publication);
}
}
valid_tracks.insert(track.sid.into());
@@ -197,7 +191,131 @@ impl RemoteParticipant {
}
}
#[inline]
pub(crate) fn on_track_published(
&self,
track_published: impl Fn(RemoteParticipant, RemoteTrackPublication) + Send + 'static,
) {
*self.remote.events.track_published.lock() = Some(Box::new(track_published));
}
pub(crate) fn on_track_unpublished(
&self,
track_unpublished: impl Fn(RemoteParticipant, RemoteTrackPublication) + Send + 'static,
) {
*self.remote.events.track_unpublished.lock() = Some(Box::new(track_unpublished));
}
pub(crate) fn on_track_subscribed(
&self,
track_subscribed: impl Fn(RemoteParticipant, RemoteTrackPublication, RemoteTrack)
+ Send
+ 'static,
) {
*self.remote.events.track_subscribed.lock() = Some(Box::new(track_subscribed));
}
pub(crate) fn on_track_unsubscribed(
&self,
track_unsubscribed: impl Fn(RemoteParticipant, RemoteTrackPublication, RemoteTrack)
+ Send
+ 'static,
) {
*self.remote.events.track_unsubscribed.lock() = Some(Box::new(track_unsubscribed));
}
pub(crate) fn on_track_subscription_failed(
&self,
track_subscription_failed: impl Fn(RemoteParticipant, TrackSid, TrackError) + Send + 'static,
) {
*self.remote.events.track_subscription_failed.lock() =
Some(Box::new(track_subscription_failed));
}
pub(crate) fn set_speaking(&self, speaking: bool) {
super::set_speaking(&self.inner, &Participant::Remote(self.clone()), speaking);
}
pub(crate) fn set_audio_level(&self, level: f32) {
super::set_audio_level(&self.inner, &Participant::Remote(self.clone()), level);
}
pub(crate) fn set_connection_quality(&self, quality: ConnectionQuality) {
super::set_connection_quality(&self.inner, &Participant::Remote(self.clone()), quality);
}
pub(crate) fn add_publication(&self, publication: TrackPublication) {
super::add_publication(
&self.inner,
&Participant::Remote(self.clone()),
publication.clone(),
);
let TrackPublication::Remote(publication) = publication else {
panic!("expected remote publication");
};
publication.on_subscription_update_needed({
let rtc_engine = self.inner.rtc_engine.clone();
let psid = self.sid().0.clone();
move |publication| {
let rtc_engine = rtc_engine.clone();
let psid = psid.clone();
tokio::spawn(async move {
let tsid = publication.sid().0.clone();
let update_subscription = proto::UpdateSubscription {
track_sids: vec![tsid.clone()],
subscribe: publication.is_subscribed(),
participant_tracks: vec![proto::ParticipantTracks {
participant_sid: psid,
track_sids: vec![tsid.clone()],
}],
};
let _ = rtc_engine
.send_request(proto::signal_request::Message::Subscription(
update_subscription,
))
.await;
});
}
});
publication.on_subscribed({
let events = self.remote.events.clone();
let participant = self.clone();
move |publication, track| {
if let Some(track_subscribed) = events.track_subscribed.lock().as_ref() {
track_subscribed(participant.clone(), publication, track);
}
}
});
publication.on_unsubscribed({
let events = self.remote.events.clone();
let participant = self.clone();
move |publication, track| {
if let Some(track_unsubscribed) = events.track_unsubscribed.lock().as_ref() {
track_unsubscribed(participant.clone(), publication, track);
}
}
});
}
pub(crate) fn remove_publication(&self, sid: &TrackSid) {
let publication =
super::remove_publication(&self.inner, &Participant::Remote(self.clone()), sid);
if let Some(publication) = publication {
let TrackPublication::Remote(publication) = publication else {
panic!("expected remote publication");
};
publication.on_subscription_update_needed(|_| {});
publication.on_subscribed(|_, _| {});
publication.on_unsubscribed(|_, _| {});
}
}
pub fn get_track_publication(&self, sid: &TrackSid) -> Option<RemoteTrackPublication> {
self.inner.tracks.read().get(sid).map(|track| {
if let TrackPublication::Remote(remote) = track {
@@ -207,63 +325,35 @@ impl RemoteParticipant {
})
}
#[inline]
pub fn sid(&self) -> ParticipantSid {
self.inner.sid()
self.inner.info.read().sid.clone()
}
#[inline]
pub fn identity(&self) -> ParticipantIdentity {
self.inner.identity()
self.inner.info.read().identity.clone()
}
#[inline]
pub fn name(&self) -> String {
self.inner.name()
self.inner.info.read().name.clone()
}
#[inline]
pub fn metadata(&self) -> String {
self.inner.metadata()
self.inner.info.read().metadata.clone()
}
#[inline]
pub fn is_speaking(&self) -> bool {
self.inner.is_speaking()
self.inner.info.read().speaking
}
#[inline]
pub fn tracks(&self) -> RwLockReadGuard<HashMap<TrackSid, TrackPublication>> {
self.inner.tracks()
pub fn tracks(&self) -> HashMap<TrackSid, TrackPublication> {
self.inner.tracks.read().clone()
}
#[inline]
pub fn audio_level(&self) -> f32 {
self.inner.audio_level()
self.inner.info.read().audio_level
}
#[inline]
pub fn connection_quality(&self) -> ConnectionQuality {
self.inner.connection_quality()
}
#[inline]
pub fn register_observer(&self) -> mpsc::UnboundedReceiver<ParticipantEvent> {
self.inner.register_observer()
}
#[inline]
pub(crate) fn set_speaking(&self, speaking: bool) {
self.inner.set_speaking(speaking);
}
#[inline]
pub(crate) fn set_audio_level(&self, level: f32) {
self.inner.set_audio_level(level);
}
#[inline]
pub(crate) fn set_connection_quality(&self, quality: ConnectionQuality) {
self.inner.set_connection_quality(quality);
self.inner.info.read().connection_quality
}
}
livekit/src/room/publication/local.rs
+47 -74
View File
@@ -1,127 +1,100 @@
use super::TrackPublicationInner;
use crate::id::TrackSid;
use crate::participant::ParticipantInternal;
use crate::track::{LocalTrack, TrackDimension, TrackKind, TrackSource};
use crate::prelude::*;
use livekit_protocol as proto;
use std::sync::{Arc, Weak};
use std::fmt::Debug;
use std::sync::Arc;
#[derive(Debug)]
struct LocalTrackPublicationInner {
publication_inner: TrackPublicationInner,
#[derive(Clone)]
pub struct LocalTrackPublication {
inner: Arc<TrackPublicationInner>,
}
#[derive(Clone, Debug)]
pub struct LocalTrackPublication {
inner: Arc<LocalTrackPublicationInner>,
impl Debug for LocalTrackPublication {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("LocalTrackPublication")
.field("sid", &self.sid())
.field("name", &self.name())
.field("kind", &self.kind())
.finish()
}
}
impl LocalTrackPublication {
pub(crate) fn new(
info: proto::TrackInfo,
participant: Weak<ParticipantInternal>,
track: LocalTrack,
) -> Self {
pub(crate) fn new(info: proto::TrackInfo, track: LocalTrack) -> Self {
Self {
inner: Arc::new(LocalTrackPublicationInner {
publication_inner: TrackPublicationInner::new(
info,
participant,
Some(track.into()),
),
}),
inner: super::new_inner(info, Some(track.into())),
}
}
pub async fn mute(&self) {}
pub(crate) fn on_muted(&self, f: impl Fn(TrackPublication, Track) + Send + 'static) {
*self.inner.events.muted.lock() = Some(Box::new(f));
}
pub async fn unmute(&self) {}
pub(crate) fn on_unmuted(&self, f: impl Fn(TrackPublication, Track) + Send + 'static) {
*self.inner.events.unmuted.lock() = Some(Box::new(f));
}
pub async fn pause_upstream(&self) {}
pub(crate) fn set_track(&self, track: Option<Track>) {
super::set_track(&self.inner, &TrackPublication::Local(self.clone()), track);
}
pub async fn resume_upstream(&self) {}
#[allow(dead_code)]
pub(crate) fn update_info(&self, info: proto::TrackInfo) {
super::update_info(&self.inner, &TrackPublication::Local(self.clone()), info);
}
/*pub fn set_muted(&self, muted: bool) {
if self.is_muted() == muted {
return;
}
pub fn mute(&self) {
self.track().mute();
}
self.track().rtc_track().set_enabled(!muted);
pub fn unmute(&self) {
self.track().unmute();
}
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()
self.inner.info.read().sid.clone()
}
#[inline]
pub fn name(&self) -> String {
self.inner.publication_inner.name()
self.inner.info.read().name.clone()
}
#[inline]
pub fn kind(&self) -> TrackKind {
self.inner.publication_inner.kind()
self.inner.info.read().kind
}
#[inline]
pub fn source(&self) -> TrackSource {
self.inner.publication_inner.source()
self.inner.info.read().source
}
#[inline]
pub fn simulcasted(&self) -> bool {
self.inner.publication_inner.simulcasted()
self.inner.info.read().simulcasted
}
#[inline]
pub fn dimension(&self) -> TrackDimension {
self.inner.publication_inner.dimension()
self.inner.info.read().dimension
}
#[inline]
pub fn track(&self) -> LocalTrack {
self.inner
.publication_inner
.track()
.info
.read()
.track
.clone()
.unwrap()
.try_into()
.unwrap()
}
#[inline]
pub fn mime_type(&self) -> String {
self.inner.publication_inner.mime_type()
self.inner.info.read().mime_type.clone()
}
#[inline]
pub fn is_muted(&self) -> bool {
self.inner.publication_inner.is_muted()
self.inner.info.read().muted
}
#[inline]
pub fn is_remote(&self) -> bool {
false
}
/*#[inline]
pub(crate) fn update_track(&self, track: Option<Track>) {
self.inner.publication_inner.update_track(track);
}*/
#[allow(dead_code)]
#[inline]
pub(crate) fn update_info(&self, info: proto::TrackInfo) {
self.inner.publication_inner.update_info(info);
}
}
livekit/src/room/publication/mod.rs
+101 -177
View File
@@ -1,14 +1,10 @@
use super::track::TrackDimension;
use crate::participant::ParticipantInternal;
use crate::prelude::*;
use crate::track::Track;
use livekit_protocol as proto;
use livekit_protocol::enum_dispatch;
use parking_lot::{Mutex, RwLock};
use proto::observer::Dispatcher;
use std::sync::Arc;
use std::sync::Weak;
use tokio::sync::Notify;
mod local;
mod remote;
@@ -16,23 +12,6 @@ mod remote;
pub use local::*;
pub use remote::*;
#[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,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SubscriptionStatus {
Desired,
@@ -64,8 +43,20 @@ impl TrackPublication {
pub fn mime_type(self: &Self) -> String;
pub fn is_muted(self: &Self) -> bool;
pub fn is_remote(self: &Self) -> bool;
pub(crate) fn on_muted(self: &Self, on_mute: impl Fn(TrackPublication, Track) + Send + 'static) -> ();
pub(crate) fn on_unmuted(self: &Self, on_unmute: impl Fn(TrackPublication, Track) + Send + 'static) -> ();
pub(crate) fn update_info(self: &Self, info: proto::TrackInfo) -> ();
);
#[allow(dead_code)]
pub(crate) fn set_track(&self, track: Option<Track>) {
match self {
TrackPublication::Local(p) => p.set_track(track),
TrackPublication::Remote(p) => p.set_track(track.map(|t| t.try_into().unwrap())),
}
}
pub fn track(&self) -> Option<Track> {
match self {
TrackPublication::Local(p) => Some(p.track().into()),
@@ -74,173 +65,106 @@ impl TrackPublication {
}
}
#[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,
struct PublicationInfo {
pub track: Option<Track>,
pub name: String,
pub sid: TrackSid,
pub kind: TrackKind,
pub source: TrackSource,
pub simulcasted: bool,
pub dimension: TrackDimension,
pub mime_type: String,
pub muted: bool,
}
#[derive(Debug)]
pub(crate) struct TrackPublicationInner {
#[derive(Default)]
struct PublicationEvents {
muted: Mutex<Option<Box<dyn Fn(TrackPublication, Track) + Send>>>,
unmuted: Mutex<Option<Box<dyn Fn(TrackPublication, Track) + Send>>>,
}
pub(super) struct TrackPublicationInner {
info: RwLock<PublicationInfo>,
dispatcher: Dispatcher<PublicationEvent>,
participant: Weak<ParticipantInternal>,
//forward_handle: Mutex<Option<tokio::task::JoinHandle<()>>>,
forward_close: Arc<Notify>,
events: Arc<PublicationEvents>,
}
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,
};
pub(super) fn new_inner(
info: proto::TrackInfo,
track: Option<Track>,
) -> Arc<TrackPublicationInner> {
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(),
}
Arc::new(TrackPublicationInner {
info: RwLock::new(info),
events: Default::default(),
})
}
pub(super) fn update_info(
inner: &TrackPublicationInner,
_publication: &TrackPublication,
new_info: proto::TrackInfo,
) {
let mut info = inner.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;
}
pub(super) fn set_track(
inner: &TrackPublicationInner,
publication: &TrackPublication,
track: Option<Track>,
) {
let mut info = inner.info.write();
if let Some(prev_track) = info.track.as_ref() {
prev_track.on_muted(|_| {});
prev_track.on_unmuted(|_| {});
}
// 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();
info.track = track.clone();
tokio::select! {
_ = notified => {
break;
}
Some(event) = track_events.recv() => {
match event {
TrackEvent::Muted => {
dispatcher.dispatch(&PublicationEvent::Muted);
}
TrackEvent::Unmuted => {
dispatcher.dispatch(&PublicationEvent::Unmuted);
}
}
if let Some(track) = track.as_ref() {
info.sid = track.sid();
track.on_muted({
let events = inner.events.clone();
let publication = publication.clone();
move |track| {
if let Some(on_muted) = events.muted.lock().as_ref() {
on_muted(publication.clone(), track);
}
}
}
}
});
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
track.on_unmuted({
let events = inner.events.clone();
let publication = publication.clone();
move |track| {
if let Some(on_unmuted) = events.unmuted.lock().as_ref() {
on_unmuted(publication.clone(), track);
}
}
});
}
}
livekit/src/room/publication/remote.rs
+180 -93
View File
@@ -1,101 +1,203 @@
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 super::{PermissionStatus, SubscriptionStatus, TrackPublication, TrackPublicationInner};
use crate::prelude::*;
use livekit_protocol as proto;
use parking_lot::RwLock;
use std::sync::{Arc, Weak};
use parking_lot::{Mutex, RwLock};
use std::fmt::Debug;
use std::sync::Arc;
#[derive(Default)]
struct RemoteEvents {
subscribed: Mutex<Option<Box<dyn Fn(RemoteTrackPublication, RemoteTrack) + Send>>>,
unsubscribed: Mutex<Option<Box<dyn Fn(RemoteTrackPublication, RemoteTrack) + Send>>>,
subscription_status_changed: Mutex<
Option<Box<dyn Fn(RemoteTrackPublication, SubscriptionStatus, SubscriptionStatus) + Send>>,
>, // Old status, new status
permission_status_changed: Mutex<
Option<Box<dyn Fn(RemoteTrackPublication, PermissionStatus, PermissionStatus) + Send>>,
>, // Old status, new status
subscription_update_needed: Mutex<Option<Box<dyn Fn(RemoteTrackPublication) + Send>>>,
}
#[derive(Debug)]
struct RemoteInfo {
subscribed: bool,
allowed: bool,
// TODO(theomonnom): other remote info
}
#[derive(Debug)]
struct RemoteInner {
publication_inner: TrackPublicationInner,
info: RwLock<RemoteInfo>,
events: RemoteEvents,
}
#[derive(Clone, Debug)]
#[derive(Clone)]
pub struct RemoteTrackPublication {
inner: Arc<RemoteInner>,
inner: Arc<TrackPublicationInner>,
remote: Arc<RemoteInner>,
}
impl Debug for RemoteTrackPublication {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("RemoteTrackPublication")
.field("is_subscribed", &self.is_subscribed())
.field("is_allowed", &self.is_allowed())
.finish()
}
}
impl RemoteTrackPublication {
pub(crate) fn new(
info: proto::TrackInfo,
participant: Weak<ParticipantInternal>,
track: Option<RemoteTrack>,
) -> Self {
pub(crate) fn new(info: proto::TrackInfo, track: Option<RemoteTrack>) -> Self {
Self {
inner: Arc::new(RemoteInner {
publication_inner: TrackPublicationInner::new(
info,
participant,
track.map(Into::into),
),
inner: super::new_inner(info, track.map(Into::into)),
remote: Arc::new(RemoteInner {
info: RwLock::new(RemoteInfo {
subscribed: false,
allowed: false,
}),
events: Default::default(),
}),
}
}
pub fn set_subscribed(&self, subscribed: bool) {
/// This is called by the RemoteParticipant when it successfully subscribe to the track or when
/// the track is being unsubscribed.
/// We register the mute events from the track here so we can forward them.
pub(crate) fn set_track(&self, track: Option<RemoteTrack>) {
let old_subscription_state = self.subscription_status();
let old_permission_state = self.permission_status();
let mut info = self.inner.info.write();
let prev_track = self.track();
if let Some(prev_track) = prev_track {
if let Some(unsubscribed) = self.remote.events.unsubscribed.lock().as_ref() {
unsubscribed(self.clone(), prev_track);
}
}
super::set_track(
&self.inner,
&TrackPublication::Remote(self.clone()),
track.clone().map(Into::into),
);
if let Some(track) = track {
if let Some(subscribed) = self.remote.events.subscribed.lock().as_ref() {
subscribed(self.clone(), track);
}
}
self.emit_subscription_update(old_subscription_state);
self.emit_permission_update(old_permission_state);
}
pub(crate) fn emit_subscription_update(&self, old_subscription_state: SubscriptionStatus) {
if old_subscription_state != self.subscription_status() {
if let Some(subscription_status_changed) = self
.remote
.events
.subscription_status_changed
.lock()
.as_ref()
{
subscription_status_changed(
self.clone(),
old_subscription_state,
self.subscription_status(),
);
}
}
}
pub(crate) fn emit_permission_update(&self, old_permission_state: PermissionStatus) {
if old_permission_state != self.permission_status() {
if let Some(subscription_permission_changed) =
self.remote.events.permission_status_changed.lock().as_ref()
{
subscription_permission_changed(
self.clone(),
old_permission_state,
self.permission_status(),
);
}
}
}
pub(crate) fn update_info(&self, info: proto::TrackInfo) {
super::update_info(&self.inner, &TrackPublication::Remote(self.clone()), info);
}
pub(crate) fn on_muted(&self, f: impl Fn(TrackPublication, Track) + Send + 'static) {
*self.inner.events.muted.lock() = Some(Box::new(f));
}
pub(crate) fn on_unmuted(&self, f: impl Fn(TrackPublication, Track) + Send + 'static) {
*self.inner.events.unmuted.lock() = Some(Box::new(f));
}
pub(crate) fn on_subscribed(
&self,
f: impl Fn(RemoteTrackPublication, RemoteTrack) + Send + 'static,
) {
*self.remote.events.subscribed.lock() = Some(Box::new(f));
}
pub(crate) fn on_unsubscribed(
&self,
f: impl Fn(RemoteTrackPublication, RemoteTrack) + Send + 'static,
) {
*self.remote.events.unsubscribed.lock() = Some(Box::new(f));
}
#[allow(dead_code)]
pub(crate) fn on_subscription_status_changed(
&self,
f: impl Fn(RemoteTrackPublication, SubscriptionStatus, SubscriptionStatus) + Send + 'static,
) {
*self.remote.events.subscription_status_changed.lock() = Some(Box::new(f));
}
#[allow(dead_code)]
pub(crate) fn on_permission_status_changed(
&self,
f: impl Fn(RemoteTrackPublication, PermissionStatus, PermissionStatus) + Send + 'static,
) {
*self.remote.events.permission_status_changed.lock() = Some(Box::new(f));
}
pub(crate) fn on_subscription_update_needed(
&self,
f: impl Fn(RemoteTrackPublication) + Send + 'static,
) {
*self.remote.events.subscription_update_needed.lock() = Some(Box::new(f));
}
pub async fn set_subscribed(&self, subscribed: bool) {
let old_subscription_state = self.subscription_status();
let old_permission_state = self.permission_status();
let mut info = self.remote.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(),
},
)
// Request to send an update to the SFU
if let Some(subscription_update_needed) = self
.remote
.events
.subscription_update_needed
.lock()
.as_ref()
{
subscription_update_needed(self.clone());
}
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(),
},
)
}
self.emit_subscription_update(old_subscription_state);
self.emit_permission_update(old_permission_state);
}
#[inline]
pub fn subscription_status(&self) -> SubscriptionStatus {
if !self.inner.info.read().subscribed {
if !self.is_subscribed() {
return SubscriptionStatus::Unsubscribed;
}
@@ -106,9 +208,8 @@ impl RemoteTrackPublication {
SubscriptionStatus::Subscribed
}
#[inline]
pub fn permission_status(&self) -> PermissionStatus {
if self.inner.info.read().allowed {
if self.is_allowed() {
PermissionStatus::Allowed
} else {
PermissionStatus::NotAllowed
@@ -116,69 +217,55 @@ impl RemoteTrackPublication {
}
pub fn is_subscribed(&self) -> bool {
self.inner.info.read().allowed && self.track().is_some()
self.is_allowed() && self.track().is_some()
}
pub fn is_allowed(&self) -> bool {
self.remote.info.read().allowed
}
#[inline]
pub fn sid(&self) -> TrackSid {
self.inner.publication_inner.sid()
self.inner.info.read().sid.clone()
}
#[inline]
pub fn name(&self) -> String {
self.inner.publication_inner.name()
self.inner.info.read().name.clone()
}
#[inline]
pub fn kind(&self) -> TrackKind {
self.inner.publication_inner.kind()
self.inner.info.read().kind
}
#[inline]
pub fn source(&self) -> TrackSource {
self.inner.publication_inner.source()
self.inner.info.read().source
}
#[inline]
pub fn simulcasted(&self) -> bool {
self.inner.publication_inner.simulcasted()
self.inner.info.read().simulcasted
}
#[inline]
pub fn dimension(&self) -> TrackDimension {
self.inner.publication_inner.dimension()
self.inner.info.read().dimension.clone()
}
#[inline]
pub fn track(&self) -> Option<RemoteTrack> {
self.inner
.publication_inner
.track()
.info
.read()
.track
.clone()
.map(|track| track.try_into().unwrap())
}
#[inline]
pub fn mime_type(&self) -> String {
self.inner.publication_inner.mime_type()
self.inner.info.read().mime_type.clone()
}
#[inline]
pub fn is_muted(&self) -> bool {
self.inner.publication_inner.is_muted()
self.inner.info.read().muted
}
#[inline]
pub fn is_remote(&self) -> bool {
true
}
#[inline]
pub(crate) fn update_track(&self, track: Option<Track>) {
self.inner.publication_inner.update_track(track);
}
#[inline]
pub(crate) fn update_info(&self, info: proto::TrackInfo) {
self.inner.publication_inner.update_info(info);
}
}
livekit/src/room/track/audio_track.rs
+43
View File
@@ -0,0 +1,43 @@
use super::track_dispatch;
use crate::prelude::*;
use livekit_protocol as proto;
use livekit_protocol::enum_dispatch;
use livekit_webrtc::prelude::*;
#[derive(Clone, Debug)]
pub enum AudioTrack {
Local(LocalAudioTrack),
Remote(RemoteAudioTrack),
}
impl AudioTrack {
track_dispatch!([Local, Remote]);
pub fn rtc_track(&self) -> RtcAudioTrack {
match self {
Self::Local(track) => track.rtc_track().into(),
Self::Remote(track) => track.rtc_track().into(),
}
}
}
impl From<AudioTrack> for Track {
fn from(track: AudioTrack) -> Self {
match track {
AudioTrack::Local(track) => Self::LocalAudio(track),
AudioTrack::Remote(track) => Self::RemoteAudio(track),
}
}
}
impl TryFrom<Track> for AudioTrack {
type Error = &'static str;
fn try_from(track: Track) -> Result<Self, Self::Error> {
match track {
Track::LocalAudio(track) => Ok(Self::Local(track)),
Track::RemoteAudio(track) => Ok(Self::Remote(track)),
_ => Err("not an audio track"),
}
}
}
livekit/src/room/track/local_audio_track.rs
+76 -92
View File
@@ -6,7 +6,6 @@ use livekit_protocol as proto;
use livekit_webrtc::prelude::*;
use std::fmt::Debug;
use std::sync::Arc;
use tokio::sync::mpsc;
#[derive(Clone)]
pub struct LocalAudioTrack {
@@ -27,7 +26,7 @@ impl Debug for LocalAudioTrack {
impl LocalAudioTrack {
pub(crate) fn new(name: String, rtc_track: RtcAudioTrack, source: RtcAudioSource) -> Self {
Self {
inner: Arc::new(TrackInner::new(
inner: Arc::new(super::new_inner(
"unknown".to_string().into(), // sid
name,
TrackKind::Audio,
@@ -37,96 +36,6 @@ impl LocalAudioTrack {
}
}
#[inline]
pub fn sid(&self) -> TrackSid {
self.inner.sid()
}
#[inline]
pub fn name(&self) -> String {
self.inner.name()
}
#[inline]
pub fn kind(&self) -> TrackKind {
self.inner.kind()
}
#[inline]
pub fn source(&self) -> TrackSource {
self.inner.source()
}
#[inline]
pub fn stream_state(&self) -> StreamState {
self.inner.stream_state()
}
#[inline]
pub fn enable(&self) {
self.inner.enable()
}
#[inline]
pub fn disable(&self) {
self.inner.disable()
}
#[inline]
pub fn is_muted(&self) -> bool {
self.inner.is_muted()
}
#[inline]
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.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.register_observer()
}
#[inline]
pub fn is_remote(&self) -> bool {
false
}
#[inline]
pub(crate) fn transceiver(&self) -> Option<RtpTransceiver> {
self.inner.transceiver()
}
#[inline]
pub(crate) fn update_transceiver(&self, transceiver: Option<RtpTransceiver>) {
self.inner.update_transceiver(transceiver)
}
#[inline]
pub(crate) fn update_info(&self, info: proto::TrackInfo) {
self.inner.update_info(info)
}
}
impl LocalAudioTrack {
pub fn create_audio_track(name: &str, source: RtcAudioSource) -> LocalAudioTrack {
let rtc_track = match source.clone() {
#[cfg(not(target_arch = "wasm32"))]
@@ -141,4 +50,79 @@ impl LocalAudioTrack {
};
Self::new(name.to_string(), rtc_track, source)
}
pub fn sid(&self) -> TrackSid {
self.inner.info.read().sid.clone()
}
pub fn name(&self) -> String {
self.inner.info.read().name.clone()
}
pub fn kind(&self) -> TrackKind {
self.inner.info.read().kind
}
pub fn source(&self) -> TrackSource {
self.inner.info.read().source
}
pub fn stream_state(&self) -> StreamState {
self.inner.info.read().stream_state
}
pub fn enable(&self) {
self.inner.rtc_track.set_enabled(true);
}
pub fn disable(&self) {
self.inner.rtc_track.set_enabled(false);
}
pub fn is_muted(&self) -> bool {
self.inner.info.read().muted
}
pub fn mute(&self) {
super::set_muted(&self.inner, &Track::LocalAudio(self.clone()), true);
}
pub fn unmute(&self) {
super::set_muted(&self.inner, &Track::LocalAudio(self.clone()), false);
}
pub fn rtc_track(&self) -> RtcAudioTrack {
if let MediaStreamTrack::Audio(audio) = self.inner.rtc_track.clone() {
return audio;
}
unreachable!();
}
pub fn rtc_source(&self) -> RtcAudioSource {
self.source.clone()
}
pub fn is_remote(&self) -> bool {
false
}
pub fn on_muted(&self, f: impl Fn(Track) + Send + 'static) {
*self.inner.events.muted.lock() = Some(Box::new(f));
}
pub fn on_unmuted(&self, f: impl Fn(Track) + Send + 'static) {
*self.inner.events.unmuted.lock() = Some(Box::new(f));
}
pub(crate) fn transceiver(&self) -> Option<RtpTransceiver> {
self.inner.info.read().transceiver.clone()
}
pub(crate) fn set_transceiver(&self, transceiver: Option<RtpTransceiver>) {
self.inner.info.write().transceiver = transceiver;
}
pub(crate) fn update_info(&self, info: proto::TrackInfo) {
super::update_info(&self.inner, &Track::LocalAudio(self.clone()), info);
}
}
livekit/src/room/track/local_track.rs
+22 -6
View File
@@ -1,12 +1,8 @@
use super::TrackInner;
use super::{track_dispatch, LocalAudioTrack, LocalVideoTrack};
use super::track_dispatch;
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 {
@@ -23,7 +19,6 @@ impl LocalTrack {
pub fn unmute(self: &Self) -> ();
);
#[inline]
pub fn rtc_track(&self) -> MediaStreamTrack {
match self {
Self::Audio(track) => track.rtc_track().into(),
@@ -31,3 +26,24 @@ impl LocalTrack {
}
}
}
impl From<LocalTrack> for Track {
fn from(track: LocalTrack) -> Self {
match track {
LocalTrack::Audio(track) => Self::LocalAudio(track),
LocalTrack::Video(track) => Self::LocalVideo(track),
}
}
}
impl TryFrom<Track> for LocalTrack {
type Error = &'static str;
fn try_from(track: Track) -> Result<Self, Self::Error> {
match track {
Track::LocalAudio(track) => Ok(Self::Audio(track)),
Track::LocalVideo(track) => Ok(Self::Video(track)),
_ => Err("not a local track"),
}
}
}
livekit/src/room/track/local_video_track.rs
+76 -92
View File
@@ -5,7 +5,6 @@ use livekit_protocol as proto;
use livekit_webrtc::prelude::*;
use std::fmt::Debug;
use std::sync::Arc;
use tokio::sync::mpsc;
#[derive(Clone)]
pub struct LocalVideoTrack {
@@ -26,7 +25,7 @@ impl Debug for LocalVideoTrack {
impl LocalVideoTrack {
pub fn new(name: String, rtc_track: RtcVideoTrack, source: RtcVideoSource) -> Self {
Self {
inner: Arc::new(TrackInner::new(
inner: Arc::new(super::new_inner(
"unknown".to_string().into(), // sid
name,
TrackKind::Video,
@@ -36,96 +35,6 @@ impl LocalVideoTrack {
}
}
#[inline]
pub fn sid(&self) -> TrackSid {
self.inner.sid()
}
#[inline]
pub fn name(&self) -> String {
self.inner.name()
}
#[inline]
pub fn kind(&self) -> TrackKind {
self.inner.kind()
}
#[inline]
pub fn source(&self) -> TrackSource {
self.inner.source()
}
#[inline]
pub fn stream_state(&self) -> StreamState {
self.inner.stream_state()
}
#[inline]
pub fn enable(&self) {
self.inner.enable()
}
#[inline]
pub fn disable(&self) {
self.inner.disable()
}
#[inline]
pub fn is_muted(&self) -> bool {
self.inner.is_muted()
}
#[inline]
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.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.register_observer()
}
#[inline]
pub fn is_remote(&self) -> bool {
false
}
#[inline]
pub(crate) fn transceiver(&self) -> Option<RtpTransceiver> {
self.inner.transceiver()
}
#[inline]
pub(crate) fn update_transceiver(&self, transceiver: Option<RtpTransceiver>) {
self.inner.update_transceiver(transceiver)
}
#[inline]
pub(crate) fn update_info(&self, info: proto::TrackInfo) {
self.inner.update_info(info)
}
}
impl LocalVideoTrack {
pub fn create_video_track(name: &str, source: RtcVideoSource) -> LocalVideoTrack {
let rtc_track = match source.clone() {
#[cfg(not(target_arch = "wasm32"))]
@@ -141,4 +50,79 @@ impl LocalVideoTrack {
Self::new(name.to_string(), rtc_track, source)
}
pub fn sid(&self) -> TrackSid {
self.inner.info.read().sid.clone()
}
pub fn name(&self) -> String {
self.inner.info.read().name.clone()
}
pub fn kind(&self) -> TrackKind {
self.inner.info.read().kind
}
pub fn source(&self) -> TrackSource {
self.inner.info.read().source
}
pub fn stream_state(&self) -> StreamState {
self.inner.info.read().stream_state
}
pub fn enable(&self) {
self.inner.rtc_track.set_enabled(true);
}
pub fn disable(&self) {
self.inner.rtc_track.set_enabled(false);
}
pub fn is_muted(&self) -> bool {
self.inner.info.read().muted
}
pub fn mute(&self) {
super::set_muted(&self.inner, &Track::LocalVideo(self.clone()), true);
}
pub fn unmute(&self) {
super::set_muted(&self.inner, &Track::LocalVideo(self.clone()), false);
}
pub fn rtc_track(&self) -> RtcVideoTrack {
if let MediaStreamTrack::Video(video) = self.inner.rtc_track.clone() {
return video;
}
unreachable!();
}
pub fn is_remote(&self) -> bool {
false
}
pub fn rtc_source(&self) -> RtcVideoSource {
self.source.clone()
}
pub fn on_muted(&self, f: impl Fn(Track) + Send + 'static) {
*self.inner.events.muted.lock() = Some(Box::new(f));
}
pub fn on_unmuted(&self, f: impl Fn(Track) + Send + 'static) {
*self.inner.events.unmuted.lock() = Some(Box::new(f));
}
pub(crate) fn transceiver(&self) -> Option<RtpTransceiver> {
self.inner.info.read().transceiver.clone()
}
pub(crate) fn set_transceiver(&self, transceiver: Option<RtpTransceiver>) {
self.inner.info.write().transceiver = transceiver;
}
pub(crate) fn update_info(&self, info: proto::TrackInfo) {
super::update_info(&self.inner, &&Track::LocalVideo(self.clone()), info);
}
}
livekit/src/room/track/mod.rs
+81 -246
View File
@@ -1,25 +1,29 @@
use crate::prelude::*;
use livekit_protocol as proto;
use livekit_protocol::enum_dispatch;
use livekit_protocol::observer::Dispatcher;
use livekit_webrtc::prelude::*;
use parking_lot::RwLock;
use parking_lot::{Mutex, RwLock};
use std::fmt::Debug;
use std::sync::Arc;
use thiserror::Error;
use tokio::sync::mpsc;
mod audio_track;
mod local_audio_track;
mod local_track;
mod local_video_track;
mod remote_audio_track;
mod remote_track;
mod remote_video_track;
mod video_track;
pub use audio_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::*;
pub use video_track::*;
#[derive(Error, Debug, Clone)]
pub enum TrackError {
@@ -48,35 +52,9 @@ pub enum TrackSource {
ScreenshareAudio,
}
#[derive(Debug, Clone)]
pub enum TrackEvent {
Muted,
Unmuted,
}
#[derive(Clone, Copy, Debug)]
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct TrackDimension(pub u32, pub u32);
#[derive(Clone, Debug)]
pub enum Track {
LocalAudio(LocalAudioTrack),
LocalVideo(LocalVideoTrack),
RemoteAudio(RemoteAudioTrack),
RemoteVideo(RemoteVideoTrack),
}
#[derive(Clone, Debug)]
pub enum VideoTrack {
Local(LocalVideoTrack),
Remote(RemoteVideoTrack),
}
#[derive(Clone, Debug)]
pub enum AudioTrack {
Local(LocalAudioTrack),
Remote(RemoteAudioTrack),
}
macro_rules! track_dispatch {
([$($variant:ident),+]) => {
enum_dispatch!(
@@ -90,21 +68,27 @@ macro_rules! track_dispatch {
pub fn disable(self: &Self) -> ();
pub fn is_muted(self: &Self) -> bool;
pub fn is_remote(self: &Self) -> bool;
pub fn register_observer(self: &Self) -> mpsc::UnboundedReceiver<TrackEvent>;
pub fn on_muted(self: &Self, on_mute: impl Fn(Track) + Send + 'static) -> ();
pub fn on_unmuted(self: &Self, on_unmute: impl Fn(Track) + Send + 'static) -> ();
pub(crate) fn transceiver(self: &Self) -> Option<RtpTransceiver>;
pub(crate) fn update_transceiver(self: &Self, transceiver: Option<RtpTransceiver>) -> ();
pub(crate) fn set_transceiver(self: &Self, transceiver: Option<RtpTransceiver>) -> ();
pub(crate) fn update_info(self: &Self, info: proto::TrackInfo) -> ();
);
};
}
pub(crate) use track_dispatch;
#[derive(Clone, Debug)]
pub enum Track {
LocalAudio(LocalAudioTrack),
LocalVideo(LocalVideoTrack),
RemoteAudio(RemoteAudioTrack),
RemoteVideo(RemoteVideoTrack),
}
impl Track {
track_dispatch!([LocalAudio, LocalVideo, RemoteAudio, RemoteVideo]);
#[inline]
pub fn rtc_track(&self) -> MediaStreamTrack {
match self {
Self::LocalAudio(track) => track.rtc_track().into(),
@@ -115,232 +99,83 @@ impl Track {
}
}
impl VideoTrack {
track_dispatch!([Local, Remote]);
pub(super) use track_dispatch;
#[inline]
pub fn rtc_track(&self) -> RtcVideoTrack {
match self {
Self::Local(track) => track.rtc_track(),
Self::Remote(track) => track.rtc_track(),
}
}
}
impl AudioTrack {
track_dispatch!([Local, Remote]);
#[inline]
pub fn rtc_track(&self) -> RtcAudioTrack {
match self {
Self::Local(track) => track.rtc_track().into(),
Self::Remote(track) => track.rtc_track().into(),
}
}
#[derive(Default)]
struct TrackEvents {
pub muted: Mutex<Option<Box<dyn Fn(Track) + Send>>>,
pub unmuted: Mutex<Option<Box<dyn Fn(Track) + Send>>>,
}
#[derive(Debug)]
struct TrackInfo {
pub sid: TrackSid,
pub name: String,
pub kind: TrackKind,
pub source: TrackSource,
pub stream_state: StreamState,
pub muted: bool,
pub transceiver: Option<RtpTransceiver>,
}
pub(super) struct TrackInner {
info: RwLock<TrackInfo>,
rtc_track: MediaStreamTrack,
events: TrackEvents,
}
pub(super) fn new_inner(
sid: TrackSid,
name: String,
kind: TrackKind,
source: TrackSource,
stream_state: StreamState,
muted: bool,
transceiver: Option<RtpTransceiver>,
}
#[derive(Debug)]
pub(crate) struct TrackInner {
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 {
info: RwLock::new(TrackInfo {
sid,
name,
kind,
source: TrackSource::Unknown,
stream_state: StreamState::Active,
muted: false,
transceiver: None,
}),
rtc_track,
dispatcher: Default::default(),
}
}
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 stream_state(&self) -> StreamState {
self.info.read().stream_state
}
pub fn is_muted(&self) -> bool {
self.info.read().muted
}
pub fn enable(&self) {
self.rtc_track.set_enabled(true);
}
pub fn disable(&self) {
self.rtc_track.set_enabled(false);
}
pub fn rtc_track(&self) -> MediaStreamTrack {
self.rtc_track.clone()
}
pub fn register_observer(&self) -> mpsc::UnboundedReceiver<TrackEvent> {
self.dispatcher.register()
}
pub fn transceiver(&self) -> Option<RtpTransceiver> {
self.info.read().transceiver.clone()
}
pub fn update_transceiver(&self, transceiver: Option<RtpTransceiver>) {
self.info.write().transceiver = transceiver;
}
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)
) -> TrackInner {
TrackInner {
info: RwLock::new(TrackInfo {
sid,
name,
kind,
source: TrackSource::Unknown,
stream_state: StreamState::Active,
muted: false,
transceiver: None,
}),
rtc_track,
events: Default::default(),
}
}
impl From<RemoteTrack> for Track {
fn from(track: RemoteTrack) -> Self {
match track {
RemoteTrack::Audio(track) => Self::RemoteAudio(track),
RemoteTrack::Video(track) => Self::RemoteVideo(track),
pub(super) fn set_muted(inner: &Arc<TrackInner>, track: &Track, muted: bool) {
let info = inner.info.read();
log::debug!("set_muted: {} {}", info.sid, muted);
if info.muted == muted {
return;
}
drop(info);
if muted {
inner.rtc_track.set_enabled(false);
} else {
inner.rtc_track.set_enabled(true);
}
inner.info.write().muted = muted;
if muted {
if let Some(on_mute) = inner.events.muted.lock().as_ref() {
on_mute(track.clone());
}
} else {
if let Some(on_unmute) = inner.events.unmuted.lock().as_ref() {
on_unmute(track.clone());
}
}
}
impl From<LocalTrack> for Track {
fn from(track: LocalTrack) -> Self {
match track {
LocalTrack::Audio(track) => Self::LocalAudio(track),
LocalTrack::Video(track) => Self::LocalVideo(track),
}
}
}
impl From<VideoTrack> for Track {
fn from(track: VideoTrack) -> Self {
match track {
VideoTrack::Local(track) => Self::LocalVideo(track),
VideoTrack::Remote(track) => Self::RemoteVideo(track),
}
}
}
impl From<AudioTrack> for Track {
fn from(track: AudioTrack) -> Self {
match track {
AudioTrack::Local(track) => Self::LocalAudio(track),
AudioTrack::Remote(track) => Self::RemoteAudio(track),
}
}
}
impl TryFrom<Track> for RemoteTrack {
type Error = &'static str;
fn try_from(track: Track) -> Result<Self, Self::Error> {
match track {
Track::RemoteAudio(track) => Ok(Self::Audio(track)),
Track::RemoteVideo(track) => Ok(Self::Video(track)),
_ => Err("not a remote track"),
}
}
}
impl TryFrom<Track> for LocalTrack {
type Error = &'static str;
fn try_from(track: Track) -> Result<Self, Self::Error> {
match track {
Track::LocalAudio(track) => Ok(Self::Audio(track)),
Track::LocalVideo(track) => Ok(Self::Video(track)),
_ => Err("not a local track"),
}
}
}
impl TryFrom<Track> for VideoTrack {
type Error = &'static str;
fn try_from(track: Track) -> Result<Self, Self::Error> {
match track {
Track::LocalVideo(track) => Ok(Self::Local(track)),
Track::RemoteVideo(track) => Ok(Self::Remote(track)),
_ => Err("not a video track"),
}
}
}
impl TryFrom<Track> for AudioTrack {
type Error = &'static str;
fn try_from(track: Track) -> Result<Self, Self::Error> {
match track {
Track::LocalAudio(track) => Ok(Self::Local(track)),
Track::RemoteAudio(track) => Ok(Self::Remote(track)),
_ => Err("not an audio track"),
}
}
}
impl From<TrackKind> for MediaType {
fn from(kind: TrackKind) -> Self {
match kind {
TrackKind::Audio => Self::Audio,
TrackKind::Video => Self::Video,
}
}
pub(super) fn update_info(inner: &Arc<TrackInner>, _track: &Track, new_info: proto::TrackInfo) {
let mut info = inner.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());
}
livekit/src/room/track/remote_audio_track.rs
+25 -37
View File
@@ -1,15 +1,13 @@
use super::remote_track;
use super::TrackInner;
use super::{remote_track, TrackInner};
use crate::prelude::*;
use livekit_protocol as proto;
use livekit_webrtc::prelude::*;
use std::fmt::Debug;
use std::sync::Arc;
use tokio::sync::mpsc;
#[derive(Clone)]
pub struct RemoteAudioTrack {
pub(crate) inner: Arc<TrackInner>,
inner: Arc<TrackInner>,
}
impl Debug for RemoteAudioTrack {
@@ -25,7 +23,7 @@ impl Debug for RemoteAudioTrack {
impl RemoteAudioTrack {
pub(crate) fn new(sid: TrackSid, name: String, rtc_track: RtcAudioTrack) -> Self {
Self {
inner: Arc::new(TrackInner::new(
inner: Arc::new(super::new_inner(
sid,
name,
TrackKind::Audio,
@@ -34,78 +32,68 @@ impl RemoteAudioTrack {
}
}
#[inline]
pub fn sid(&self) -> TrackSid {
self.inner.sid()
self.inner.info.read().sid.clone()
}
#[inline]
pub fn name(&self) -> String {
self.inner.name()
self.inner.info.read().name.clone()
}
#[inline]
pub fn kind(&self) -> TrackKind {
self.inner.kind()
self.inner.info.read().kind
}
#[inline]
pub fn source(&self) -> TrackSource {
self.inner.source()
self.inner.info.read().source
}
#[inline]
pub fn stream_state(&self) -> StreamState {
self.inner.stream_state()
self.inner.info.read().stream_state
}
#[inline]
pub fn enable(&self) {
self.inner.enable()
self.inner.rtc_track.set_enabled(true);
}
#[inline]
pub fn disable(&self) {
self.inner.disable()
self.inner.rtc_track.set_enabled(false);
}
#[inline]
pub fn is_muted(&self) -> bool {
self.inner.is_muted()
self.inner.info.read().muted
}
#[inline]
pub fn rtc_track(&self) -> RtcAudioTrack {
if let MediaStreamTrack::Audio(audio) = self.inner.rtc_track() {
if let MediaStreamTrack::Audio(audio) = self.inner.rtc_track.clone() {
return audio;
}
unreachable!()
unreachable!();
}
#[inline]
pub fn register_observer(&self) -> mpsc::UnboundedReceiver<TrackEvent> {
self.inner.register_observer()
}
#[inline]
pub fn is_remote(&self) -> bool {
true
}
pub fn on_muted(&self, f: impl Fn(Track) + Send + 'static) {
*self.inner.events.muted.lock() = Some(Box::new(f));
}
pub fn on_unmuted(&self, f: impl Fn(Track) + Send + 'static) {
*self.inner.events.unmuted.lock() = Some(Box::new(f));
}
#[allow(dead_code)]
#[inline]
pub(crate) fn transceiver(&self) -> Option<RtpTransceiver> {
self.inner.transceiver()
self.inner.info.read().transceiver.clone()
}
#[inline]
#[allow(dead_code)]
pub(crate) fn update_transceiver(&self, transceiver: Option<RtpTransceiver>) {
self.inner.update_transceiver(transceiver)
pub(crate) fn set_transceiver(&self, transceiver: Option<RtpTransceiver>) {
self.inner.info.write().transceiver = transceiver;
}
#[inline]
pub(crate) fn update_info(&self, info: proto::TrackInfo) {
remote_track::update_info(&self.inner, info);
remote_track::update_info(&self.inner, &Track::RemoteAudio(self.clone()), info);
}
}
livekit/src/room/track/remote_track.rs
+24 -6
View File
@@ -1,13 +1,10 @@
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 {
@@ -27,7 +24,28 @@ impl RemoteTrack {
}
}
pub(crate) fn update_info(track: &Arc<TrackInner>, new_info: proto::TrackInfo) {
track.update_info(new_info.clone());
track.set_muted(new_info.muted);
pub(super) fn update_info(inner: &Arc<TrackInner>, track: &Track, new_info: proto::TrackInfo) {
super::update_info(inner, track, new_info.clone());
super::set_muted(inner, track, new_info.muted);
}
impl From<RemoteTrack> for Track {
fn from(track: RemoteTrack) -> Self {
match track {
RemoteTrack::Audio(track) => Self::RemoteAudio(track),
RemoteTrack::Video(track) => Self::RemoteVideo(track),
}
}
}
impl TryFrom<Track> for RemoteTrack {
type Error = &'static str;
fn try_from(track: Track) -> Result<Self, Self::Error> {
match track {
Track::RemoteAudio(track) => Ok(Self::Audio(track)),
Track::RemoteVideo(track) => Ok(Self::Video(track)),
_ => Err("not a local track"),
}
}
}
livekit/src/room/track/remote_video_track.rs
+26 -36
View File
@@ -1,14 +1,14 @@
use super::{remote_track, TrackInner};
use super::remote_track;
use super::TrackInner;
use crate::prelude::*;
use livekit_protocol as proto;
use livekit_webrtc::prelude::*;
use std::fmt::Debug;
use std::sync::Arc;
use tokio::sync::mpsc;
#[derive(Clone)]
pub struct RemoteVideoTrack {
pub(crate) inner: Arc<TrackInner>,
inner: Arc<TrackInner>,
}
impl Debug for RemoteVideoTrack {
@@ -24,7 +24,7 @@ impl Debug for RemoteVideoTrack {
impl RemoteVideoTrack {
pub(crate) fn new(sid: TrackSid, name: String, rtc_track: RtcVideoTrack) -> Self {
Self {
inner: Arc::new(TrackInner::new(
inner: Arc::new(super::new_inner(
sid,
name,
TrackKind::Video,
@@ -33,78 +33,68 @@ impl RemoteVideoTrack {
}
}
#[inline]
pub fn sid(&self) -> TrackSid {
self.inner.sid()
self.inner.info.read().sid.clone()
}
#[inline]
pub fn name(&self) -> String {
self.inner.name()
self.inner.info.read().name.clone()
}
#[inline]
pub fn kind(&self) -> TrackKind {
self.inner.kind()
self.inner.info.read().kind
}
#[inline]
pub fn source(&self) -> TrackSource {
self.inner.source()
self.inner.info.read().source
}
#[inline]
pub fn stream_state(&self) -> StreamState {
self.inner.stream_state()
self.inner.info.read().stream_state
}
#[inline]
pub fn enable(&self) {
self.inner.enable()
self.inner.rtc_track.set_enabled(true);
}
#[inline]
pub fn disable(&self) {
self.inner.disable()
self.inner.rtc_track.set_enabled(false);
}
#[inline]
pub fn is_muted(&self) -> bool {
self.inner.is_muted()
self.inner.info.read().muted
}
#[inline]
pub fn rtc_track(&self) -> RtcVideoTrack {
if let MediaStreamTrack::Video(video) = self.inner.rtc_track() {
if let MediaStreamTrack::Video(video) = self.inner.rtc_track.clone() {
return video;
}
unreachable!()
unreachable!();
}
#[inline]
pub fn register_observer(&self) -> mpsc::UnboundedReceiver<TrackEvent> {
self.inner.register_observer()
}
#[inline]
pub fn is_remote(&self) -> bool {
true
}
pub fn on_muted(&self, f: impl Fn(Track) + Send + 'static) {
*self.inner.events.muted.lock() = Some(Box::new(f));
}
pub fn on_unmuted(&self, f: impl Fn(Track) + Send + 'static) {
*self.inner.events.unmuted.lock() = Some(Box::new(f));
}
#[allow(dead_code)]
#[inline]
pub(crate) fn transceiver(&self) -> Option<RtpTransceiver> {
self.inner.transceiver()
self.inner.info.read().transceiver.clone()
}
#[allow(dead_code)]
#[inline]
pub(crate) fn update_transceiver(&self, transceiver: Option<RtpTransceiver>) {
self.inner.update_transceiver(transceiver)
pub(crate) fn set_transceiver(&self, transceiver: Option<RtpTransceiver>) {
self.inner.info.write().transceiver = transceiver;
}
#[inline]
pub(crate) fn update_info(&self, info: proto::TrackInfo) {
remote_track::update_info(&self.inner, info);
remote_track::update_info(&self.inner, &Track::RemoteVideo(self.clone()), info);
}
}
livekit/src/room/track/video_track.rs
+44
View File
@@ -0,0 +1,44 @@
use super::track_dispatch;
use crate::prelude::*;
use livekit_protocol as proto;
use livekit_protocol::enum_dispatch;
use livekit_webrtc::prelude::*;
#[derive(Clone, Debug)]
pub enum VideoTrack {
Local(LocalVideoTrack),
Remote(RemoteVideoTrack),
}
impl VideoTrack {
track_dispatch!([Local, Remote]);
#[inline]
pub fn rtc_track(&self) -> RtcVideoTrack {
match self {
Self::Local(track) => track.rtc_track(),
Self::Remote(track) => track.rtc_track(),
}
}
}
impl From<VideoTrack> for Track {
fn from(track: VideoTrack) -> Self {
match track {
VideoTrack::Local(track) => Self::LocalVideo(track),
VideoTrack::Remote(track) => Self::RemoteVideo(track),
}
}
}
impl TryFrom<Track> for VideoTrack {
type Error = &'static str;
fn try_from(track: Track) -> Result<Self, Self::Error> {
match track {
Track::LocalVideo(track) => Ok(Self::Local(track)),
Track::RemoteVideo(track) => Ok(Self::Remote(track)),
_ => Err("not a video track"),
}
}
}
livekit/src/rtc_engine/rtc_session.rs
+4 -1
View File
@@ -654,7 +654,10 @@ impl SessionInner {
if track.kind() == TrackKind::Video {
let capabilities = LkRuntime::instance()
.pc_factory()
.get_rtp_sender_capabilities(track.kind().into());
.get_rtp_sender_capabilities(match track.kind() {
TrackKind::Video => MediaType::Video,
TrackKind::Audio => MediaType::Audio,
});
let mut matched = Vec::new();
let mut partial_matched = Vec::new();