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

Merge pull request #9 from livekit/theo/events

Browse Source 
finish LiveKit implementation & added Rust demo
This commit is contained in:
Théo Monnom
2023-01-01 22:38:54 +01:00
committed by GitHub
parent 57adc07363 87dc42b09e
commit 36aa00343b
44 changed files with 3335 additions and 1704 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.cargo/config
-2
View File
@@ -1,2 +0,0 @@
[target.x86_64-pc-windows-msvc]
rustflags = ["-C", "target-feature=+crt-static"]
Cargo.lock
Generated
+4
View File
@@ -512,6 +512,10 @@ dependencies = [
[[package]]
name = "livekit-utils"
version = "0.1.0"
dependencies = [
"parking_lot",
"tokio",
]
[[package]]
name = "livekit-webrtc"
crates/livekit-core/src/events.rs
-140
View File
@@ -1,140 +0,0 @@
use futures_util::future::BoxFuture;
use thiserror::Error;
type EventHandler<T> = Box<dyn FnMut(T) -> BoxFuture<'static, ()> + Send + Sync>;
macro_rules! event_setter {
($fnc:ident, $event:ty) => {
pub fn $fnc<F, Fut>(&self, mut callback: F)
where
F: FnMut($event) -> Fut + Send + Sync + 'static,
Fut: Future<Output = ()> + Send + 'static,
{
*self.$fnc.lock() = Some(Box::new(move |event| Box::pin(callback(event))));
}
};
}
#[derive(Error, Debug, Clone)]
pub enum TrackError {
#[error("could not find published track with sid: {0}")]
TrackNotFound(String),
}
pub mod room {
use super::{EventHandler, TrackError};
use crate::room::id::TrackSid;
use crate::room::participant::remote_participant::RemoteParticipant;
use crate::room::publication::RemoteTrackPublication;
use crate::room::track::remote_track::RemoteTrackHandle;
use crate::room::RoomHandle;
use futures::future::Future;
use parking_lot::Mutex;
use std::sync::Arc;
#[derive(Clone)]
pub struct ParticipantConnectedEvent {
pub room_handle: RoomHandle,
pub participant: Arc<RemoteParticipant>,
}
#[derive(Clone)]
pub struct ParticipantDisconnectedEvent {
pub room_handle: RoomHandle,
pub participant: Arc<RemoteParticipant>,
}
#[derive(Clone)]
pub struct TrackSubscribedEvent {
pub room_handle: RoomHandle,
pub track: RemoteTrackHandle,
pub publication: RemoteTrackPublication,
pub participant: Arc<RemoteParticipant>,
}
#[derive(Clone)]
pub struct TrackPublishedEvent {
pub room_handle: RoomHandle,
pub publication: RemoteTrackPublication,
pub participant: Arc<RemoteParticipant>,
}
#[derive(Clone)]
pub struct TrackSubscriptionFailedEvent {
pub room_handle: RoomHandle,
pub error: TrackError,
pub sid: TrackSid,
pub participant: Arc<RemoteParticipant>,
}
pub(crate) type OnParticipantConnectedHandler = EventHandler<ParticipantConnectedEvent>;
pub(crate) type OnParticipantDisconnectedHandler = EventHandler<ParticipantDisconnectedEvent>;
pub(crate) type OnTrackSubscribedEventHandler = EventHandler<TrackSubscribedEvent>;
pub(crate) type OnTrackPublishedEventHandler = EventHandler<TrackPublishedEvent>;
pub(crate) type OnTrackSubscriptionFailedHandler = EventHandler<TrackSubscriptionFailedEvent>;
#[derive(Default)]
pub struct RoomEvents {
pub(crate) on_participant_connected: Mutex<Option<OnParticipantConnectedHandler>>,
pub(crate) on_participant_disconnected: Mutex<Option<OnParticipantDisconnectedHandler>>,
pub(crate) on_track_subscribed: Mutex<Option<OnTrackSubscribedEventHandler>>,
pub(crate) on_track_published: Mutex<Option<OnTrackPublishedEventHandler>>,
pub(crate) on_track_subscription_failed: Mutex<Option<OnTrackSubscriptionFailedHandler>>,
}
impl RoomEvents {
event_setter!(on_participant_connected, ParticipantConnectedEvent);
event_setter!(on_participant_disconnected, ParticipantDisconnectedEvent);
event_setter!(on_track_subscribed, TrackSubscribedEvent);
event_setter!(on_track_published, TrackPublishedEvent);
event_setter!(on_track_subscription_failed, TrackSubscriptionFailedEvent);
}
}
pub mod participant {
use super::{EventHandler, TrackError};
use crate::room::id::TrackSid;
use crate::room::participant::remote_participant::RemoteParticipant;
use crate::room::publication::RemoteTrackPublication;
use crate::room::track::remote_track::RemoteTrackHandle;
use futures::future::Future;
use parking_lot::Mutex;
use std::sync::Arc;
#[derive(Clone)]
pub struct TrackPublishedEvent {
pub publication: RemoteTrackPublication,
pub participant: Arc<RemoteParticipant>,
}
#[derive(Clone)]
pub struct TrackSubscribedEvent {
pub track: RemoteTrackHandle,
pub publication: RemoteTrackPublication,
pub participant: Arc<RemoteParticipant>,
}
#[derive(Clone)]
pub struct TrackSubscriptionFailedEvent {
pub sid: TrackSid,
pub error: TrackError,
pub participant: Arc<RemoteParticipant>,
}
pub(crate) type TrackPublishedHandler = EventHandler<TrackPublishedEvent>;
pub(crate) type TrackSubscribedHandler = EventHandler<TrackSubscribedEvent>;
pub(crate) type TrackSubscriptionFailedHandler = EventHandler<TrackSubscriptionFailedEvent>;
#[derive(Default)]
pub struct ParticipantEvents {
pub(crate) on_track_published: Mutex<Option<TrackPublishedHandler>>,
pub(crate) on_track_subscribed: Mutex<Option<TrackSubscribedHandler>>,
pub(crate) on_track_subscription_failed: Mutex<Option<TrackSubscriptionFailedHandler>>,
}
impl ParticipantEvents {
event_setter!(on_track_published, TrackPublishedEvent);
event_setter!(on_track_subscribed, TrackSubscribedEvent);
event_setter!(on_track_subscription_failed, TrackSubscriptionFailedEvent);
}
}
crates/livekit-core/src/lib.rs
-1
View File
@@ -4,7 +4,6 @@ pub mod proto {
include!(concat!(env!("OUT_DIR"), "/livekit.rs"));
}
mod events;
mod rtc_engine;
mod signal_client;
crates/livekit-core/src/room/mod.rs
+88 -290
View File
@@ -1,317 +1,115 @@
use parking_lot::lock_api::RwLockUpgradableReadGuard;
use parking_lot::{Mutex, RwLock};
use std::collections::HashMap;
use std::sync::atomic::AtomicU8;
use self::participant::ConnectionQuality;
use self::room_session::{ConnectionState, RoomSession, SessionHandle};
use crate::proto::data_packet;
use crate::room::id::TrackSid;
use crate::room::participant::remote_participant::RemoteParticipant;
use crate::room::participant::Participant;
use crate::room::publication::RemoteTrackPublication;
use crate::room::publication::TrackPublication;
use crate::room::track::remote_track::RemoteTrackHandle;
use crate::rtc_engine::EngineError;
use std::fmt::Debug;
use std::sync::Arc;
use self::id::ParticipantSid;
use self::participant::local_participant::LocalParticipant;
use self::participant::remote_participant::RemoteParticipant;
use self::participant::ParticipantInternalTrait;
use self::participant::ParticipantTrait;
use crate::events::room::{
ParticipantConnectedEvent, ParticipantDisconnectedEvent, RoomEvents, TrackSubscribedEvent,
};
use crate::proto;
use crate::proto::participant_info;
use thiserror::Error;
use tracing::{debug, error};
use tokio::sync::mpsc;
use crate::rtc_engine::{EngineError, EngineEvent, EngineEvents, RTCEngine};
use crate::signal_client::SignalOptions;
pub use crate::rtc_engine::SimulateScenario;
pub mod id;
pub mod participant;
pub mod publication;
pub mod room_session;
pub mod track;
pub type RoomEvents = mpsc::UnboundedReceiver<RoomEvent>;
pub type RoomEmitter = mpsc::UnboundedSender<RoomEvent>;
pub type RoomResult<T> = Result<T, RoomError>;
#[derive(Error, Debug)]
pub enum RoomError {
#[error("internal RTCEngine failure")]
#[error("engine : {0}")]
Engine(#[from] EngineError),
#[error("internal Room failure")]
#[error("room failure: {0}")]
Internal(String),
}
type RoomResult<T> = Result<T, RoomError>;
#[derive(Error, Debug, Clone)]
pub enum TrackError {
#[error("could not find published track with sid: {0}")]
TrackNotFound(String),
}
#[derive(Clone, Debug)]
pub enum RoomEvent {
ParticipantConnected(Arc<RemoteParticipant>),
ParticipantDisconnected(Arc<RemoteParticipant>),
TrackSubscribed {
track: RemoteTrackHandle,
publication: RemoteTrackPublication,
participant: Arc<RemoteParticipant>,
},
TrackPublished {
publication: RemoteTrackPublication,
participant: Arc<RemoteParticipant>,
},
TrackUnpublished {
publication: RemoteTrackPublication,
participant: Arc<RemoteParticipant>,
},
TrackUnsubscribed {
track: RemoteTrackHandle,
publication: RemoteTrackPublication,
participant: Arc<RemoteParticipant>,
},
TrackSubscriptionFailed {
error: TrackError,
sid: TrackSid,
participant: Arc<RemoteParticipant>,
},
TrackMuted {
publication: TrackPublication,
participant: Participant,
},
TrackUnmuted {
publication: TrackPublication,
participant: Participant,
},
ActiveSpeakersChanged {
speakers: Vec<Participant>,
},
ConnectionQualityChanged {
quality: ConnectionQuality,
participant: Participant,
},
DataReceived {
payload: Arc<Vec<u8>>,
kind: data_packet::Kind,
participant: Arc<RemoteParticipant>,
},
ConnectionStateChanged(ConnectionState),
Connected,
Disconnected,
Reconnecting,
Reconnected,
}
#[derive(Debug)]
pub enum ConnectionState {
Disconnected,
Connecting,
Connected,
Reconnecting,
}
struct RoomInner {
state: AtomicU8, // ConnectionState
sid: Mutex<String>,
name: Mutex<String>,
participants: RwLock<HashMap<ParticipantSid, Arc<RemoteParticipant>>>,
rtc_engine: Arc<RTCEngine>,
local_participant: Arc<LocalParticipant>,
}
pub struct Room {
inner: Option<Arc<RoomInner>>,
events: Arc<RoomEvents>,
handle: SessionHandle,
}
impl Room {
pub fn new() -> Room {
Self {
inner: None,
events: Default::default(),
}
pub async fn connect(url: &str, token: &str) -> RoomResult<(Self, RoomEvents)> {
let (emitter, events) = mpsc::unbounded_channel();
let handle = SessionHandle::connect(emitter, url, token).await?;
Ok((Self { handle }, events))
}
pub async fn connect(&mut self, url: &str, token: &str) -> RoomResult<()> {
let (rtc_engine, engine_events) =
RTCEngine::connect(url, token, SignalOptions::default()).await?;
let rtc_engine = Arc::new(rtc_engine);
let join_response = rtc_engine.join_response();
let local_participant = Arc::new(LocalParticipant::new(
rtc_engine.clone(),
join_response.participant.unwrap().clone(),
));
let room_info = join_response.room.unwrap();
let inner = Arc::new(RoomInner {
state: AtomicU8::new(ConnectionState::Connecting as u8),
sid: Mutex::new(room_info.sid),
name: Mutex::new(room_info.name),
participants: Default::default(),
rtc_engine,
local_participant,
});
self.inner = Some(inner.clone());
// Add already connected participants
for pi in join_response.other_participants {
let p = Self::create_participant(inner.clone(), self.events.clone(), pi.clone());
p.update_info(pi).await;
}
tokio::spawn(Self::room_task(inner, self.events.clone(), engine_events));
Ok(())
pub async fn close(self) {
self.handle.close().await;
}
pub fn events(&self) -> Arc<RoomEvents> {
self.events.clone()
}
pub fn get_handle(&self) -> Option<RoomHandle> {
self.inner.as_ref().map(|inner| RoomHandle {
inner: inner.clone(),
})
}
async fn room_task(
room_inner: Arc<RoomInner>,
room_events: Arc<RoomEvents>,
mut engine_events: EngineEvents,
) {
while let Some(event) = engine_events.recv().await {
if let Err(err) =
Self::handle_event(room_inner.clone(), room_events.clone(), event).await
{
error!("failed to handle engine event: {:?}", err);
}
}
}
async fn handle_event(
room_inner: Arc<RoomInner>,
room_events: Arc<RoomEvents>,
event: EngineEvent,
) -> RoomResult<()> {
match event {
EngineEvent::ParticipantUpdate(update) => {
Self::handle_participant_update(room_inner.clone(), room_events.clone(), update)
.await
}
EngineEvent::AddTrack {
rtp_receiver,
streams,
} => {
if streams.is_empty() {
Err(RoomError::Internal(
"AddTrack event with empty streams".to_string(),
))?;
}
let first_stream_id = streams.first().unwrap().id();
let stream_id = unpack_stream_id(&first_stream_id);
if stream_id.is_none() {
Err(RoomError::Internal(format!(
"AddTrack event with invalid track_id: {:?}",
first_stream_id
)))?;
}
let (participant_sid, track_sid) = stream_id.unwrap();
let remote_participant =
Self::get_participant(room_inner.clone(), &participant_sid.to_string().into());
if let Some(remote_participant) = remote_participant {
remote_participant.add_subscribed_media_track(
track_sid.to_string().into(),
rtp_receiver.track(),
);
} else {
// The server should send participant updates before sending a new offer
// So this should not happen.
Err(RoomError::Internal(format!(
"AddTrack event with invalid participant_sid: {:?}",
participant_sid
)))?;
}
}
}
Ok(())
}
async fn handle_participant_update(
room_inner: Arc<RoomInner>,
room_events: Arc<RoomEvents>,
update: proto::ParticipantUpdate,
) {
for pi in update.participants {
if pi.sid == room_inner.local_participant.sid()
|| pi.identity == room_inner.local_participant.identity()
{
room_inner.local_participant.clone().update_info(pi).await;
continue;
}
let remote_participant =
Self::get_participant(room_inner.clone(), &pi.sid.clone().into());
if let Some(remote_participant) = remote_participant {
if pi.state == participant_info::State::Disconnected as i32 {
// Participant disconencted
Self::handle_participant_disconnect(
room_inner.clone(),
room_events.clone(),
remote_participant,
)
} else {
// Participant is already connected, update the informations
remote_participant.update_info(pi).await;
}
} else {
// Create a new participant and call OnConnect event
let remote_participant =
Self::create_participant(room_inner.clone(), room_events.clone(), pi);
let mut handler = room_events.on_participant_connected.lock();
if let Some(cb) = handler.as_mut() {
cb(ParticipantConnectedEvent {
room_handle: RoomHandle::from(room_inner.clone()),
participant: remote_participant.clone(),
});
}
}
}
}
fn handle_participant_disconnect(
room_inner: Arc<RoomInner>,
room_events: Arc<RoomEvents>,
remote_participant: Arc<RemoteParticipant>,
) {
room_inner
.participants
.write()
.remove(&remote_participant.sid());
// TODO(theomonnom): Unpublish all tracks
let mut handler = room_events.on_participant_disconnected.lock();
if let Some(cb) = handler.as_mut() {
cb(ParticipantDisconnectedEvent {
room_handle: RoomHandle::from(room_inner.clone()),
participant: remote_participant.clone(),
});
}
}
fn get_participant(
room_inner: Arc<RoomInner>,
sid: &ParticipantSid,
) -> Option<Arc<RemoteParticipant>> {
room_inner.participants.read().get(sid).cloned()
}
fn create_participant(
room_inner: Arc<RoomInner>,
room_events: Arc<RoomEvents>,
pi: proto::ParticipantInfo,
) -> Arc<RemoteParticipant> {
let p = Arc::new(RemoteParticipant::new(pi.clone()));
// Forward participantevents to room events
p.internal_events().on_track_subscribed({
let room_events = room_events.clone();
let room_inner = room_inner.clone();
move |event| {
let room_events = room_events.clone();
let room_inner = room_inner.clone();
async move {
if let Some(cb) = room_events.clone().on_track_subscribed.lock().as_mut() {
cb(TrackSubscribedEvent {
room_handle: RoomHandle::from(room_inner.clone()),
track: event.track,
participant: event.participant,
publication: event.publication,
})
.await;
}
}
}
});
room_inner
.participants
.write()
.insert(pi.sid.into(), p.clone());
p
}
}
#[derive(Clone)]
pub struct RoomHandle {
inner: Arc<RoomInner>,
}
impl RoomHandle {
fn from(room_inner: Arc<RoomInner>) -> Self {
Self { inner: room_inner }
}
pub fn sid(&self) -> String {
self.inner.sid.lock().clone()
}
pub fn name(&self) -> String {
self.inner.name.lock().clone()
}
pub fn local_participant(&self) -> Arc<LocalParticipant> {
self.inner.local_participant.clone()
}
}
fn unpack_stream_id(stream_id: &str) -> Option<(&str, &str)> {
let split: Vec<&str> = stream_id.split('|').collect();
if split.len() == 2 {
let participant_sid = split.get(0).unwrap();
let track_sid = split.get(1).unwrap();
Some((participant_sid, track_sid))
} else {
None
pub fn session(&self) -> RoomSession {
self.handle.session()
}
}
crates/livekit-core/src/room/participant/local_participant.rs
+37 -16
View File
@@ -1,22 +1,35 @@
use crate::proto::{data_packet, DataPacket, UserPacket};
use crate::room::participant::{impl_participant_trait, ParticipantShared, ParticipantInternalTrait};
use super::ConnectionQuality;
use crate::proto::{data_packet, DataPacket, ParticipantInfo, UserPacket};
use crate::room::id::{ParticipantIdentity, ParticipantSid, TrackSid};
use crate::room::participant::{
impl_participant_trait, ParticipantEvent, ParticipantInternalTrait, ParticipantShared,
ParticipantTrait,
};
use crate::room::publication::TrackPublication;
use crate::room::RoomError;
use crate::rtc_engine::RTCEngine;
use parking_lot::RwLockReadGuard;
use std::collections::HashMap;
use std::sync::atomic::Ordering;
use std::sync::Arc;
use tokio::sync::mpsc;
#[derive(Debug)]
pub struct LocalParticipant {
shared: ParticipantShared,
rtc_engine: Arc<RTCEngine>,
}
impl LocalParticipant {
pub(crate) fn new(rtc_engine: Arc<RTCEngine>, info: ParticipantInfo) -> Self {
pub(crate) fn new(
rtc_engine: Arc<RTCEngine>,
sid: ParticipantSid,
identity: ParticipantIdentity,
name: String,
metadata: String,
) -> Self {
Self {
shared: ParticipantShared::new(
info.sid.into(),
info.identity.into(),
info.name,
info.metadata,
),
shared: ParticipantShared::new(sid, identity, name, metadata),
rtc_engine,
}
}
@@ -29,7 +42,7 @@ impl LocalParticipant {
let data = DataPacket {
kind: kind as i32,
value: Some(data_packet::Value::User(UserPacket {
participant_sid: "".to_string(), /*self.sid().to_owned()*/
participant_sid: self.sid().to_string(),
payload: data.to_vec(),
destination_sids: vec![],
})),
@@ -40,15 +53,23 @@ impl LocalParticipant {
.await
.map_err(Into::into)
}
pub(crate) async fn update_info(self: Arc<Self>, info: ParticipantInfo) {
self.shared.update_info(info);
}
}
impl ParticipantInternalTrait for LocalParticipant {
fn internal_events(&self) -> Arc<ParticipantEvents> {
self.shared.internal_events.clone()
fn update_info(self: &Arc<Self>, info: ParticipantInfo, _emit_events: bool) {
self.shared.update_info(info);
}
fn set_speaking(&self, speaking: bool) {
self.shared.set_speaking(speaking);
}
fn set_audio_level(&self, level: f32) {
self.shared.set_audio_level(level);
}
fn set_connection_quality(&self, quality: ConnectionQuality) {
self.shared.set_connection_quality(quality);
}
}
crates/livekit-core/src/room/participant/mod.rs
+146 -34
View File
@@ -1,28 +1,103 @@
use crate::events::participant::ParticipantEvents;
use super::publication::RemoteTrackPublication;
use super::TrackError;
use crate::proto;
use crate::proto::ParticipantInfo;
use crate::room::id::{ParticipantIdentity, ParticipantSid, TrackSid};
use crate::room::participant::local_participant::LocalParticipant;
use crate::room::participant::remote_participant::RemoteParticipant;
use crate::room::publication::{TrackPublication, TrackPublicationTrait};
use futures_util::future::BoxFuture;
use crate::room::track::remote_track::RemoteTrackHandle;
use livekit_utils::enum_dispatch;
use parking_lot::{Mutex, RwLock};
use livekit_utils::observer::Dispatcher;
use parking_lot::{Mutex, RwLock, RwLockReadGuard};
use proto::data_packet;
use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU8, Ordering};
use std::sync::Arc;
use tokio::sync::mpsc;
pub mod local_participant;
pub mod remote_participant;
type OnTrackSubscribed = Box<dyn FnMut(ParticipantHandle) -> BoxFuture<'static, ()> + Send + Sync>;
#[derive(Debug, Clone)]
pub enum ParticipantEvent {
TrackPublished {
publication: RemoteTrackPublication,
},
TrackUnpublished {
publication: RemoteTrackPublication,
},
TrackSubscribed {
track: RemoteTrackHandle,
publication: RemoteTrackPublication,
},
TrackUnsubscribed {
track: RemoteTrackHandle,
publication: RemoteTrackPublication,
},
TrackSubscriptionFailed {
error: TrackError,
sid: TrackSid,
},
DataReceived {
payload: Arc<Vec<u8>>,
kind: data_packet::Kind,
},
SpeakingChanged {
speaking: bool,
},
TrackMuted {
publication: TrackPublication,
},
TrackUnmuted {
publication: TrackPublication,
},
ConnectionQualityChanged {
quality: ConnectionQuality,
},
}
#[derive(Debug, Clone, Copy, Eq, PartialEq)]
#[repr(u8)]
pub enum ConnectionQuality {
Unknown,
Excellent,
Good,
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)]
pub(super) struct ParticipantShared {
pub(super) events: Arc<ParticipantEvents>,
pub(super) internal_events: Arc<ParticipantEvents>,
pub(super) sid: Mutex<ParticipantSid>,
pub(super) identity: Mutex<ParticipantIdentity>,
pub(super) name: Mutex<String>,
pub(super) metadata: Mutex<String>,
pub(super) tracks: RwLock<HashMap<TrackSid, TrackPublication>>,
pub(super) speaking: AtomicBool,
pub(super) audio_level: AtomicU32,
pub(super) connection_quality: AtomicU8,
pub(super) dispatcher: Mutex<Dispatcher<ParticipantEvent>>,
}
impl ParticipantShared {
@@ -33,13 +108,15 @@ impl ParticipantShared {
metadata: String,
) -> Self {
Self {
events: Default::default(),
internal_events: Default::default(),
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),
dispatcher: Default::default(),
}
}
@@ -50,69 +127,84 @@ impl ParticipantShared {
*self.metadata.lock() = info.metadata; // TODO(theomonnom): callback MetadataChanged
}
pub(crate) fn set_speaking(&self, speaking: bool) {
self.speaking.store(speaking, Ordering::SeqCst);
}
pub(crate) fn set_audio_level(&self, audio_level: f32) {
self.audio_level
.store(audio_level.to_bits(), Ordering::SeqCst)
}
pub(crate) fn register_observer(&self) -> mpsc::UnboundedReceiver<ParticipantEvent> {
self.dispatcher.lock().register()
}
pub(crate) fn set_connection_quality(&self, quality: ConnectionQuality) {
self.connection_quality
.store(quality as u8, Ordering::SeqCst);
}
pub(crate) fn add_track_publication(&self, publication: TrackPublication) {
self.tracks.write().insert(publication.sid(), publication);
}
}
pub(crate) trait ParticipantInternalTrait {
fn internal_events(&self) -> Arc<ParticipantEvents>;
fn set_speaking(&self, speaking: bool);
fn set_audio_level(&self, level: f32);
fn set_connection_quality(&self, quality: ConnectionQuality);
fn update_info(self: &Arc<Self>, info: ParticipantInfo, emit_events: bool);
}
pub trait ParticipantTrait {
fn events(&self) -> Arc<ParticipantEvents>;
fn sid(&self) -> ParticipantSid;
fn identity(&self) -> ParticipantIdentity;
fn name(&self) -> String;
fn metadata(&self) -> String;
fn is_speaking(&self) -> bool;
fn audio_level(&self) -> f32;
fn connection_quality(&self) -> ConnectionQuality;
fn tracks(&self) -> RwLockReadGuard<HashMap<TrackSid, TrackPublication>>;
fn register_observer(&self) -> mpsc::UnboundedReceiver<ParticipantEvent>;
}
#[derive(Clone)]
pub enum ParticipantHandle {
#[derive(Debug, Clone)]
pub enum Participant {
Local(Arc<LocalParticipant>),
Remote(Arc<RemoteParticipant>),
}
impl ParticipantHandle {
// TODO(theomonnom): Add async support to wrap_variants ...
pub(crate) async fn update_info(&self, info: ParticipantInfo) {
match self {
Self::Local(inner) => inner.clone().update_info(info).await,
Self::Remote(inner) => inner.clone().update_info(info).await,
}
}
}
// TODO(theomonnom): Should I provide a WeakParticipant here ?
impl ParticipantInternalTrait for ParticipantHandle {
impl Participant {
enum_dispatch!(
[Local, Remote]
fnc!(internal_events, &Self, [], Arc<ParticipantEvents>);
fnc!(pub(crate), update_info, &Self, [info: ParticipantInfo, emit_events: bool], ());
fnc!(pub(crate), set_speaking, &Self, [speaking: bool], ());
fnc!(pub(crate), set_audio_level, &Self, [audio_level: f32], ());
fnc!(pub(crate), set_connection_quality, &Self, [quality: ConnectionQuality], ());
);
}
impl ParticipantTrait for ParticipantHandle {
impl ParticipantTrait for Participant {
enum_dispatch!(
[Local, Remote]
fnc!(events, &Self, [], Arc<ParticipantEvents>);
fnc!(sid, &Self, [], ParticipantSid);
fnc!(identity, &Self, [], ParticipantIdentity);
fnc!(name, &Self, [], String);
fnc!(metadata, &Self, [], String);
fnc!(is_speaking, &Self, [], bool);
fnc!(audio_level, &Self, [], f32);
fnc!(connection_quality, &Self, [], ConnectionQuality);
fnc!(tracks, &Self, [], RwLockReadGuard<HashMap<TrackSid, TrackPublication>>);
fnc!(register_observer, &Self, [], mpsc::UnboundedReceiver<ParticipantEvent>);
);
}
macro_rules! impl_participant_trait {
($x:ty) => {
use crate::events::participant::ParticipantEvents;
use crate::proto::ParticipantInfo;
use crate::room::id::{ParticipantIdentity, ParticipantSid};
use std::sync::Arc;
impl crate::room::participant::ParticipantTrait for $x {
fn events(&self) -> Arc<ParticipantEvents> {
self.shared.events.clone()
}
fn sid(&self) -> ParticipantSid {
self.shared.sid.lock().clone()
}
@@ -128,6 +220,26 @@ macro_rules! impl_participant_trait {
fn metadata(&self) -> String {
self.shared.metadata.lock().clone()
}
fn is_speaking(&self) -> bool {
self.shared.speaking.load(Ordering::SeqCst)
}
fn audio_level(&self) -> f32 {
f32::from_bits(self.shared.audio_level.load(Ordering::SeqCst))
}
fn connection_quality(&self) -> ConnectionQuality {
self.shared.connection_quality.load(Ordering::SeqCst).into()
}
fn tracks(&self) -> RwLockReadGuard<HashMap<TrackSid, TrackPublication>> {
self.shared.tracks.read()
}
fn register_observer(&self) -> mpsc::UnboundedReceiver<ParticipantEvent> {
self.shared.register_observer()
}
}
};
}
crates/livekit-core/src/room/participant/remote_participant.rs
+168 -151
View File
@@ -1,10 +1,9 @@
use crate::events::participant::{
TrackPublishedEvent, TrackSubscribedEvent, TrackSubscriptionFailedEvent,
};
use crate::events::TrackError;
use crate::room::id::TrackSid;
use super::ConnectionQuality;
use crate::proto::{data_packet, DataPacket, ParticipantInfo, UserPacket};
use crate::room::id::{ParticipantIdentity, ParticipantSid, TrackSid};
use crate::room::participant::{
impl_participant_trait, ParticipantInternalTrait, ParticipantShared,
impl_participant_trait, ParticipantEvent, ParticipantInternalTrait, ParticipantShared,
ParticipantTrait,
};
use crate::room::publication::{
RemoteTrackPublication, TrackPublication, TrackPublicationInternalTrait, TrackPublicationTrait,
@@ -12,141 +11,38 @@ use crate::room::publication::{
use crate::room::track::remote_audio_track::RemoteAudioTrack;
use crate::room::track::remote_track::RemoteTrackHandle;
use crate::room::track::remote_video_track::RemoteVideoTrack;
use crate::room::track::{TrackKind, TrackTrait, TrackHandle};
use crate::room::track::{TrackKind, TrackTrait};
use crate::room::TrackError;
use livekit_webrtc::media_stream::MediaStreamTrackHandle;
use parking_lot::RwLockReadGuard;
use std::collections::HashMap;
use std::collections::HashSet;
use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::time::Duration;
use tokio::time::{sleep, timeout};
use tracing::{info, error};
use super::ParticipantTrait;
use tokio::sync::mpsc;
use tokio::time::timeout;
use tracing::{debug, error, instrument, Level};
const ADD_TRACK_TIMEOUT: Duration = Duration::from_secs(5);
#[derive(Debug)]
pub struct RemoteParticipant {
shared: ParticipantShared,
}
impl RemoteParticipant {
pub(crate) fn new(info: ParticipantInfo) -> Self {
pub(crate) fn new(
sid: ParticipantSid,
identity: ParticipantIdentity,
name: String,
metadata: String,
) -> Self {
Self {
shared: ParticipantShared::new(
info.sid.into(),
info.identity.into(),
info.name,
info.metadata,
),
shared: ParticipantShared::new(sid, identity, name, metadata),
}
}
pub(crate) fn add_subscribed_media_track(
self: Arc<Self>,
sid: TrackSid,
media_track: MediaStreamTrackHandle,
) {
tokio::spawn(async move {
let wait_publication = {
let participant = self.clone();
let sid = sid.clone();
async move {
loop {
let publication = participant.get_track_publication(&sid);
if let Some(publication) = publication {
return publication;
}
sleep(Duration::from_millis(50)).await;
}
}
};
if let Ok(remote_publication) = timeout(ADD_TRACK_TIMEOUT, wait_publication).await {
let track = match remote_publication.kind() {
TrackKind::Audio => {
if let MediaStreamTrackHandle::Audio(rtc_track) = media_track {
let audio_track = RemoteAudioTrack::new(
remote_publication.sid().into(),
remote_publication.name(),
rtc_track,
);
RemoteTrackHandle::Audio(Arc::new(audio_track))
} else {
unreachable!();
}
}
TrackKind::Video => {
if let MediaStreamTrackHandle::Video(rtc_track) = media_track {
let video_track = RemoteVideoTrack::new(
remote_publication.sid().into(),
remote_publication.name(),
rtc_track,
);
RemoteTrackHandle::Video(Arc::new(video_track))
} else {
unreachable!()
}
}
_ => unreachable!(),
};
info!("starting track: {:?}", sid);
remote_publication.update_track(Some(track.clone().into()));
self.shared
.add_track_publication(TrackPublication::Remote(remote_publication.clone()));
track.start();
let event = TrackSubscribedEvent {
track,
publication: remote_publication,
participant: self.clone(),
};
if let Some(cb) = self
.shared
.internal_events
.on_track_subscribed
.lock()
.as_mut()
{
cb(event.clone()).await;
}
if let Some(cb) = self.shared.events.on_track_subscribed.lock().as_mut() {
cb(event).await;
}
} else {
error!("could not find published track with sid: {:?}", sid);
let event = TrackSubscriptionFailedEvent {
sid: sid.clone(),
error: TrackError::TrackNotFound(sid.clone().to_string()),
participant: self.clone(),
};
if let Some(cb) = self
.shared
.internal_events
.on_track_subscription_failed
.lock()
.as_mut()
{
cb(event.clone()).await;
}
if let Some(cb) = self
.shared
.events
.on_track_subscription_failed
.lock()
.as_mut()
{
cb(event).await;
}
}
});
}
fn get_track_publication(&self, sid: &TrackSid) -> Option<RemoteTrackPublication> {
self.shared.tracks.read().get(sid).map(|track| {
if let TrackPublication::Remote(remote) = track {
@@ -157,11 +53,129 @@ impl RemoteParticipant {
})
}
pub(crate) async fn update_info(self: Arc<Self>, info: ParticipantInfo) {
/// Called by the RoomSession when receiving data by 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: data_packet::Kind) {
self.shared
.dispatcher
.lock()
.dispatch(&ParticipantEvent::DataReceived {
payload: data,
kind,
});
}
#[instrument(level = Level::DEBUG)]
pub(crate) async fn add_subscribed_media_track(
self: Arc<Self>,
sid: TrackSid,
media_track: MediaStreamTrackHandle,
) {
let wait_publication = {
let participant = self.clone();
let sid = sid.clone();
async move {
loop {
let publication = participant.get_track_publication(&sid);
if let Some(publication) = publication {
return publication;
}
tokio::task::yield_now().await;
}
}
};
if let Ok(remote_publication) = timeout(ADD_TRACK_TIMEOUT, wait_publication).await {
let track = match remote_publication.kind() {
TrackKind::Audio => {
if let MediaStreamTrackHandle::Audio(rtc_track) = media_track {
let audio_track = RemoteAudioTrack::new(
remote_publication.sid().into(),
remote_publication.name(),
rtc_track,
);
RemoteTrackHandle::Audio(Arc::new(audio_track))
} else {
unreachable!();
}
}
TrackKind::Video => {
if let MediaStreamTrackHandle::Video(rtc_track) = media_track {
let video_track = RemoteVideoTrack::new(
remote_publication.sid().into(),
remote_publication.name(),
rtc_track,
);
RemoteTrackHandle::Video(Arc::new(video_track))
} else {
unreachable!()
}
}
_ => unreachable!(),
};
debug!("starting track: {:?}", sid);
remote_publication.update_track(Some(track.clone().into()));
self.shared
.add_track_publication(TrackPublication::Remote(remote_publication.clone()));
track.start();
self.shared
.dispatcher
.lock()
.dispatch(&ParticipantEvent::TrackSubscribed {
track: track,
publication: remote_publication,
});
} else {
error!("could not find published track with sid: {:?}", sid);
self.shared
.dispatcher
.lock()
.dispatch(&ParticipantEvent::TrackSubscriptionFailed {
sid: sid.clone(),
error: TrackError::TrackNotFound(sid.clone().to_string()),
});
}
}
pub(crate) fn unpublish_track(self: &Arc<Self>, sid: &TrackSid, emit_events: bool) {
if let Some(publication) = self.get_track_publication(sid) {
// Unsubscribe to the track if needed
if let Some(track) = publication.track() {
track.stop();
self.shared
.dispatcher
.lock()
.dispatch(&ParticipantEvent::TrackUnsubscribed {
track: track.clone(),
publication: publication.clone(),
});
}
if emit_events {
self.shared
.dispatcher
.lock()
.dispatch(&ParticipantEvent::TrackUnpublished {
publication: publication.clone(),
});
}
publication.update_track(None);
}
}
}
impl ParticipantInternalTrait for RemoteParticipant {
fn update_info(self: &Arc<Self>, info: ParticipantInfo, emit_events: bool) {
self.shared.update_info(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());
@@ -170,35 +184,38 @@ impl RemoteParticipant {
self.shared
.add_track_publication(TrackPublication::Remote(publication.clone()));
// This is a new track, fire publish events
let event = TrackPublishedEvent {
participant: self.clone(),
publication: publication.clone(),
};
if let Some(cb) = self
.shared
.internal_events
.on_track_published
.lock()
.as_mut()
{
cb(event.clone()).await;
}
if let Some(cb) = self.shared.events.on_track_published.lock().as_mut() {
cb(event).await;
// This is a new track, dispatch publish event
if emit_events {
self.shared
.dispatcher
.lock()
.dispatch(&ParticipantEvent::TrackPublished { publication });
}
}
valid_tracks.insert(track.sid.into());
}
}
}
impl ParticipantInternalTrait for RemoteParticipant {
fn internal_events(&self) -> Arc<ParticipantEvents> {
self.shared.internal_events.clone()
// remove tracks that are no longer valid
for (sid, _) in self.shared.tracks.read().iter() {
if valid_tracks.contains(sid) {
continue;
}
self.unpublish_track(sid, emit_events);
}
}
fn set_speaking(&self, speaking: bool) {
self.shared.set_speaking(speaking);
}
fn set_audio_level(&self, level: f32) {
self.shared.set_audio_level(level);
}
fn set_connection_quality(&self, quality: ConnectionQuality) {
self.shared.set_connection_quality(quality);
}
}
crates/livekit-core/src/room/publication/mod.rs
+94 -18
View File
@@ -4,13 +4,15 @@ use crate::room::id::ParticipantSid;
use crate::room::id::TrackSid;
use crate::room::track::local_track::LocalTrackHandle;
use crate::room::track::remote_track::RemoteTrackHandle;
use crate::room::track::{TrackHandle, TrackKind, TrackSource};
use crate::room::track::{TrackHandle, TrackKind, TrackSource, TrackTrait};
use livekit_utils::enum_dispatch;
use livekit_utils::observer::Dispatcher;
use parking_lot::Mutex;
use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
use std::sync::Arc;
use tokio::sync::{mpsc, oneshot};
use super::track::TrackDimension;
use super::track::{TrackDimension, TrackEvent};
pub(crate) trait TrackPublicationInternalTrait {
fn update_track(&self, track: Option<TrackHandle>);
@@ -22,9 +24,11 @@ pub trait TrackPublicationTrait {
fn sid(&self) -> TrackSid;
fn kind(&self) -> TrackKind;
fn source(&self) -> TrackSource;
fn muted(&self) -> bool;
fn simulcasted(&self) -> bool;
}
#[derive(Debug)]
pub(super) struct TrackPublicationShared {
pub(super) track: Mutex<Option<TrackHandle>>,
pub(super) name: Mutex<String>,
@@ -34,7 +38,10 @@ pub(super) struct TrackPublicationShared {
pub(super) simulcasted: AtomicBool,
pub(super) dimension: Mutex<TrackDimension>,
pub(super) mime_type: Mutex<String>,
pub(super) participant: ParticipantSid, // TODO(theomonnom) Use WeakParticipant instead
pub(super) muted: AtomicBool,
pub(super) participant: ParticipantSid,
pub(super) dispatcher: Mutex<Dispatcher<TrackEvent>>,
pub(super) close_sender: Mutex<Option<oneshot::Sender<()>>>,
}
impl TrackPublicationShared {
@@ -54,13 +61,56 @@ impl TrackPublicationShared {
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_sender: Default::default(),
participant,
})
}
pub fn update_track(self: &Arc<Self>, track: Option<TrackHandle>) {
let mut old_track = self.track.lock();
if let Some(close_sender) = self.close_sender.lock().take() {
let _ = close_sender.send(());
}
*old_track = track.clone();
if let Some(track) = track {
let (close_sender, close_receiver) = oneshot::channel();
self.close_sender.lock().replace(close_sender);
let track_receiver = track.register_observer();
tokio::spawn(
self.clone()
.publication_task(close_receiver, track_receiver),
);
}
}
/// Task used to forward TrackHandle's events to the TrackPublications's dispatcher
async fn publication_task(
self: Arc<Self>,
mut close_receiver: oneshot::Receiver<()>,
mut track_receiver: mpsc::UnboundedReceiver<TrackEvent>,
) {
loop {
tokio::select! {
Some(event) = track_receiver.recv() => {
self.dispatcher.lock().dispatch(&event);
}
_ = &mut close_receiver => {
break;
}
}
}
}
pub fn update_info(&self, info: 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::from(TrackType::from_i32(info.r#type).unwrap()) as u8,
Ordering::SeqCst,
@@ -70,12 +120,23 @@ impl TrackPublicationShared {
Ordering::SeqCst,
);
self.simulcasted.store(info.simulcast, Ordering::SeqCst);
*self.dimension.lock() = TrackDimension(info.width, info.height);
*self.mime_type.lock() = info.mime_type;
self.muted.store(info.muted, Ordering::SeqCst);
if let Some(track) = self.track.lock().as_ref() {
track.set_muted(info.muted);
}
}
}
#[derive(Clone)]
impl Drop for TrackPublicationShared {
fn drop(&mut self) {
if let Some(close_sender) = self.close_sender.lock().take() {
let _ = close_sender.send(());
}
}
}
#[derive(Clone, Debug)]
pub enum TrackPublication {
Local(LocalTrackPublication),
Remote(RemoteTrackPublication),
@@ -106,22 +167,13 @@ impl TrackPublicationTrait for TrackPublication {
fnc!(name, &Self, [], String);
fnc!(kind, &Self, [], TrackKind);
fnc!(source, &Self, [], TrackSource);
fnc!(muted, &Self, [], bool);
fnc!(simulcasted, &Self, [], bool);
);
}
macro_rules! impl_publication_trait {
($x:ident) => {
impl TrackPublicationInternalTrait for $x {
fn update_track(&self, track: Option<TrackHandle>) {
*self.shared.track.lock() = track;
}
fn update_info(&self, info: TrackInfo) {
self.shared.update_info(info);
}
}
impl TrackPublicationTrait for $x {
fn name(&self) -> String {
self.shared.name.lock().clone()
@@ -142,11 +194,15 @@ macro_rules! impl_publication_trait {
fn simulcasted(&self) -> bool {
self.shared.simulcasted.load(Ordering::SeqCst)
}
fn muted(&self) -> bool {
self.shared.muted.load(Ordering::SeqCst)
}
}
};
}
#[derive(Clone)]
#[derive(Clone, Debug)]
pub struct LocalTrackPublication {
shared: Arc<TrackPublicationShared>,
}
@@ -161,7 +217,17 @@ impl LocalTrackPublication {
}
}
#[derive(Clone)]
impl TrackPublicationInternalTrait for LocalTrackPublication {
fn update_track(&self, track: Option<TrackHandle>) {
self.shared.update_track(track);
}
fn update_info(&self, info: TrackInfo) {
self.shared.update_info(info);
}
}
#[derive(Clone, Debug)]
pub struct RemoteTrackPublication {
shared: Arc<TrackPublicationShared>,
}
@@ -182,5 +248,15 @@ impl RemoteTrackPublication {
}
}
impl TrackPublicationInternalTrait for RemoteTrackPublication {
fn update_track(&self, track: Option<TrackHandle>) {
self.shared.update_track(track);
}
fn update_info(&self, info: TrackInfo) {
self.shared.update_info(info);
}
}
impl_publication_trait!(LocalTrackPublication);
impl_publication_trait!(RemoteTrackPublication);
crates/livekit-core/src/room/room_session.rs
+571
View File
@@ -0,0 +1,571 @@
use super::id::{ParticipantIdentity, ParticipantSid};
use super::participant::local_participant::LocalParticipant;
use super::participant::remote_participant::RemoteParticipant;
use super::participant::{ConnectionQuality, Participant, ParticipantEvent};
use super::participant::{ParticipantInternalTrait, ParticipantTrait};
use super::{RoomEmitter, RoomError, RoomEvent, RoomResult, SimulateScenario};
use crate::proto::{self, participant_info, SpeakerInfo};
use crate::rtc_engine::{EngineEvent, EngineEvents, EngineResult, RTCEngine};
use crate::signal_client::SignalOptions;
use parking_lot::{Mutex, RwLock};
use std::collections::HashMap;
use std::sync::atomic::{AtomicU8, Ordering};
use std::sync::Arc;
use tokio::sync::mpsc;
use tokio::sync::oneshot;
use tokio::task::JoinHandle;
use tracing::{error, instrument, Level};
#[derive(Debug, Clone, Copy, Eq, PartialEq)]
pub enum ConnectionState {
Disconnected,
Connected,
Reconnecting,
Unknown,
}
impl From<u8> for ConnectionState {
fn from(value: u8) -> Self {
match value {
0 => ConnectionState::Disconnected,
1 => ConnectionState::Connected,
2 => ConnectionState::Reconnecting,
_ => ConnectionState::Unknown,
}
}
}
/// Internal representation of a RoomSession
#[derive(Debug)]
struct SessionInner {
state: AtomicU8, // ConnectionState
sid: Mutex<String>,
name: Mutex<String>,
participants: RwLock<HashMap<ParticipantSid, Arc<RemoteParticipant>>>,
participants_tasks: RwLock<HashMap<ParticipantSid, (JoinHandle<()>, oneshot::Sender<()>)>>,
active_speakers: RwLock<Vec<Participant>>,
rtc_engine: Arc<RTCEngine>,
local_participant: Arc<LocalParticipant>,
room_emitter: RoomEmitter,
}
#[derive(Debug)]
pub(crate) struct SessionHandle {
session: RoomSession,
session_task: JoinHandle<()>,
close_emitter: oneshot::Sender<()>,
}
/// RoomSession represents a connection to a room.
/// It can be cloned and shared across threads.
#[derive(Debug, Clone)]
pub struct RoomSession {
inner: Arc<SessionInner>,
}
impl SessionHandle {
pub async fn connect(room_emitter: RoomEmitter, url: &str, token: &str) -> RoomResult<Self> {
let (rtc_engine, engine_events) = RTCEngine::new();
let rtc_engine = Arc::new(rtc_engine);
rtc_engine
.connect(url, token, SignalOptions::default())
.await?;
let join_response = rtc_engine.join_response().unwrap();
let pi = join_response.participant.unwrap().clone();
let local_participant = Arc::new(LocalParticipant::new(
rtc_engine.clone(),
pi.sid.into(),
pi.identity.into(),
pi.name,
pi.metadata,
));
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),
name: Mutex::new(room_info.name),
participants: Default::default(),
participants_tasks: Default::default(),
active_speakers: Default::default(),
rtc_engine,
local_participant,
room_emitter,
});
for pi in join_response.other_participants {
let participant = {
let pi = pi.clone();
inner.create_participant(pi.sid.into(), pi.identity.into(), pi.name, pi.metadata)
};
participant.update_info(pi.clone(), false);
}
let (close_emitter, close_receiver) = oneshot::channel();
let session_task = tokio::spawn(inner.clone().room_task(engine_events, close_receiver));
inner.update_connection_state(ConnectionState::Connected);
let session = Self {
session: RoomSession::from(inner),
session_task,
close_emitter,
};
Ok(session)
}
pub async fn close(self) {
self.session.inner.close().await;
let _ = self.close_emitter.send(());
let _ = self.session_task.await;
}
pub fn session(&self) -> RoomSession {
self.session.clone()
}
}
impl RoomSession {
fn from(inner: Arc<SessionInner>) -> Self {
Self { inner }
}
pub fn sid(&self) -> String {
self.inner.sid.lock().clone()
}
pub fn name(&self) -> String {
self.inner.name.lock().clone()
}
pub fn local_participant(&self) -> Arc<LocalParticipant> {
self.inner.local_participant.clone()
}
pub fn connection_state(&self) -> ConnectionState {
self.inner.state.load(Ordering::Acquire).try_into().unwrap()
}
pub fn participants(&self) -> &RwLock<HashMap<ParticipantSid, Arc<RemoteParticipant>>> {
&self.inner.participants
}
pub async fn simulate_scenario(&self, scenario: SimulateScenario) -> EngineResult<()> {
self.inner.rtc_engine.simulate_scenario(scenario).await
}
}
impl SessionInner {
#[instrument(level = Level::DEBUG)]
async fn room_task(
self: Arc<Self>,
mut engine_events: EngineEvents,
mut close_receiver: oneshot::Receiver<()>,
) {
loop {
tokio::select! {
res = engine_events.recv() => {
match res {
Some(event) => {
if let Err(err) = self.on_engine_event(event).await {
error!("failed to handle engine event: {:?}", err);
}
},
_ => panic!("engine_events has been closed unexpectedly")
};
},
_ = &mut close_receiver => {
break;
}
}
}
}
/// Listen to the Participant events and forward them to the Room Dispatcher
#[instrument(level = Level::DEBUG)]
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() => {
match res {
Some(event) => {
if let Err(err) = self.on_participant_event(&participant, event).await {
error!("failed to handle participant event for {:?}: {:?}", participant.sid(), err);
}
},
_ => panic!("engine_events has been closed unexpectedly")
};
},
_ = &mut close_rx => {
break;
},
}
}
}
#[instrument(level = Level::DEBUG)]
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 } => {
let _ = self.room_emitter.send(RoomEvent::TrackPublished {
participant: remote_participant.clone(),
publication,
});
}
ParticipantEvent::TrackUnpublished { publication } => {
let _ = self.room_emitter.send(RoomEvent::TrackUnpublished {
participant: remote_participant.clone(),
publication,
});
}
ParticipantEvent::TrackSubscribed { track, publication } => {
let _ = self.room_emitter.send(RoomEvent::TrackSubscribed {
participant: remote_participant.clone(),
track,
publication,
});
}
ParticipantEvent::TrackUnsubscribed { track, publication } => {
let _ = self.room_emitter.send(RoomEvent::TrackUnsubscribed {
participant: remote_participant.clone(),
track,
publication,
});
}
_ => {}
};
}
Ok(())
}
#[instrument(level = Level::DEBUG)]
async fn on_engine_event(self: &Arc<Self>, event: EngineEvent) -> RoomResult<()> {
match event {
EngineEvent::ParticipantUpdate(update) => self.handle_participant_update(update),
EngineEvent::MediaTrack {
track,
stream,
receiver: _,
} => {
let stream_id = stream.id();
let lk_stream_id = unpack_stream_id(&stream_id);
if lk_stream_id.is_none() {
Err(RoomError::Internal(format!(
"MediaTrack event with invalid track_id: {:?}",
&stream_id
)))?;
}
let (participant_sid, track_sid) = lk_stream_id.unwrap();
let track_sid = track_sid.to_owned().into();
let remote_participant = self.get_participant(&participant_sid.to_string().into());
if let Some(remote_participant) = remote_participant {
tokio::spawn(async move {
remote_participant
.add_subscribed_media_track(track_sid, track)
.await;
});
} else {
// The server should send participant updates before sending a new offer
// So this should never happen.
Err(RoomError::Internal(format!(
"AddTrack event with invalid participant_sid: {:?}",
participant_sid
)))?;
}
}
EngineEvent::Resuming => {
if self.update_connection_state(ConnectionState::Reconnecting) {
let _ = self.room_emitter.send(RoomEvent::Reconnecting);
}
}
EngineEvent::Resumed => {
self.update_connection_state(ConnectionState::Connected);
let _ = self.room_emitter.send(RoomEvent::Reconnected);
// TODO(theomonnom): Update subscriptions settings
// TODO(theomonnom): Send sync state
}
EngineEvent::Restarting => self.handle_restarting(),
EngineEvent::Restarted => self.handle_restarted(),
EngineEvent::Disconnected => self.handle_disconnected(),
EngineEvent::Data {
payload,
kind,
participant_sid,
} => {
let payload = Arc::new(payload);
if let Some(participant) = self.get_participant(&participant_sid.into()) {
let _ = self.room_emitter.send(RoomEvent::DataReceived {
payload: payload.clone(),
kind,
participant: participant.clone(),
});
participant.on_data_received(payload, kind);
}
}
EngineEvent::SpeakersChanged { speakers } => self.handle_speakers_changed(speakers),
EngineEvent::ConnectionQuality { updates } => {
self.handle_connection_quality_update(updates)
}
}
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)]
fn update_connection_state(&self, state: ConnectionState) -> bool {
let old_state = self.state.load(Ordering::Acquire);
if old_state == state as u8 {
return false;
}
self.state.store(state as u8, Ordering::Release);
let _ = self
.room_emitter
.send(RoomEvent::ConnectionStateChanged(state));
return true;
}
/// 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>, update: proto::ParticipantUpdate) {
for pi in update.participants {
if pi.sid == self.local_participant.sid()
|| pi.identity == self.local_participant.identity()
{
self.local_participant.clone().update_info(pi, true);
continue;
}
let remote_participant = self.get_participant(&pi.sid.clone().into());
if let Some(remote_participant) = remote_participant {
if pi.state == participant_info::State::Disconnected as i32 {
// Participant disconnected
self.clone()
.handle_participant_disconnect(remote_participant)
} else {
// Participant is already connected, update the it
remote_participant.update_info(pi.clone(), true);
}
} else {
// Create a new participant
let remote_participant = {
let pi = pi.clone();
self.create_participant(pi.sid.into(), pi.identity.into(), pi.name, pi.metadata)
};
let _ = self
.room_emitter
.send(RoomEvent::ParticipantConnected(remote_participant.clone()));
remote_participant.update_info(pi.clone(), true);
}
}
}
/// 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<SpeakerInfo>) {
let mut speakers = Vec::new();
for speaker in speakers_info {
let participant = {
if speaker.sid == self.local_participant.sid() {
Participant::Local(self.local_participant.clone())
} else {
if let Some(participant) = self.get_participant(&speaker.sid.into()) {
Participant::Remote(participant)
} else {
continue;
}
}
};
participant.set_speaking(speaker.active);
participant.set_audio_level(speaker.level);
if speaker.active {
speakers.push(participant);
}
}
speakers.sort_by(|a, b| a.audio_level().partial_cmp(&b.audio_level()).unwrap());
*self.active_speakers.write() = speakers.clone();
let _ = self
.room_emitter
.send(RoomEvent::ActiveSpeakersChanged { speakers });
}
/// 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 = {
if update.participant_sid == self.local_participant.sid() {
Participant::Local(self.local_participant.clone())
} else {
if let Some(participant) = self.get_participant(&update.participant_sid.into())
{
Participant::Remote(participant)
} else {
continue;
}
}
};
let quality: ConnectionQuality = proto::ConnectionQuality::from_i32(update.quality)
.unwrap()
.into();
participant.set_connection_quality(quality);
let _ = self.room_emitter.send(RoomEvent::ConnectionQualityChanged {
participant,
quality,
});
}
}
#[instrument(level = Level::DEBUG)]
fn handle_restarting(self: &Arc<Self>) {
// Remove existing participants/subscriptions on full reconnect
for (_, participant) in self.participants.read().iter() {
self.clone()
.handle_participant_disconnect(participant.clone());
}
if self.update_connection_state(ConnectionState::Reconnecting) {
let _ = self.room_emitter.send(RoomEvent::Reconnecting);
}
}
#[instrument(level = Level::DEBUG)]
fn handle_restarted(self: &Arc<Self>) {
// Full reconnect succeeded!
let join_response = self.rtc_engine.join_response().unwrap();
self.update_connection_state(ConnectionState::Connected);
let _ = self.room_emitter.send(RoomEvent::Reconnected);
if let Some(pi) = join_response.participant {
self.local_participant.update_info(pi, true); // The sid may have changed
}
self.handle_participant_update(proto::ParticipantUpdate {
participants: join_response.other_participants,
});
// TODO(theomonnom): unpublish & republish tracks
}
#[instrument(level = Level::DEBUG)]
fn handle_disconnected(&self) {
if self.state.load(Ordering::Acquire) == ConnectionState::Disconnected as u8 {
return;
}
self.update_connection_state(ConnectionState::Disconnected);
let _ = self.room_emitter.send(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,
identity: ParticipantIdentity,
name: String,
metadata: String,
) -> Arc<RemoteParticipant> {
let participant = Arc::new(RemoteParticipant::new(
sid.clone(),
identity,
name,
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));
self.participants.write().insert(sid, participant.clone());
participant
}
/// A participant has disconnected
/// Cleanup the participant and emit an event
#[instrument(level = Level::DEBUG)]
fn handle_participant_disconnect(self: Arc<Self>, remote_participant: Arc<RemoteParticipant>) {
tokio::spawn(async move {
for (sid, _) in &*remote_participant.tracks() {
remote_participant.unpublish_track(&sid, true);
}
// Close the participant task
if let Some((task, close_tx)) = self
.participants_tasks
.write()
.remove(&remote_participant.sid())
{
let _ = close_tx.send(());
let _ = task.await;
}
self.participants.write().remove(&remote_participant.sid());
let _ = self.room_emitter.send(RoomEvent::ParticipantDisconnected(
remote_participant.clone(),
));
});
}
fn get_participant(&self, sid: &ParticipantSid) -> Option<Arc<RemoteParticipant>> {
self.participants.read().get(sid).cloned()
}
}
fn unpack_stream_id(stream_id: &str) -> Option<(&str, &str)> {
let split: Vec<&str> = stream_id.split('|').collect();
if split.len() == 2 {
let participant_sid = split.get(0).unwrap();
let track_sid = split.get(1).unwrap();
Some((participant_sid, track_sid))
} else {
None
}
}
crates/livekit-core/src/room/track/events.rs
-1
View File
@@ -1 +0,0 @@
pub struct TrackEvents {}
crates/livekit-core/src/room/track/local_audio_track.rs
+1
View File
@@ -1,5 +1,6 @@
use crate::room::track::{impl_track_trait, TrackShared};
#[derive(Debug)]
pub struct LocalAudioTrack {
shared: TrackShared,
}
crates/livekit-core/src/room/track/local_video_track.rs
+1
View File
@@ -1,5 +1,6 @@
use crate::room::track::{impl_track_trait, TrackShared};
#[derive(Debug)]
pub struct LocalVideoTrack {
shared: TrackShared,
}
crates/livekit-core/src/room/track/mod.rs
+50 -5
View File
@@ -5,13 +5,14 @@ use crate::room::track::local_video_track::LocalVideoTrack;
use crate::room::track::remote_audio_track::RemoteAudioTrack;
use crate::room::track::remote_video_track::RemoteVideoTrack;
use livekit_utils::enum_dispatch;
use livekit_utils::observer::Dispatcher;
use livekit_webrtc::media_stream::{MediaStreamTrackHandle, MediaStreamTrackTrait};
use parking_lot::Mutex;
use std::sync::atomic::AtomicU8;
use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
use std::sync::Arc;
use tokio::sync::mpsc;
pub mod audio_track;
pub mod events;
pub mod local_audio_track;
pub mod local_track;
pub mod local_video_track;
@@ -97,6 +98,7 @@ impl From<ProtoTrackSource> for TrackSource {
}
}
#[derive(Clone, Copy, Debug)]
pub struct TrackDimension(pub u32, pub u32);
pub trait TrackTrait {
@@ -106,14 +108,25 @@ pub trait TrackTrait {
fn stream_state(&self) -> StreamState;
fn start(&self);
fn stop(&self);
fn register_observer(&self) -> mpsc::UnboundedReceiver<TrackEvent>;
fn set_muted(&self, muted: bool);
}
#[derive(Debug, Clone)]
pub enum TrackEvent {
Mute,
Unmute,
}
#[derive(Debug)]
pub(super) struct TrackShared {
pub(super) sid: Mutex<TrackSid>,
pub(super) name: Mutex<String>,
pub(super) kind: AtomicU8, // TrackKind
pub(super) stream_state: AtomicU8, // StreamState
pub(super) muted: AtomicBool,
pub(super) rtc_track: MediaStreamTrackHandle,
pub(super) dispatcher: Mutex<Dispatcher<TrackEvent>>,
}
impl TrackShared {
@@ -128,7 +141,9 @@ impl TrackShared {
name: Mutex::new(name),
kind: AtomicU8::new(kind as u8),
stream_state: AtomicU8::new(StreamState::Active as u8),
muted: AtomicBool::new(false),
rtc_track,
dispatcher: Default::default(),
}
}
@@ -139,9 +154,28 @@ impl TrackShared {
pub(crate) fn stop(&self) {
self.rtc_track.set_enabled(false);
}
pub(crate) fn set_muted(&self, muted: bool) {
if self.muted.load(Ordering::SeqCst) == muted {
return;
}
self.muted.store(muted, Ordering::SeqCst);
self.rtc_track.set_enabled(!muted);
self.dispatcher.lock().dispatch(if muted {
&TrackEvent::Mute
} else {
&TrackEvent::Unmute
});
}
pub(crate) fn register_observer(&self) -> mpsc::UnboundedReceiver<TrackEvent> {
self.dispatcher.lock().register()
}
}
#[derive(Clone)]
#[derive(Clone, Debug)]
pub enum TrackHandle {
LocalVideo(Arc<LocalVideoTrack>),
LocalAudio(Arc<LocalAudioTrack>),
@@ -158,6 +192,8 @@ impl TrackTrait for TrackHandle {
fnc!(stream_state, &Self, [], StreamState);
fnc!(start, &Self, [], ());
fnc!(stop, &Self, [], ());
fnc!(register_observer, &Self, [], mpsc::UnboundedReceiver<TrackEvent>);
fnc!(set_muted, &Self, [muted: bool], ());
);
}
@@ -177,9 +213,10 @@ impl TrackHandle {
macro_rules! impl_track_trait {
($x:ident) => {
use crate::room::id::TrackSid;
use crate::room::track::{StreamState, TrackKind, TrackTrait};
use std::sync::atomic::Ordering;
use tokio::sync::mpsc;
use $crate::room::id::TrackSid;
use $crate::room::track::{StreamState, TrackEvent, TrackKind, TrackTrait};
impl TrackTrait for $x {
fn sid(&self) -> TrackSid {
@@ -205,6 +242,14 @@ macro_rules! impl_track_trait {
fn stop(&self) {
self.shared.stop();
}
fn register_observer(&self) -> mpsc::UnboundedReceiver<TrackEvent> {
self.shared.register_observer()
}
fn set_muted(&self, muted: bool) {
self.shared.set_muted(muted);
}
}
};
}
crates/livekit-core/src/room/track/remote_audio_track.rs
+1
View File
@@ -2,6 +2,7 @@ use crate::room::track::{impl_track_trait, TrackShared};
use livekit_webrtc::media_stream::{AudioTrack, MediaStreamTrackHandle};
use std::sync::Arc;
#[derive(Debug)]
pub struct RemoteAudioTrack {
shared: TrackShared,
}
crates/livekit-core/src/room/track/remote_track.rs
+5 -3
View File
@@ -1,15 +1,15 @@
use std::sync::Arc;
use super::{StreamState, TrackKind};
use crate::room::id::TrackSid;
use crate::room::track::remote_audio_track::RemoteAudioTrack;
use crate::room::track::remote_video_track::RemoteVideoTrack;
use crate::room::track::TrackHandle;
use crate::room::track::{TrackHandle, TrackEvent};
use tokio::sync::mpsc;
use livekit_utils::enum_dispatch;
use super::TrackTrait;
#[derive(Clone)]
#[derive(Clone, Debug)]
pub enum RemoteTrackHandle {
Audio(Arc<RemoteAudioTrack>),
Video(Arc<RemoteVideoTrack>),
@@ -24,6 +24,8 @@ impl TrackTrait for RemoteTrackHandle {
fnc!(stream_state, &Self, [], StreamState);
fnc!(start, &Self, [], ());
fnc!(stop, &Self, [], ());
fnc!(register_observer, &Self, [], mpsc::UnboundedReceiver<TrackEvent>);
fnc!(set_muted, &Self, [muted: bool], ());
);
}
crates/livekit-core/src/room/track/remote_video_track.rs
+1
View File
@@ -3,6 +3,7 @@ use std::sync::Arc;
use crate::room::track::{impl_track_trait, TrackShared};
#[derive(Debug)]
pub struct RemoteVideoTrack {
shared: TrackShared,
}
crates/livekit-core/src/rtc_engine/lk_runtime.rs
+5 -5
View File
@@ -1,6 +1,6 @@
use std::fmt::{Debug, Formatter};
use tracing::{event, Level};
use tracing::trace;
use livekit_webrtc::peer_connection_factory::PeerConnectionFactory;
use livekit_webrtc::webrtc::RTCRuntime;
@@ -19,9 +19,9 @@ impl Debug for LKRuntime {
}
}
impl LKRuntime {
pub fn new() -> Self {
event!(Level::TRACE, "LKRuntime::new()");
impl Default for LKRuntime {
fn default() -> Self {
trace!("LKRuntime::default()");
let rtc_runtime = RTCRuntime::new();
Self {
pc_factory: PeerConnectionFactory::new(rtc_runtime.clone()),
@@ -32,6 +32,6 @@ impl LKRuntime {
impl Drop for LKRuntime {
fn drop(&mut self) {
event!(Level::TRACE, "LKRuntime::drop()");
trace!("LKRuntime::drop()");
}
}
crates/livekit-core/src/rtc_engine/mod.rs
+392 -553
View File
File diff suppressed because it is too large Load Diff
crates/livekit-core/src/rtc_engine/pc_transport.rs
+47 -12
View File
@@ -11,6 +11,8 @@ use livekit_webrtc::peer_connection::{
};
use livekit_webrtc::rtc_error::RTCError;
use crate::proto::SignalTarget;
const NEGOTIATION_FREQUENCY: Duration = Duration::from_millis(150);
pub type OnOfferHandler = Box<
@@ -20,11 +22,12 @@ pub type OnOfferHandler = Box<
>;
pub struct PCTransport {
signal_target: SignalTarget,
peer_connection: PeerConnection,
pending_candidates: Vec<IceCandidate>,
on_offer_handler: Option<OnOfferHandler>,
restarting_ice: bool,
renegotiate: bool,
restarting_ice: bool,
}
impl Debug for PCTransport {
@@ -34,8 +37,9 @@ impl Debug for PCTransport {
}
impl PCTransport {
pub fn new(peer_connection: PeerConnection) -> Self {
pub fn new(peer_connection: PeerConnection, signal_target: SignalTarget) -> Self {
Self {
signal_target,
peer_connection,
pending_candidates: Vec::default(),
on_offer_handler: None,
@@ -54,24 +58,37 @@ impl PCTransport {
&mut self.peer_connection
}
pub fn signal_target(&self) -> SignalTarget {
self.signal_target.clone()
}
pub fn on_offer(&mut self, handler: OnOfferHandler) {
self.on_offer_handler = Some(handler);
}
#[tracing::instrument]
pub fn prepare_ice_restart(&mut self) {
self.restarting_ice = true;
}
pub fn close(&mut self) {
self.peer_connection.close();
}
#[tracing::instrument(level = Level::DEBUG)]
pub async fn add_ice_candidate(&mut self, ice_candidate: IceCandidate) -> Result<(), RTCError> {
if self.peer_connection.remote_description().is_none() {
self.pending_candidates.push(ice_candidate);
if self.peer_connection.remote_description().is_some() && !self.restarting_ice {
self.peer_connection
.add_ice_candidate(ice_candidate)
.await?;
return Ok(());
}
self.peer_connection
.add_ice_candidate(ice_candidate)
.await?;
self.pending_candidates.push(ice_candidate);
Ok(())
}
#[tracing::instrument]
#[tracing::instrument(level = Level::DEBUG)]
pub async fn set_remote_description(
&mut self,
remote_description: SessionDescription,
@@ -94,15 +111,33 @@ impl PCTransport {
Ok(())
}
#[tracing::instrument]
#[tracing::instrument(level = Level::DEBUG)]
pub async fn negotiate(&mut self) -> Result<(), RTCError> {
// TODO(theomonnom) Debounce here with NEGOTIATION_FREQUENCY
self.create_and_send_offer(RTCOfferAnswerOptions::default())
.await
}
#[tracing::instrument]
async fn create_and_send_offer(
#[tracing::instrument(level = Level::DEBUG)]
pub async fn create_anwser(
&mut self,
offer: SessionDescription,
options: RTCOfferAnswerOptions,
) -> Result<SessionDescription, RTCError> {
self.set_remote_description(offer).await?;
let answer = self
.peer_connection()
.create_answer(RTCOfferAnswerOptions::default())
.await?;
self.peer_connection()
.set_local_description(answer.clone())
.await?;
Ok(answer)
}
#[tracing::instrument(level = Level::DEBUG)]
pub async fn create_and_send_offer(
&mut self,
options: RTCOfferAnswerOptions,
) -> Result<(), RTCError> {
crates/livekit-core/src/rtc_engine/rtc_events.rs
+38 -13
View File
@@ -11,11 +11,13 @@ use tokio::sync::mpsc;
use crate::proto::SignalTarget;
use crate::rtc_engine::pc_transport::OnOfferHandler;
pub(super) type RTCEmitter = mpsc::UnboundedSender<RTCEvent>;
pub(super) type RTCEvents = mpsc::UnboundedReceiver<RTCEvent>;
use super::pc_transport::PCTransport;
pub type RTCEmitter = mpsc::UnboundedSender<RTCEvent>;
pub type RTCEvents = mpsc::UnboundedReceiver<RTCEvent>;
#[derive(Debug)]
pub(super) enum RTCEvent {
pub enum RTCEvent {
IceCandidate {
ice_candidate: IceCandidate,
target: SignalTarget,
@@ -28,6 +30,7 @@ pub(super) enum RTCEvent {
data_channel: DataChannel,
target: SignalTarget,
},
// TODO (theomonnom): Move Offer to PCTransport
Offer {
offer: SessionDescription,
target: SignalTarget,
@@ -43,19 +46,16 @@ pub(super) enum RTCEvent {
},
}
/// Handlers used to forward event to a channel
/// Handlers used to forward events to a channel
/// Every callback here is called on the signaling thread
pub(super) fn on_connection_change(
target: SignalTarget,
emitter: RTCEmitter,
) -> OnConnectionChangeHandler {
fn on_connection_change(target: SignalTarget, emitter: RTCEmitter) -> OnConnectionChangeHandler {
Box::new(move |state| {
let _ = emitter.send(RTCEvent::ConnectionChange { state, target });
})
}
pub(super) fn on_ice_candidate(target: SignalTarget, emitter: RTCEmitter) -> OnIceCandidateHandler {
fn on_ice_candidate(target: SignalTarget, emitter: RTCEmitter) -> OnIceCandidateHandler {
Box::new(move |ice_candidate| {
let _ = emitter.send(RTCEvent::IceCandidate {
ice_candidate,
@@ -64,7 +64,7 @@ pub(super) fn on_ice_candidate(target: SignalTarget, emitter: RTCEmitter) -> OnI
})
}
pub(super) fn on_offer(target: SignalTarget, emitter: RTCEmitter) -> OnOfferHandler {
fn on_offer(target: SignalTarget, emitter: RTCEmitter) -> OnOfferHandler {
Box::new(move |offer| {
let _ = emitter.send(RTCEvent::Offer { offer, target });
@@ -72,7 +72,7 @@ pub(super) fn on_offer(target: SignalTarget, emitter: RTCEmitter) -> OnOfferHand
})
}
pub(super) fn on_data_channel(target: SignalTarget, emitter: RTCEmitter) -> OnDataChannelHandler {
fn on_data_channel(target: SignalTarget, emitter: RTCEmitter) -> OnDataChannelHandler {
Box::new(move |mut data_channel| {
data_channel.on_message(on_message(emitter.clone()));
@@ -83,7 +83,7 @@ pub(super) fn on_data_channel(target: SignalTarget, emitter: RTCEmitter) -> OnDa
})
}
pub(super) fn on_add_track(target: SignalTarget, emitter: RTCEmitter) -> OnAddTrackHandler {
fn on_add_track(target: SignalTarget, emitter: RTCEmitter) -> OnAddTrackHandler {
Box::new(move |rtp_receiver, streams| {
let _ = emitter.send(RTCEvent::AddTrack {
rtp_receiver,
@@ -93,7 +93,28 @@ pub(super) fn on_add_track(target: SignalTarget, emitter: RTCEmitter) -> OnAddTr
})
}
pub(super) fn on_message(emitter: RTCEmitter) -> OnMessageHandler {
pub fn forward_pc_events(transport: &mut PCTransport, rtc_emitter: RTCEmitter) {
let signal_target = transport.signal_target();
transport
.peer_connection()
.on_ice_candidate(on_ice_candidate(signal_target, rtc_emitter.clone()));
transport
.peer_connection()
.on_data_channel(on_data_channel(signal_target, rtc_emitter.clone()));
transport
.peer_connection()
.on_add_track(on_add_track(signal_target, rtc_emitter.clone()));
transport
.peer_connection()
.on_connection_change(on_connection_change(signal_target, rtc_emitter.clone()));
transport.on_offer(on_offer(transport.signal_target(), rtc_emitter.clone()));
}
fn on_message(emitter: RTCEmitter) -> OnMessageHandler {
Box::new(move |data, binary| {
let _ = emitter.send(RTCEvent::Data {
data: data.to_vec(),
@@ -101,3 +122,7 @@ pub(super) fn on_message(emitter: RTCEmitter) -> OnMessageHandler {
});
})
}
pub fn forward_dc_events(dc: &mut DataChannel, rtc_emitter: RTCEmitter) {
dc.on_message(on_message(rtc_emitter.clone()));
}
crates/livekit-core/src/rtc_engine/rtc_session.rs
+743
View File
@@ -0,0 +1,743 @@
use livekit_webrtc::media_stream::{MediaStream, MediaStreamTrackHandle};
use livekit_webrtc::rtp_receiver::RtpReceiver;
use parking_lot::Mutex;
use std::convert::TryInto;
use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
use std::sync::Arc;
use std::time::Duration;
use tokio::task::JoinHandle;
use tokio::sync::{mpsc, watch, Mutex as AsyncMutex};
use tokio::time::sleep;
use prost::Message;
use serde::{Deserialize, Serialize};
use tracing::{debug, error, trace, warn};
use crate::{proto, signal_client};
use livekit_webrtc::data_channel::{DataChannel, DataChannelInit, DataState};
use livekit_webrtc::jsep::{IceCandidate, SessionDescription};
use livekit_webrtc::peer_connection::{
IceConnectionState, PeerConnectionState, RTCOfferAnswerOptions,
};
use livekit_webrtc::peer_connection_factory::RTCConfiguration;
use crate::proto::data_packet::Value;
use crate::proto::{
data_packet, signal_request, signal_response, CandidateProtocol, DataPacket, DisconnectReason,
JoinResponse, SignalTarget, TrickleRequest,
};
use crate::rtc_engine::lk_runtime::LKRuntime;
use crate::rtc_engine::pc_transport::PCTransport;
use crate::rtc_engine::rtc_events::{RTCEvent, RTCEvents};
use crate::signal_client::{SignalClient, SignalEvent, SignalEvents, SignalOptions};
use super::{rtc_events, EngineError, EngineResult, SimulateScenario};
pub const MAX_ICE_CONNECT_TIMEOUT: Duration = Duration::from_secs(15);
pub const LOSSY_DC_LABEL: &str = "_lossy";
pub const RELIABLE_DC_LABEL: &str = "_reliable";
pub type SessionEmitter = mpsc::UnboundedSender<SessionEvent>;
pub type SessionEvents = mpsc::UnboundedReceiver<SessionEvent>;
#[derive(Debug)]
pub enum SessionEvent {
Data {
participant_sid: String,
payload: Vec<u8>,
kind: proto::data_packet::Kind,
},
MediaTrack {
track: MediaStreamTrackHandle,
stream: MediaStream,
receiver: RtpReceiver,
},
SpeakersChanged {
speakers: Vec<proto::SpeakerInfo>,
},
ConnectionQuality {
updates: Vec<proto::ConnectionQualityInfo>,
},
// TODO(theomonnom): Move entirely the reconnection logic on mod.rs
Close {
source: String,
reason: DisconnectReason,
can_reconnect: bool,
full_reconnect: bool,
retry_now: bool,
},
Connected,
}
#[repr(u8)]
pub enum PCState {
New,
Connected,
Disconnected,
Reconnecting,
Closed,
}
impl TryInto<PCState> for u8 {
type Error = &'static str;
fn try_into(self) -> Result<PCState, Self::Error> {
match self {
0 => Ok(PCState::New),
1 => Ok(PCState::Connected),
2 => Ok(PCState::Disconnected),
3 => Ok(PCState::Reconnecting),
4 => Ok(PCState::Closed),
_ => Err("invalid PCState"),
}
}
}
#[derive(Serialize, Deserialize)]
#[allow(non_snake_case)]
struct IceCandidateJSON {
sdpMid: String,
sdpMLineIndex: i32,
candidate: String,
}
#[derive(Debug, Clone, Default)]
pub struct SessionInfo {
pub url: String,
pub token: String,
pub options: SignalOptions,
pub join_response: JoinResponse,
}
/// Fields shared with rtc_task and signal_task
#[derive(Debug)]
struct SessionInner {
info: SessionInfo,
signal_client: Arc<SignalClient>,
pc_state: AtomicU8, // PCState
has_published: AtomicBool,
publisher_pc: AsyncMutex<PCTransport>,
subscriber_pc: AsyncMutex<PCTransport>,
// Publisher data channels
// used to send data to other participants ( The SFU forwards the messages )
lossy_dc: DataChannel,
reliable_dc: DataChannel,
// Keep a strong reference to the subscriber datachannels,
// so we can receive data from other participants
subscriber_dc: Mutex<Vec<DataChannel>>,
emitter: SessionEmitter,
}
/// This struct holds a WebRTC session
/// The session changes at every reconnection
///
/// RTCSession is also responsable for the signaling and the negotation
#[derive(Debug)]
pub struct RTCSession {
lk_runtime: Arc<LKRuntime>,
inner: Arc<SessionInner>,
close_emitter: watch::Sender<bool>, // false = is_running
signal_task: JoinHandle<()>,
rtc_task: JoinHandle<()>,
}
impl RTCSession {
pub async fn connect(
url: &str,
token: &str,
options: SignalOptions,
lk_runtime: Arc<LKRuntime>,
session_emitter: SessionEmitter,
) -> EngineResult<Self> {
// Connect to the SignalClient
let (signal_client, mut signal_events) = SignalClient::new();
let signal_client = Arc::new(signal_client);
signal_client.connect(url, token, options.clone()).await?;
let join_response = signal_client::utils::next_join_response(&mut signal_events).await?;
debug!("received JoinResponse: {:?}", join_response);
let (rtc_emitter, rtc_events) = mpsc::unbounded_channel();
let rtc_config = RTCConfiguration::from(join_response.clone());
let mut publisher_pc = PCTransport::new(
lk_runtime
.pc_factory
.create_peer_connection(rtc_config.clone())?,
SignalTarget::Publisher,
);
let mut subscriber_pc = PCTransport::new(
lk_runtime
.pc_factory
.create_peer_connection(rtc_config.clone())?,
SignalTarget::Subscriber,
);
let mut lossy_dc = publisher_pc.peer_connection().create_data_channel(
LOSSY_DC_LABEL,
DataChannelInit {
ordered: true,
max_retransmits: Some(0),
..DataChannelInit::default()
},
)?;
let mut reliable_dc = publisher_pc.peer_connection().create_data_channel(
RELIABLE_DC_LABEL,
DataChannelInit {
ordered: true,
..DataChannelInit::default()
},
)?;
// Forward events received in the Signaling Thread to our rtc channel
rtc_events::forward_pc_events(&mut publisher_pc, rtc_emitter.clone());
rtc_events::forward_pc_events(&mut subscriber_pc, rtc_emitter.clone());
rtc_events::forward_dc_events(&mut lossy_dc, rtc_emitter.clone());
rtc_events::forward_dc_events(&mut reliable_dc, rtc_emitter.clone());
let session_info = SessionInfo {
url: url.to_owned(),
token: token.to_owned(),
options,
join_response,
};
let (close_emitter, close_receiver) = watch::channel(false);
let inner = Arc::new(SessionInner {
info: session_info,
pc_state: AtomicU8::new(PCState::New as u8),
has_published: Default::default(),
signal_client,
publisher_pc: AsyncMutex::new(publisher_pc),
subscriber_pc: AsyncMutex::new(subscriber_pc),
lossy_dc,
reliable_dc,
subscriber_dc: Default::default(),
emitter: session_emitter,
});
// Start session tasks
let signal_task = tokio::spawn(
inner
.clone()
.signal_task(signal_events, close_receiver.clone()),
);
let rtc_task = tokio::spawn(inner.clone().rtc_task(rtc_events, close_receiver.clone()));
if !inner.info.join_response.subscriber_primary {
inner.negotiate_publisher().await?;
}
let session = Self {
lk_runtime,
inner: inner.clone(),
close_emitter,
signal_task,
rtc_task,
};
Ok(session)
}
/// Close the PeerConnections and the SignalClient
#[tracing::instrument]
pub async fn close(self) {
// Close the tasks
let _ = self.close_emitter.send(true);
let _ = self.rtc_task.await;
let _ = self.signal_task.await;
self.inner.close().await;
}
pub async fn publish_data(
&self,
data: &DataPacket,
kind: data_packet::Kind,
) -> Result<(), EngineError> {
self.inner.publish_data(data, kind).await
}
pub async fn restart(&self) -> EngineResult<()> {
self.inner.restart_session().await
}
pub async fn wait_pc_connection(&self) -> EngineResult<()> {
self.inner.wait_pc_connection().await
}
pub async fn simulate_scenario(&self, scenario: SimulateScenario) {
self.inner.simulate_scenario(scenario).await
}
}
impl RTCSession {
pub fn info(&self) -> &SessionInfo {
&self.inner.info
}
pub fn state(&self) -> PCState {
self.inner
.pc_state
.load(Ordering::SeqCst)
.try_into()
.unwrap()
}
pub fn publisher(&self) -> &AsyncMutex<PCTransport> {
&self.inner.publisher_pc
}
pub fn subscriber(&self) -> &AsyncMutex<PCTransport> {
&self.inner.subscriber_pc
}
pub fn signal_client(&self) -> &Arc<SignalClient> {
&self.inner.signal_client
}
pub fn data_channel(&self, kind: data_packet::Kind) -> &DataChannel {
&self.inner.data_channel(kind)
}
}
impl SessionInner {
async fn rtc_task(
self: Arc<Self>,
mut rtc_events: RTCEvents,
mut close_receiver: watch::Receiver<bool>,
) {
loop {
tokio::select! {
res = rtc_events.recv() => {
if let Some(event) = res {
if let Err(err) = self.on_rtc_event(event).await {
error!("failed to handle rtc event: {:?}", err);
}
} else {
panic!("rtc_events has been closed unexpectedly");
}
},
_ = close_receiver.changed() => {
break;
}
}
}
}
async fn signal_task(
self: Arc<Self>,
mut signal_events: SignalEvents,
mut close_receiver: watch::Receiver<bool>,
) {
loop {
tokio::select! {
res = signal_events.recv() => {
if let Some(signal) = res {
match signal {
SignalEvent::Open => {}
SignalEvent::Signal(signal) => {
if let Err(err) = self.on_signal_event(signal).await {
error!("failed to handle signal: {:?}", err);
}
}
SignalEvent::Close => {
self.on_session_disconnected("SignalClient closed", DisconnectReason::UnknownReason, true, false, false);
}
}
} else {
panic!("signal_events has been closed unexpectedly");
}
},
_ = close_receiver.changed() => {
break;
}
}
}
}
async fn on_signal_event(&self, event: signal_response::Message) -> EngineResult<()> {
match event {
signal_response::Message::Answer(answer) => {
trace!("received publisher answer: {:?}", answer);
let answer = SessionDescription::from(answer.r#type.parse().unwrap(), &answer.sdp)?;
self.publisher_pc
.lock()
.await
.set_remote_description(answer)
.await?;
}
signal_response::Message::Offer(offer) => {
trace!("received subscriber offer: {:?}", offer);
let offer = SessionDescription::from(offer.r#type.parse().unwrap(), &offer.sdp)?;
let answer = self
.subscriber_pc
.lock()
.await
.create_anwser(offer, RTCOfferAnswerOptions::default())
.await?;
self.signal_client
.send(signal_request::Message::Answer(proto::SessionDescription {
r#type: "answer".to_string(),
sdp: answer.to_string(),
}))
.await;
}
signal_response::Message::Trickle(trickle) => {
let target = SignalTarget::from_i32(trickle.target).unwrap();
let ice_candidate = {
let json = serde_json::from_str::<IceCandidateJSON>(&trickle.candidate_init)?;
IceCandidate::from(&json.sdpMid, json.sdpMLineIndex, &json.candidate)?
};
trace!("received ice_candidate {:?} {:?}", target, ice_candidate);
if target == SignalTarget::Publisher {
self.publisher_pc
.lock()
.await
.add_ice_candidate(ice_candidate)
.await?;
} else {
self.subscriber_pc
.lock()
.await
.add_ice_candidate(ice_candidate)
.await?;
}
}
signal_response::Message::Leave(leave) => {
self.on_session_disconnected(
"received leave",
leave.reason(),
leave.can_reconnect,
true,
true,
);
}
signal_response::Message::SpeakersChanged(speaker) => {
let _ = self.emitter.send(SessionEvent::SpeakersChanged {
speakers: speaker.speakers,
});
}
signal_response::Message::ConnectionQuality(quality) => {
let _ = self.emitter.send(SessionEvent::ConnectionQuality {
updates: quality.updates,
});
}
_ => {}
}
Ok(())
}
async fn on_rtc_event(&self, event: RTCEvent) -> EngineResult<()> {
match event {
RTCEvent::IceCandidate {
ice_candidate,
target,
} => {
self.signal_client
.send(signal_request::Message::Trickle(TrickleRequest {
candidate_init: serde_json::to_string(&IceCandidateJSON {
sdpMid: ice_candidate.sdp_mid(),
sdpMLineIndex: ice_candidate.sdp_mline_index(),
candidate: ice_candidate.candidate(),
})?,
target: target as i32,
}))
.await;
}
RTCEvent::ConnectionChange { state, target } => {
trace!("connection change, {:?} {:?}", state, target);
let is_primary = self.info.join_response.subscriber_primary
&& target == SignalTarget::Subscriber;
if is_primary && state == PeerConnectionState::Connected {
let old_state = self
.pc_state
.swap(PCState::Connected as u8, Ordering::SeqCst);
if old_state == PCState::New as u8 {
let _ = self.emitter.send(SessionEvent::Connected);
}
} else if state == PeerConnectionState::Failed {
self.pc_state
.store(PCState::Disconnected as u8, Ordering::SeqCst);
self.on_session_disconnected(
"pc_state failed",
DisconnectReason::UnknownReason,
true,
false,
false,
);
}
}
RTCEvent::DataChannel {
data_channel,
target: _,
} => {
self.subscriber_dc.lock().push(data_channel);
}
RTCEvent::Offer { offer, target: _ } => {
// Send the publisher offer to the server
self.signal_client
.send(signal_request::Message::Offer(proto::SessionDescription {
r#type: "offer".to_string(),
sdp: offer.to_string(),
}))
.await;
}
RTCEvent::AddTrack {
rtp_receiver,
mut streams,
target: _,
} => {
if !streams.is_empty() {
let _ = self.emitter.send(SessionEvent::MediaTrack {
track: rtp_receiver.track(),
stream: streams.remove(0),
receiver: rtp_receiver,
});
} else {
warn!("AddTrack event with no streams");
}
}
RTCEvent::Data { data, binary } => {
if !binary {
Err(EngineError::Internal(
"text messages aren't supported".to_string(),
))?;
}
let data = DataPacket::decode(&*data)?;
match data.value.unwrap() {
Value::User(user) => {
let _ = self.emitter.send(SessionEvent::Data {
participant_sid: user.participant_sid,
payload: user.payload,
kind: data_packet::Kind::from_i32(data.kind).unwrap(),
});
}
Value::Speaker(_) => {}
}
}
}
Ok(())
}
/// Called when the SignalClient or one of the PeerConnection has lost the connection
/// The RTCEngine may try a reconnect.
fn on_session_disconnected(
&self,
source: &str,
reason: DisconnectReason,
can_reconnect: bool,
retry_now: bool,
full_reconnect: bool,
) {
let _ = self.emitter.send(SessionEvent::Close {
source: source.to_owned(),
reason,
can_reconnect,
retry_now,
full_reconnect,
});
}
#[tracing::instrument]
async fn close(&self) {
self.signal_client.close().await;
self.publisher_pc.lock().await.close();
self.subscriber_pc.lock().await.close();
}
#[tracing::instrument]
async fn simulate_scenario(&self, scenario: SimulateScenario) {
match scenario {
SimulateScenario::SignalReconnect => {
self.signal_client.close().await;
}
SimulateScenario::Speaker => {
self.signal_client
.send(signal_request::Message::Simulate(proto::SimulateScenario {
scenario: Some(proto::simulate_scenario::Scenario::SpeakerUpdate(3)),
}))
.await;
}
SimulateScenario::NodeFailure => {
self.signal_client
.send(signal_request::Message::Simulate(proto::SimulateScenario {
scenario: Some(proto::simulate_scenario::Scenario::NodeFailure(true)),
}))
.await;
}
SimulateScenario::ServerLeave => {
self.signal_client
.send(signal_request::Message::Simulate(proto::SimulateScenario {
scenario: Some(proto::simulate_scenario::Scenario::ServerLeave(true)),
}))
.await;
}
SimulateScenario::Migration => {
self.signal_client
.send(signal_request::Message::Simulate(proto::SimulateScenario {
scenario: Some(proto::simulate_scenario::Scenario::Migration(true)),
}))
.await;
}
SimulateScenario::ForceTcp => {
self.signal_client
.send(signal_request::Message::Simulate(proto::SimulateScenario {
scenario: Some(
proto::simulate_scenario::Scenario::SwitchCandidateProtocol(
CandidateProtocol::Tcp as i32,
),
),
}))
.await;
}
SimulateScenario::ForceTls => {
self.signal_client
.send(signal_request::Message::Simulate(proto::SimulateScenario {
scenario: Some(
proto::simulate_scenario::Scenario::SwitchCandidateProtocol(
CandidateProtocol::Tls as i32,
),
),
}))
.await;
}
}
}
#[tracing::instrument(skip(data))]
async fn publish_data(
&self,
data: &DataPacket,
kind: data_packet::Kind,
) -> Result<(), EngineError> {
self.ensure_publisher_connected(kind).await?;
self.data_channel(kind)
.send(&data.encode_to_vec(), true)
.map_err(Into::into)
}
/// Try to restart the session by doing an ICE Restart (The SignalClient is also restarted)
/// This reconnection if more seemless than the full reconnection implemented in ['RTCEngine']
async fn restart_session(&self) -> EngineResult<()> {
self.signal_client.close().await;
let mut options = self.info.options.clone();
options.sid = self.info.join_response.participant.clone().unwrap().sid;
options.reconnect = true;
self.signal_client
.connect(&self.info.url, &self.info.token, options)
.await?;
self.subscriber_pc.lock().await.prepare_ice_restart();
if self.has_published.load(Ordering::Acquire) {
self.publisher_pc
.lock()
.await
.create_and_send_offer(RTCOfferAnswerOptions {
ice_restart: true,
..Default::default()
})
.await?;
}
self.wait_pc_connection().await?;
self.signal_client.flush_queue().await;
Ok(())
}
// Wait for PCState to become PCState::Connected
// Timeout after ['MAX_ICE_CONNECT_TIMEOUT']
async fn wait_pc_connection(&self) -> EngineResult<()> {
let wait_connected = async move {
while self.pc_state.load(Ordering::Acquire) != PCState::Connected as u8 {
tokio::task::yield_now().await;
}
};
tokio::select! {
_ = wait_connected => Ok(()),
_ = sleep(MAX_ICE_CONNECT_TIMEOUT) => {
let err = EngineError::Connection("wait_pc_connection timed out".to_string());
Err(err)
}
}
}
/// Start publisher negotiation
async fn negotiate_publisher(&self) -> EngineResult<()> {
self.has_published.store(true, Ordering::Release);
let res = self.publisher_pc.lock().await.negotiate().await;
if let Err(err) = &res {
error!("failed to negotiate the publisher: {:?}", err);
}
res.map_err(Into::into)
}
/// Ensure the Publisher PC is connected, if not, start the negotiation
/// This is required when sending data to the server
async fn ensure_publisher_connected(&self, kind: data_packet::Kind) -> EngineResult<()> {
if !self.info.join_response.subscriber_primary {
return Ok(());
}
if !self.publisher_pc.lock().await.is_connected()
&& self
.publisher_pc
.lock()
.await
.peer_connection()
.ice_connection_state()
!= IceConnectionState::IceConnectionChecking
{
let _ = self.negotiate_publisher().await;
}
let dc = self.data_channel(kind);
if dc.state() == DataState::Open {
return Ok(());
}
// Wait until the PeerConnection is connected
let wait_connected = async {
while self.publisher_pc.lock().await.is_connected() && dc.state() == DataState::Open {
tokio::task::yield_now().await;
}
};
// TODO(theomonnom) Avoid 15 seconds deadlock on the RTCEngine by recv close here
tokio::select! {
_ = wait_connected => Ok(()),
_ = sleep(MAX_ICE_CONNECT_TIMEOUT) => {
let err = EngineError::Connection("could not establish publisher connection: timeout".to_string());
error!(error = ?err);
Err(err)
}
}
}
fn data_channel(&self, kind: data_packet::Kind) -> &DataChannel {
if kind == data_packet::Kind::Reliable {
&self.reliable_dc
} else {
&self.lossy_dc
}
}
}
crates/livekit-core/src/signal_client/mod.rs
+57 -25
View File
@@ -1,15 +1,18 @@
use std::fmt::Debug;
use std::time::Duration;
use livekit_webrtc::peer_connection_factory::{
ContinualGatheringPolicy, ICEServer, IceTransportsType, RTCConfiguration,
};
use parking_lot::RwLock;
use thiserror::Error;
use tokio::sync::mpsc;
use tokio_tungstenite::tungstenite::Error as WsError;
use crate::proto::{signal_request, signal_response, JoinResponse};
use crate::signal_client::signal_stream::SignalStream;
use tracing::{instrument, Level};
mod signal_stream;
@@ -21,7 +24,7 @@ pub const JOIN_RESPONSE_TIMEOUT: Duration = Duration::from_secs(5);
#[derive(Error, Debug)]
pub enum SignalError {
#[error("websocket failure")]
#[error("ws failure: {0}")]
WsError(#[from] WsError),
#[error("failed to parse the url")]
UrlParse(#[from] url::ParseError),
@@ -33,18 +36,18 @@ pub enum SignalError {
/// Events used by the RTCEngine who will handle the reconnection logic
#[derive(Debug)]
pub(crate) enum SignalEvent {
pub enum SignalEvent {
Open,
Signal(signal_response::Message),
Close,
}
#[derive(Debug)]
pub(crate) struct SignalOptions {
reconnect: bool,
auto_subscribe: bool,
sid: String,
adaptive_stream: bool,
#[derive(Debug, Clone)]
pub struct SignalOptions {
pub(crate) reconnect: bool,
pub(crate) sid: String,
pub auto_subscribe: bool,
pub adaptive_stream: bool,
}
impl Default for SignalOptions {
@@ -60,32 +63,59 @@ impl Default for SignalOptions {
#[derive(Debug)]
pub struct SignalClient {
stream: SignalStream,
stream: RwLock<Option<SignalStream>>,
emitter: SignalEmitter,
}
impl SignalClient {
pub(crate) async fn connect(
pub fn new() -> (Self, SignalEvents) {
let (emitter, events) = mpsc::channel(8);
(
Self {
stream: Default::default(),
emitter,
},
events,
)
}
#[instrument(level = Level::DEBUG, skip(url, token, options))]
pub async fn connect(
&self,
url: &str,
token: &str,
options: SignalOptions,
) -> SignalResult<(Self, SignalEvents)> {
let (emitter, events) = mpsc::channel(8);
let stream = SignalStream::connect(url, token, options, emitter.clone()).await?;
// TODO(theomonnom) Retry initial connection
Ok((Self { stream, emitter }, events))
) -> SignalResult<()> {
let stream = SignalStream::connect(url, token, options, self.emitter.clone()).await?;
*self.stream.write() = Some(stream);
Ok(())
}
pub async fn send(&self, signal: signal_request::Message) {
if let Err(_) = self.stream.send(signal).await {
// TODO(theomonnom) Queue message ( Ignore on full reconnect )
#[instrument(level = Level::DEBUG)]
pub async fn close(&self) {
if let Some(stream) = self.stream.write().take() {
stream.close().await;
}
}
pub async fn reconnect(&self) {
// TODO(theomonnom) Close & recreate SignalStream, also send the queue if needed
#[instrument(level = Level::DEBUG)]
pub async fn send(&self, signal: signal_request::Message) {
if let Some(stream) = self.stream.read().as_ref() {
if stream.send(signal).await.is_ok() {
return;
}
}
// TODO(theomonnom): enqueue message
}
pub async fn clear_queue(&self) {
// TODO(theomonnom): impl
}
#[instrument(level = Level::DEBUG)]
pub async fn flush_queue(&self) {
// TODO(theomonnom): impl
}
}
@@ -112,13 +142,15 @@ impl From<JoinResponse> for RTCConfiguration {
pub mod utils {
use crate::proto::{signal_response, JoinResponse};
use crate::signal_client::{SignalError, SignalEvent, SignalResult, JOIN_RESPONSE_TIMEOUT};
use tokio::sync::mpsc;
use tokio::time::timeout;
use tokio_tungstenite::tungstenite::Error as WsError;
use tracing::{event, Level};
use tracing::{event, instrument, Level};
use super::SignalEvents;
#[instrument(level = Level::DEBUG, skip(receiver))]
pub(crate) async fn next_join_response(
receiver: &mut mpsc::Receiver<SignalEvent>,
receiver: &mut SignalEvents,
) -> SignalResult<JoinResponse> {
let join = async {
while let Some(event) = receiver.recv().await {
crates/livekit-core/src/signal_client/signal_stream.rs
+7 -7
View File
@@ -4,6 +4,7 @@ use prost::Message as ProstMessage;
use tokio::net::TcpStream;
use tokio::sync::{mpsc, oneshot};
use tokio::task::JoinHandle;
use tokio_tungstenite::tungstenite::protocol::frame::coding::CloseCode;
use tokio_tungstenite::tungstenite::protocol::CloseFrame;
use tokio_tungstenite::tungstenite::Message;
@@ -60,6 +61,7 @@ impl SignalStream {
.append_pair("access_token", token)
.append_pair("protocol", PROTOCOL_VERSION.to_string().as_str())
.append_pair("reconnect", if options.reconnect { "1" } else { "0" })
.append_pair("sid", &options.sid)
.append_pair(
"auto_subscribe",
if options.auto_subscribe { "1" } else { "0" },
@@ -69,17 +71,15 @@ impl SignalStream {
if options.adaptive_stream { "1" } else { "0" },
);
event!(Level::DEBUG, "connecting to websocket: {}", lk_url);
event!(Level::INFO, "connecting to SignalClient: {}", lk_url);
let (ws_stream, _) = connect_async(lk_url).await?;
event!(Level::DEBUG, "connected to websocket");
let _ = emitter.send(SignalEvent::Open).await;
let (ws_writer, ws_reader) = ws_stream.split();
let (internal_tx, internal_rx) = mpsc::channel::<InternalMessage>(8);
let write_handle =
tokio::spawn(Self::handle_write(internal_rx, ws_writer, emitter.clone()));
let read_handle = tokio::spawn(Self::handle_read(internal_tx.clone(), ws_reader, emitter));
let write_handle = tokio::spawn(Self::write_task(internal_rx, ws_writer, emitter.clone()));
let read_handle = tokio::spawn(Self::read_task(internal_tx.clone(), ws_reader, emitter));
Ok(Self {
internal_tx,
@@ -119,7 +119,7 @@ impl SignalStream {
/// This task is used to send messages to the websocket
/// It is also responsible for closing the connection
async fn handle_write(
async fn write_task(
mut internal_rx: mpsc::Receiver<InternalMessage>,
mut ws_writer: SplitSink<WebSocket, Message>,
emitter: SignalEmitter,
@@ -170,7 +170,7 @@ impl SignalStream {
/// and dispatch them through the EventEmitter.
///
/// It can also send messages to [handle_write] task ( Used e.g. answer to pings )
async fn handle_read(
async fn read_task(
internal_tx: mpsc::Sender<InternalMessage>,
mut ws_reader: SplitStream<WebSocket>,
emitter: SignalEmitter,
crates/livekit-utils/Cargo.toml
+2
View File
@@ -6,3 +6,5 @@ edition = "2021"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
parking_lot = "0.12.1"
tokio = { version = "1", features = ["full"] }
crates/livekit-utils/src/enum_dispatch.rs
+7 -5
View File
@@ -1,3 +1,6 @@
// TODO(theomonnom): Match the complete function signature like:
// - pub(crate) fn update_info(&self, info: ParticipantInfo) -> ();
#[macro_export]
macro_rules! enum_dispatch {
// This arm is used to avoid nested loops with the arguments
@@ -10,16 +13,15 @@ macro_rules! enum_dispatch {
}
};
($fnc:ident, $self:ty, [$($arg:ident: $t:ty),*], $ret:ty, [$($variant:ident),+]) => {
fn $fnc(self: $self, $($arg: $t),*) -> $ret {
($vis:vis$(,)? $fnc:ident, $self:ty, [$($arg:ident: $t:ty),*], $ret:ty, [$($variant:ident),+]) => {
$vis fn $fnc(self: $self, $($arg: $t),*) -> $ret {
enum_dispatch!(@match self $fnc ($($arg,)*) [$($variant),+])
}
};
($variants:tt $(fnc!($fnc:ident, $self:ty, $args:tt, $ret:ty);)+) => {
($variants:tt $(fnc!($vis:vis$(,)? $fnc:ident, $self:ty, $args:tt, $ret:ty);)+) => {
$(
enum_dispatch!($fnc, $self, $args, $ret, $variants);
enum_dispatch!($vis, $fnc, $self, $args, $ret, $variants);
)*
};
}
crates/livekit-utils/src/lib.rs
+1
View File
@@ -1 +1,2 @@
pub mod enum_dispatch;
pub mod observer;
crates/livekit-utils/src/observer.rs
+39
View File
@@ -0,0 +1,39 @@
// Really basic implementation of the observer pattern using mpsc channels.
// Currently unbounded channels
use tokio::sync::mpsc;
#[derive(Debug)]
pub struct Dispatcher<T>
where
T: Clone,
{
senders: Vec<mpsc::UnboundedSender<T>>,
}
impl<T> Default for Dispatcher<T>
where
T: Clone,
{
fn default() -> Self {
Self {
senders: Default::default(),
}
}
}
impl<T> Dispatcher<T>
where
T: Clone,
{
pub fn register(&mut self) -> mpsc::UnboundedReceiver<T> {
let (tx, rx) = mpsc::unbounded_channel();
self.senders.push(tx);
rx
}
pub fn dispatch(&mut self, msg: &T) {
self.senders
.retain(|sender| sender.send(msg.clone()).is_err());
}
}
crates/livekit-webrtc/libwebrtc-sys/build.rs
+2 -1
View File
@@ -62,6 +62,7 @@ fn main() {
path::PathBuf::from("./include"),
libwebrtc_dir.clone(),
libwebrtc_dir.join("third_party/abseil-cpp/"),
libwebrtc_dir.join("third_party/libyuv/include/"),
libwebrtc_dir.join("third_party/libc++/"),
// For mac & ios
libwebrtc_dir.join("sdk/objc"),
@@ -124,7 +125,7 @@ fn main() {
println!("cargo:rustc-link-lib=dylib=d3d11");
println!("cargo:rustc-link-lib=dylib=dxgi");
println!("cargo:rustc-link-lib=dylib=dwmapi");
println!("cargo:rustc-link-lib=dylib=webrtc");
println!("cargo:rustc-link-lib=static=webrtc");
builder
.flag("/std:c++17")
crates/livekit-webrtc/libwebrtc-sys/include/livekit/data_channel.h
+2 -2
View File
@@ -24,10 +24,10 @@ class DataChannel {
void register_observer(NativeDataChannelObserver& observer);
void unregister_observer();
bool send(const DataBuffer& buffer);
bool send(const DataBuffer& buffer) const;
rust::String label() const;
DataState state() const;
void close();
void close() const;
private:
std::shared_ptr<RTCRuntime> rtc_runtime_;
crates/livekit-webrtc/libwebrtc-sys/src/data_channel.cpp
+2 -2
View File
@@ -24,7 +24,7 @@ void DataChannel::unregister_observer() {
data_channel_->UnregisterObserver();
}
bool DataChannel::send(const DataBuffer& buffer) {
bool DataChannel::send(const DataBuffer& buffer) const {
return data_channel_->Send(webrtc::DataBuffer{
rtc::CopyOnWriteBuffer(buffer.ptr, buffer.len), buffer.binary});
}
@@ -37,7 +37,7 @@ DataState DataChannel::state() const {
return static_cast<DataState>(data_channel_->state());
}
void DataChannel::close() {
void DataChannel::close() const {
return data_channel_->Close();
}
crates/livekit-webrtc/libwebrtc-sys/src/data_channel.rs
+4 -2
View File
@@ -68,10 +68,10 @@ pub mod ffi {
);
fn unregister_observer(self: Pin<&mut DataChannel>);
fn send(self: Pin<&mut DataChannel>, data: &DataBuffer) -> bool;
fn send(self: &DataChannel, data: &DataBuffer) -> bool;
fn label(self: &DataChannel) -> String;
fn state(self: &DataChannel) -> DataState;
fn close(self: Pin<&mut DataChannel>);
fn close(self: &DataChannel);
fn create_data_channel_init(init: DataChannelInit) -> UniquePtr<NativeDataChannelInit>;
fn create_native_data_channel_observer(
@@ -83,8 +83,10 @@ pub mod ffi {
}
unsafe impl Send for ffi::DataChannel {}
unsafe impl Sync for ffi::DataChannel {}
unsafe impl Send for ffi::NativeDataChannelObserver {}
unsafe impl Sync for ffi::NativeDataChannelObserver {}
// DataChannelObserver
crates/livekit-webrtc/libwebrtc-sys/src/webrtc.cpp
+2 -2
View File
@@ -8,7 +8,7 @@
namespace livekit {
RTCRuntime::RTCRuntime() {
rtc::LogMessage::LogToDebug(rtc::LS_INFO);
// rtc::LogMessage::LogToDebug(rtc::LS_INFO);
RTC_LOG(LS_INFO) << "RTCRuntime()";
RTC_CHECK(rtc::InitializeSSL()) << "Failed to InitializeSSL()";
@@ -43,4 +43,4 @@ rtc::Thread* RTCRuntime::signaling_thread() const {
std::shared_ptr<RTCRuntime> create_rtc_runtime() {
return std::make_shared<RTCRuntime>();
}
} // namespace livekit
} // namespace livekit
crates/livekit-webrtc/src/data_channel.rs
+3 -4
View File
@@ -58,7 +58,7 @@ impl DataChannel {
dc
}
pub fn send(&mut self, data: &[u8], binary: bool) -> Result<(), DataSendError> {
pub fn send(&self, data: &[u8], binary: bool) -> Result<(), DataSendError> {
let buffer = sys_dc::ffi::DataBuffer {
ptr: data.as_ptr(),
len: data.len(),
@@ -66,7 +66,6 @@ impl DataChannel {
};
self.cxx_handle
.pin_mut()
.send(&buffer)
.then_some(())
.ok_or(DataSendError {})
@@ -80,8 +79,8 @@ impl DataChannel {
self.cxx_handle.state()
}
pub fn close(&mut self) {
self.cxx_handle.pin_mut().close();
pub fn close(&self) {
self.cxx_handle.close();
}
pub fn on_state_change(&mut self, handler: OnStateChangeHandler) {
crates/livekit-webrtc/src/media_stream.rs
+9 -22
View File
@@ -1,6 +1,7 @@
use cxx::UniquePtr;
use libwebrtc_sys::media_stream as sys_ms;
use libwebrtc_sys::MEDIA_TYPE_VIDEO;
use livekit_utils::enum_dispatch;
use std::fmt::{Debug, Formatter};
use std::pin::Pin;
use std::sync::{Arc, Mutex};
@@ -25,17 +26,6 @@ pub enum MediaStreamTrackHandle {
Video(Arc<VideoTrack>),
}
macro_rules! shared_getter {
($x:ident, $ret:ty) => {
fn $x(&self) -> $ret {
match self {
Self::Video(inner) => inner.$x(),
Self::Audio(inner) => inner.$x(),
}
}
};
}
impl MediaStreamTrackHandle {
pub(crate) fn new(cxx_handle: UniquePtr<sys_ms::ffi::MediaStreamTrack>) -> Self {
unsafe {
@@ -66,17 +56,14 @@ impl Debug for MediaStreamTrackHandle {
}
impl MediaStreamTrackTrait for MediaStreamTrackHandle {
shared_getter!(kind, String);
shared_getter!(id, String);
shared_getter!(enabled, bool);
shared_getter!(state, TrackState);
fn set_enabled(&self, enabled: bool) -> bool {
match self {
Self::Video(inner) => inner.set_enabled(enabled),
Self::Audio(inner) => inner.set_enabled(enabled),
}
}
enum_dispatch!(
[Audio, Video]
fnc!(kind, &Self, [], String);
fnc!(id, &Self, [], String);
fnc!(enabled, &Self, [], bool);
fnc!(state, &Self, [], TrackState);
fnc!(set_enabled, &Self, [enabled: bool], bool);
);
}
pub struct AudioTrack {
examples/.cargo/config
+2
View File
@@ -0,0 +1,2 @@
[target.x86_64-pc-windows-msvc]
rustflags = ["-C", "target-feature=+crt-static"]
examples/Cargo.lock
Generated
+239 -201
View File
File diff suppressed because it is too large Load Diff
examples/Cargo.toml
+1 -1
View File
@@ -1,4 +1,4 @@
[workspace]
members = ["*"]
exclude = ["target"]
exclude = ["target", ".cargo"]
resolver = "2"
examples/simple_room/Cargo.toml
+1 -1
View File
@@ -11,8 +11,8 @@ livekit = { path = "../.." }
futures = "0.3"
wgpu = "0.14.0"
winit = "0.27.5"
parking_lot = "0.12.1"
egui = { git = "https://github.com/emilk/egui" }
egui-wgpu = { git = "https://github.com/emilk/egui", features = ["winit"] }
egui-winit = { git = "https://github.com/emilk/egui" }
egui_demo_lib = { git = "https://github.com/emilk/egui" }
examples/simple_room/src/app.rs
+361
View File
@@ -0,0 +1,361 @@
use crate::events::UiCmd;
use crate::video_renderer::VideoRenderer;
use crate::{events::AsyncCmd, video_grid::VideoGrid};
use egui_wgpu::WgpuConfiguration;
use livekit::room::track::remote_track::RemoteTrackHandle;
use parking_lot::Mutex;
use std::sync::{
atomic::{AtomicBool, Ordering},
Arc,
};
use tokio::sync::mpsc;
use livekit::room::{ConnectionState, Room, RoomError, SimulateScenario};
// Useful default constants for developing
const DEFAULT_URL: &str = "ws://localhost:7880";
const DEFAULT_TOKEN : &str = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjIzODQ4MDY0NzMsImlzcyI6IkFQSXpLYkFTaUNWYWtnSiIsIm5hbWUiOiJuYXRpdmUiLCJuYmYiOjE2NjQ4MDY0NzMsInN1YiI6Im5hdGl2ZSIsInZpZGVvIjp7InJvb21DcmVhdGUiOnRydWUsInJvb21Kb2luIjp0cnVlfX0.BgVdBnq3XFD3_BQHoe1azqjifYysubgFl6Qlzu9IQGI";
// eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjIzODQ4MDY3MzAsImlzcyI6IkFQSXpLYkFTaUNWYWtnSiIsIm5hbWUiOiJ3ZWIiLCJuYmYiOjE2NjQ4MDY3MzAsInN1YiI6IndlYiIsInZpZGVvIjp7InJvb21DcmVhdGUiOnRydWUsInJvb21Kb2luIjp0cnVlfX0.VbDoULjX1CVGZu2sPy3SvWYlVZUBXxQVPmdB9BnmlN4
use winit::{
event::*,
event_loop::{ControlFlow, EventLoop},
window::{WindowBuilder, WindowId},
};
struct AppState {
room: Mutex<Room>,
connecting: AtomicBool,
}
struct App {
state: Arc<AppState>,
video_renderers: Vec<VideoRenderer>,
egui_context: egui::Context,
egui_state: egui_winit::State,
egui_painter: egui_wgpu::winit::Painter,
window: winit::window::Window,
cmd_tx: mpsc::UnboundedSender<AsyncCmd>,
cmd_rx: mpsc::UnboundedReceiver<UiCmd>,
// UI State
lk_url: String,
lk_token: String,
connection_failure: Option<String>,
room_state: ConnectionState,
}
pub fn run(rt: tokio::runtime::Runtime) {
rt.block_on(async {
let event_loop = EventLoop::new();
let window = WindowBuilder::new()
.with_title("LiveKit - NativeSDK")
.build(&event_loop)
.unwrap();
let egui_context = egui::Context::default();
let egui_state = egui_winit::State::new(&event_loop);
let mut egui_painter = egui_wgpu::winit::Painter::new(WgpuConfiguration::default(), 1, 32);
unsafe {
egui_painter.set_window(Some(&window));
}
let (async_cmd_tx, mut async_cmd_rx) = mpsc::unbounded_channel::<AsyncCmd>();
let (ui_cmd_tx, ui_cmd_rx) = mpsc::unbounded_channel::<UiCmd>();
let state = Arc::new(AppState {
room: Mutex::new(Room::new()),
connecting: AtomicBool::new(false),
});
let mut app = App {
state: state.clone(),
video_renderers: Vec::default(),
egui_context,
egui_state,
egui_painter,
window,
cmd_tx: async_cmd_tx,
cmd_rx: ui_cmd_rx,
lk_url: DEFAULT_URL.to_owned(),
lk_token: DEFAULT_TOKEN.to_owned(),
connection_failure: None,
room_state: ConnectionState::Connected,
};
// Async event loop
tokio::spawn(async move {
{
let events = state.room.lock().events();
events.on_track_subscribed({
let ui_cmd_tx = ui_cmd_tx.clone();
move |event| {
let ui_cmd_tx = ui_cmd_tx.clone();
async move {
ui_cmd_tx.send(UiCmd::TrackSubscribed { event }).unwrap();
}
}
});
}
while let Some(event) = async_cmd_rx.recv().await {
match event {
AsyncCmd::RoomConnect { url, token } => {
state.connecting.store(true, Ordering::SeqCst);
let mut room = state.room.lock();
ui_cmd_tx
.send(UiCmd::ConnectResult {
result: room.connect(&url, &token).await,
})
.unwrap();
state.connecting.store(false, Ordering::SeqCst);
}
AsyncCmd::SimulateScenario { scenario } => {
if let Some(handle) = state.room.lock().get_handle() {
let _ = handle.simulate_scenario(scenario).await;
}
}
}
}
});
tokio::task::block_in_place(move || loop {
// UI/Main Thread
event_loop.run(move |event, _, control_flow| {
app.update(event, control_flow);
});
});
});
}
impl App {
fn update<T>(&mut self, event: Event<'_, T>, control_flow: &mut ControlFlow) {
if let Ok(cmd) = self.cmd_rx.try_recv() {
match cmd {
UiCmd::ConnectResult { result } => {
if let Err(err) = result {
self.connection_failure = Some(err.to_string());
} else {
self.connection_failure = None
}
}
UiCmd::TrackSubscribed { event } => {
match event.track {
RemoteTrackHandle::Video(video_track) => {
// Create a new VideoRenderer
let video_renderer = VideoRenderer::new(
self.egui_painter.render_state().clone().unwrap(),
video_track.rtc_track(),
);
self.video_renderers.push(video_renderer);
}
RemoteTrackHandle::Audio(_) => {
// The demo doesn't support Audio rendering at the moment.
}
};
}
}
}
match event {
Event::WindowEvent { window_id, event } => {
if let Some(flow) = self.on_window_event(window_id, event) {
*control_flow = flow;
}
}
Event::RedrawRequested(window_id) if window_id == self.window.id() => {
self.render();
}
Event::RedrawEventsCleared => {
self.window.request_redraw();
}
_ => {}
};
}
fn on_window_event(
&mut self,
_window_id: WindowId,
event: WindowEvent<'_>,
) -> Option<ControlFlow> {
if self
.egui_state
.on_event(&self.egui_context, &event)
.consumed
{
return None;
}
match event {
WindowEvent::CloseRequested => Some(ControlFlow::Exit),
WindowEvent::Resized(inner_size) => {
self.egui_painter
.on_window_resized(inner_size.width, inner_size.height);
None
}
WindowEvent::ScaleFactorChanged { new_inner_size, .. } => {
self.egui_painter
.on_window_resized(new_inner_size.width, new_inner_size.height);
None
}
_ => None,
}
}
fn ui(&mut self, ui: &mut egui::Ui) {
egui::TopBottomPanel::top("top_panel").show(ui.ctx(), |ui| {
egui::menu::bar(ui, |ui| {
ui.menu_button("Tools", |ui| {
if ui.button("Logs").clicked() {}
if ui.button("Profiler").clicked() {}
if ui.button("WebRTC Stats").clicked() {}
if ui.button("Events").clicked() {}
});
ui.menu_button("Simulate", |ui| {
if ui.button("SignalReconnect").clicked() {
let _ = self.cmd_tx.send(AsyncCmd::SimulateScenario {
scenario: SimulateScenario::SignalReconnect,
});
}
if ui.button("Speaker").clicked() {
let _ = self.cmd_tx.send(AsyncCmd::SimulateScenario {
scenario: SimulateScenario::Speaker,
});
}
if ui.button("NodeFailure").clicked() {
let _ = self.cmd_tx.send(AsyncCmd::SimulateScenario {
scenario: SimulateScenario::NodeFailure,
});
}
if ui.button("ServerLeave").clicked() {
let _ = self.cmd_tx.send(AsyncCmd::SimulateScenario {
scenario: SimulateScenario::ServerLeave,
});
}
if ui.button("Migration").clicked() {
let _ = self.cmd_tx.send(AsyncCmd::SimulateScenario {
scenario: SimulateScenario::Migration,
});
}
if ui.button("ForceTcp").clicked() {
let _ = self.cmd_tx.send(AsyncCmd::SimulateScenario {
scenario: SimulateScenario::ForceTcp,
});
}
if ui.button("ForceTls").clicked() {
let _ = self.cmd_tx.send(AsyncCmd::SimulateScenario {
scenario: SimulateScenario::ForceTls,
});
}
});
});
});
egui::SidePanel::right("room_panel")
.default_width(256.0)
.show(ui.ctx(), |ui| {
ui.heading("Livekit - Connect to a room");
ui.separator();
ui.horizontal(|ui| {
ui.label("URL: ");
ui.text_edit_singleline(&mut self.lk_url);
});
ui.horizontal(|ui| {
ui.label("Token: ");
ui.text_edit_singleline(&mut self.lk_token);
});
ui.horizontal(|ui| {
let connecting = self.state.connecting.load(Ordering::SeqCst);
ui.set_enabled(!connecting);
if ui.button("Connect").clicked() {
self.connection_failure = None;
let _ = self.cmd_tx.send(AsyncCmd::RoomConnect {
url: self.lk_url.clone(),
token: self.lk_token.clone(),
});
}
if connecting {
ui.spinner();
}
});
if let Some(err) = &self.connection_failure {
ui.colored_label(egui::Color32::RED, err);
}
ui.separator();
});
egui::CentralPanel::default().show(ui.ctx(), |ui| {
egui::ScrollArea::vertical().show(ui, |ui| {
VideoGrid::new("default_grid")
.max_columns(6)
.show(ui, |ui| {
if self.room_state == ConnectionState::Disconnected {
for _ in 0..20 {
ui.video_frame(|ui| {
egui::Frame::none().fill(egui::Color32::DARK_GRAY).show(
ui,
|ui| {
ui.allocate_space(ui.available_size());
},
);
});
}
} else {
for video_renderer in &self.video_renderers {
ui.video_frame(|ui| {
if let Some(tex) = video_renderer.texture_id() {
ui.painter().image(
tex,
ui.available_rect_before_wrap(),
egui::Rect::from_min_max(
egui::pos2(0.0, 0.0),
egui::pos2(1.0, 1.0),
),
egui::Color32::WHITE,
);
}
});
}
}
});
});
});
}
fn render(&mut self) {
self.egui_state
.set_pixels_per_point(egui_winit::native_pixels_per_point(&self.window));
let raw_inputs = self.egui_state.take_egui_input(&self.window);
let full_output = self.egui_context.clone().run(raw_inputs, |ctx| {
egui::CentralPanel::default().show(ctx, |ui| {
self.ui(ui);
});
});
let clipped_primitives = self.egui_context.tessellate(full_output.shapes);
self.egui_painter.paint_and_update_textures(
egui_winit::native_pixels_per_point(&self.window),
egui::Rgba::BLACK,
&clipped_primitives,
&full_output.textures_delta,
);
self.egui_state.handle_platform_output(
&self.window,
&self.egui_context,
full_output.platform_output,
);
}
}
examples/simple_room/src/events.rs
+17
View File
@@ -0,0 +1,17 @@
use livekit::{events::TrackSubscribedEvent, room::SimulateScenario};
#[derive(Debug)]
pub enum AsyncCmd {
RoomConnect { url: String, token: String },
SimulateScenario { scenario: SimulateScenario }
}
#[derive(Debug)]
pub enum UiCmd {
ConnectResult {
result: livekit::room::RoomResult<()>,
},
TrackSubscribed {
event: TrackSubscribedEvent,
},
}
examples/simple_room/src/main.rs
+6 -183
View File
@@ -1,192 +1,15 @@
use std::convert::TryInto;
use std::ops::DerefMut;
use std::{num::NonZeroU32, time::Duration};
use egui_wgpu::WgpuConfiguration;
use livekit::webrtc::media_stream::VideoTrack;
use livekit::webrtc::video_frame_buffer::{
PlanarYuv8Buffer, PlanarYuvBuffer, VideoFrameBufferTrait,
};
use livekit::webrtc::yuv_helper;
use std::sync::{Arc, Mutex};
use video_renderer::VideoRenderer;
use wgpu::{Device, Queue};
use tokio::time::sleep;
use livekit::room::track::remote_track::RemoteTrackHandle;
use livekit::room::{Room, RoomError};
const URL: &str = "ws://localhost:7880";
const TOKEN : &str = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjIzODQ4MDY0NzMsImlzcyI6IkFQSXpLYkFTaUNWYWtnSiIsIm5hbWUiOiJuYXRpdmUiLCJuYmYiOjE2NjQ4MDY0NzMsInN1YiI6Im5hdGl2ZSIsInZpZGVvIjp7InJvb21DcmVhdGUiOnRydWUsInJvb21Kb2luIjp0cnVlfX0.BgVdBnq3XFD3_BQHoe1azqjifYysubgFl6Qlzu9IQGI";
// eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjIzODQ4MDY3MzAsImlzcyI6IkFQSXpLYkFTaUNWYWtnSiIsIm5hbWUiOiJ3ZWIiLCJuYmYiOjE2NjQ4MDY3MzAsInN1YiI6IndlYiIsInZpZGVvIjp7InJvb21DcmVhdGUiOnRydWUsInJvb21Kb2luIjp0cnVlfX0.VbDoULjX1CVGZu2sPy3SvWYlVZUBXxQVPmdB9BnmlN4
mod app;
mod events;
mod video_grid;
mod video_renderer;
use winit::{
event::*,
event_loop::{ControlFlow, EventLoop},
window::{Window, WindowBuilder, WindowId},
};
struct AppState {
room: Room,
demo: egui_demo_lib::DemoWindows,
egui_context: egui::Context,
egui_state: egui_winit::State,
egui_painter: egui_wgpu::winit::Painter,
window: winit::window::Window,
}
impl AppState {
fn on_event<T>(&mut self, event: Event<'_, T>, control_flow: &mut ControlFlow) {
match event {
Event::WindowEvent { window_id, event } => {
if let Some(flow) = self.on_window_event(window_id, event) {
*control_flow = flow;
}
}
Event::RedrawRequested(window_id) if window_id == self.window.id() => {
self.render();
}
Event::RedrawEventsCleared => {
self.window.request_redraw();
}
_ => {}
};
}
fn on_window_event(
&mut self,
_window_id: WindowId,
event: WindowEvent<'_>,
) -> Option<ControlFlow> {
if self
.egui_state
.on_event(&self.egui_context, &event)
.consumed
{
return None;
}
match event {
WindowEvent::CloseRequested => Some(ControlFlow::Exit),
WindowEvent::Resized(inner_size) => {
self.egui_painter
.on_window_resized(inner_size.width, inner_size.height);
None
}
WindowEvent::ScaleFactorChanged { new_inner_size, .. } => {
self.egui_painter
.on_window_resized(new_inner_size.width, new_inner_size.height);
None
}
_ => None,
}
}
fn render(&mut self) {
let raw_inputs = self.egui_state.take_egui_input(&self.window);
let full_output = self.egui_context.run(raw_inputs, |ctx| {
//self.ui(ctx);
});
let clipped_primitives = self.egui_context.tessellate(full_output.shapes);
self.egui_painter.paint_and_update_textures(
egui_winit::native_pixels_per_point(&self.window),
egui::Rgba::BLACK,
&clipped_primitives,
&full_output.textures_delta,
);
self.egui_state.handle_platform_output(
&self.window,
&self.egui_context,
full_output.platform_output,
);
}
}
struct App {
rt: tokio::runtime::Runtime,
}
impl App {
pub fn new(rt: tokio::runtime::Runtime) -> Self {
Self { rt }
}
pub fn run(&mut self) {
self.rt.block_on(async {
let event_loop = EventLoop::new();
let window = WindowBuilder::new().build(&event_loop).unwrap();
let egui_context = egui::Context::default();
let egui_state = egui_winit::State::new(&event_loop);
let mut egui_painter =
egui_wgpu::winit::Painter::new(WgpuConfiguration::default(), 1, 32);
unsafe {
egui_painter.set_window(Some(&window));
}
let mut inner = AppState {
room: Room::new(),
demo: egui_demo_lib::DemoWindows::default(),
egui_context,
egui_state,
egui_painter,
window,
};
inner
.room
.events()
.on_participant_connected(|_event| async move {});
inner.room.events().on_track_subscribed({
let test = Arc::new(Mutex::new(None));
let egui_render = inner.egui_painter.render_state().clone().unwrap();
move |event| {
let test = test.clone();
let egui_render = egui_render.clone();
async move {
let track = event.publication.track().unwrap();
if let RemoteTrackHandle::Video(video_track) = track {
*test.lock().unwrap() =
Some(VideoRenderer::new(egui_render, video_track.rtc_track()))
}
}
}
});
inner.room.connect(URL, TOKEN).await.unwrap();
tokio::spawn(async {
loop {
println!("Test");
tokio::time::sleep(Duration::from_secs(5)).await;
}
});
tokio::task::block_in_place(move || loop {
event_loop.run(move |event, _, control_flow| {
inner.on_event(event, control_flow);
});
});
});
}
}
fn main() {
tracing_subscriber::fmt::init();
let rt = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.unwrap();
let mut app = App::new(rt);
app.run();
app::run(rt);
}
examples/simple_room/src/video_grid.rs
+175
View File
@@ -0,0 +1,175 @@
use std::cmp;
#[derive(Debug, Clone, Default, PartialEq)]
struct State {
num_videos: u32,
}
impl State {
pub fn load(ctx: &egui::Context, id: egui::Id) -> Option<Self> {
ctx.data().get_temp(id)
}
pub fn store(self, ctx: &egui::Context, id: egui::Id) {
ctx.data().insert_temp(id, self);
}
}
pub const DEFAULT_VIDEO_SIZE: egui::Vec2 = egui::vec2(320.0, 180.0);
pub const DEFAULT_MAX_COLUMNS: u32 = 4;
pub const DEFAULT_SPACING: f32 = 16.0;
pub struct VideoGrid {
id: egui::Id,
// Current frame
available_rect: egui::Rect,
prev_state: State,
curr_state: State,
video_index: u32, // Kinda "cursor"
// Options
min_video_size: egui::Vec2,
max_columns: u32,
spacing: f32,
}
impl VideoGrid {
pub fn new(id_source: impl std::hash::Hash) -> Self {
Self {
id: egui::Id::new(id_source),
available_rect: egui::Rect::NAN,
prev_state: State::default(),
curr_state: State::default(),
video_index: 0,
min_video_size: DEFAULT_VIDEO_SIZE,
max_columns: DEFAULT_MAX_COLUMNS,
spacing: DEFAULT_SPACING,
}
}
pub fn show<R>(
mut self,
ui: &mut egui::Ui,
grid: impl FnOnce(&mut VideoGridContext) -> R,
) -> egui::InnerResponse<R> {
// TODO(theomonnom): Should I care about the current egui layout?
let prev_state = State::load(ui.ctx(), self.id);
let is_first_frame = prev_state.is_none();
self.prev_state = prev_state.unwrap_or_default();
self.available_rect = ui.available_rect_before_wrap();
ui.ctx()
.check_for_id_clash(self.id, self.available_rect, "VideoGrid");
ui.allocate_ui_at_rect(self.available_rect, |ui| {
ui.set_visible(!is_first_frame);
let mut ctx = VideoGridContext {
layout: &mut self,
ui,
};
let res = grid(&mut ctx);
// Save the new state
if self.curr_state != self.prev_state {
self.curr_state.clone().store(ui.ctx(), self.id);
ui.ctx().request_repaint();
}
res
})
}
fn next_frame_rect(&mut self) -> egui::Rect {
assert!(self.available_rect.is_finite());
assert!(self.spacing <= self.min_video_size.x);
// increment the amount of videos for the next frame
self.curr_state.num_videos += 1;
let num_videos = self.prev_state.num_videos;
if num_videos == 0 {
return egui::Rect::NOTHING;
}
let max_columns = self.max_columns;
let minimum_size = self.min_video_size;
let available_size = self.available_rect.size();
let calc_min_width =
|columns: u32| columns as f32 * minimum_size.x + (columns - 1) as f32 * self.spacing;
let total_columns = {
let mut est = (available_size.x / minimum_size.x) as u32 + 1;
if available_size.x < calc_min_width(est) {
est -= 1;
}
cmp::max(1, cmp::min(est, max_columns))
};
let aspect_ratio = minimum_size.x / minimum_size.y;
let remaining_width = available_size.x - calc_min_width(total_columns);
let w = minimum_size.x + remaining_width / total_columns as f32;
let h = w / aspect_ratio;
let x_index = self.video_index % total_columns;
let y_index = self.video_index / total_columns;
let x = {
let mut x = x_index as f32 * (w + self.spacing);
// vertically center the last row
let total_rows = num_videos / total_columns + 1;
if (y_index + 1) == total_rows {
let nb_items = num_videos - (total_rows - 1) * total_columns; // nb. of items on the last row
x += (total_columns - nb_items) as f32 * (w + self.spacing) / 2.0;
}
x
};
let y = y_index as f32 * (h + self.spacing);
let min = egui::pos2(x, y) + self.available_rect.left_top().to_vec2();
let max = egui::pos2(w, h) + min.to_vec2();
self.video_index += 1;
egui::Rect { min, max }
}
}
impl VideoGrid {
pub fn min_video_size(mut self, min_video_size: egui::Vec2) -> Self {
self.min_video_size = min_video_size;
self
}
pub fn max_columns(mut self, max_columns: u32) -> Self {
self.max_columns = max_columns;
self
}
pub fn spacing(mut self, spacing: f32) -> Self {
self.spacing = spacing;
self
}
}
pub struct VideoGridContext<'a> {
layout: &'a mut VideoGrid,
ui: &'a mut egui::Ui,
}
impl<'a> VideoGridContext<'a> {
pub fn video_frame(&mut self, add_contents: impl FnOnce(&mut egui::Ui)) -> egui::Response {
let frame_rect = self.layout.next_frame_rect();
let mut child_ui = self.ui.child_ui(frame_rect, egui::Layout::default());
add_contents(&mut child_ui);
self.ui.allocate_rect(frame_rect, egui::Sense::hover())
}
}
examples/simple_room/src/video_renderer.rs
+4 -2
View File
@@ -3,6 +3,7 @@ use livekit::webrtc::video_frame_buffer::PlanarYuv8Buffer;
use livekit::webrtc::video_frame_buffer::PlanarYuvBuffer;
use livekit::webrtc::video_frame_buffer::VideoFrameBufferTrait;
use livekit::webrtc::yuv_helper;
use tracing::debug_span;
use std::convert::TryInto;
use std::num::NonZeroU32;
use std::{
@@ -101,6 +102,9 @@ impl VideoRenderer {
let internal = internal.clone();
Box::new(move |_frame, buffer| {
let span = debug_span!("texture_upload");
let _enter = span.enter();
let mut internal = internal.lock().unwrap();
let buffer = buffer.to_i420();
@@ -149,8 +153,6 @@ impl VideoRenderer {
copy_layout,
copy_size,
);
println!("wrote");
})
});