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

trying to receive VideoFrame, progress...

Browse Source
This commit is contained in:
Théo Monnom
2022-11-28 15:52:37 +01:00
parent a415748f06
commit 39fa798c7f
20 changed files with 572 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/livekit-core/src/events.rs
+1 -1
View File
@@ -8,7 +8,7 @@ macro_rules! event_setter {
pub fn $fnc<F, Fut>(&self, mut callback: F)
where
F: FnMut($event) -> Fut + Send + Sync + 'static,
Fut: Future<Output = ()> + Send + Sync + 'static,
Fut: Future<Output = ()> + Send + 'static,
{
*self.$fnc.lock() = Some(Box::new(move |event| Box::pin(callback(event))));
}
crates/livekit-core/src/room/mod.rs
+47 -26
View File
@@ -9,11 +9,13 @@ 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};
use crate::events::room::{
ParticipantConnectedEvent, ParticipantDisconnectedEvent, RoomEvents, TrackSubscribedEvent,
};
use crate::proto;
use crate::proto::participant_info;
use thiserror::Error;
use tracing::error;
use tracing::{debug, error};
use crate::rtc_engine::{EngineError, EngineEvent, EngineEvents, RTCEngine};
use crate::signal_client::SignalOptions;
@@ -63,10 +65,6 @@ impl Room {
}
}
pub fn events(&self) -> Arc<RoomEvents> {
self.events.clone()
}
pub async fn connect(&mut self, url: &str, token: &str) -> RoomResult<()> {
let (rtc_engine, engine_events) =
RTCEngine::connect(url, token, SignalOptions::default()).await?;
@@ -88,11 +86,21 @@ impl Room {
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 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(),
@@ -121,6 +129,7 @@ impl Room {
match event {
EngineEvent::ParticipantUpdate(update) => {
Self::handle_participant_update(room_inner.clone(), room_events.clone(), update)
.await
}
EngineEvent::AddTrack {
rtp_receiver,
@@ -164,7 +173,7 @@ impl Room {
Ok(())
}
fn handle_participant_update(
async fn handle_participant_update(
room_inner: Arc<RoomInner>,
room_events: Arc<RoomEvents>,
update: proto::ParticipantUpdate,
@@ -173,7 +182,7 @@ impl Room {
if pi.sid == room_inner.local_participant.sid()
|| pi.identity == room_inner.local_participant.identity()
{
room_inner.local_participant.update_info(pi);
room_inner.local_participant.clone().update_info(pi).await;
continue;
}
@@ -190,12 +199,12 @@ impl Room {
)
} else {
// Participant is already connected, update the informations
remote_participant.update_info(pi);
remote_participant.update_info(pi).await;
}
} else {
// Create a new participant and call OnConnect event
let remote_participant =
Self::get_or_create_participant(room_inner.clone(), room_events.clone(), pi);
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 {
@@ -235,29 +244,41 @@ impl Room {
room_inner.participants.read().get(sid).cloned()
}
fn get_or_create_participant(
fn create_participant(
room_inner: Arc<RoomInner>,
room_events: Arc<RoomEvents>,
pi: proto::ParticipantInfo,
) -> Arc<RemoteParticipant> {
let participants = room_inner.participants.upgradable_read();
let sid = pi.sid.clone().into();
if let Some(p) = participants.get(&sid) {
p.update_info(pi);
p.clone()
} else {
let mut participants = RwLockUpgradableReadGuard::upgrade(participants);
let p = Arc::new(RemoteParticipant::new(pi));
let p = Arc::new(RemoteParticipant::new(pi.clone()));
// Forward participantevents to room events
p.internal_events().on_track_published({
// 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();
|event| async move {}
});
let room_inner = room_inner.clone();
participants.insert(sid, p.clone());
p
}
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
}
}
crates/livekit-core/src/room/participant/local_participant.rs
+11 -1
View File
@@ -1,5 +1,5 @@
use crate::proto::{data_packet, DataPacket, UserPacket};
use crate::room::participant::{impl_participant_trait, ParticipantShared};
use crate::room::participant::{impl_participant_trait, ParticipantShared, ParticipantInternalTrait};
use crate::room::RoomError;
use crate::rtc_engine::RTCEngine;
@@ -40,6 +40,16 @@ 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()
}
}
impl_participant_trait!(LocalParticipant);
crates/livekit-core/src/room/participant/mod.rs
+10 -12
View File
@@ -65,7 +65,6 @@ pub trait ParticipantTrait {
fn identity(&self) -> ParticipantIdentity;
fn name(&self) -> String;
fn metadata(&self) -> String;
fn update_info(&self, info: ParticipantInfo);
}
#[derive(Clone)]
@@ -74,6 +73,16 @@ pub enum ParticipantHandle {
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,
}
}
}
impl ParticipantInternalTrait for ParticipantHandle {
wrap_variants!(
[Local, Remote]
@@ -89,7 +98,6 @@ impl ParticipantTrait for ParticipantHandle {
fnc!(identity, ParticipantIdentity, []);
fnc!(name, String, []);
fnc!(metadata, String, []);
fnc!(update_info, (), [info: ParticipantInfo]);
);
}
@@ -100,12 +108,6 @@ macro_rules! impl_participant_trait {
use crate::room::id::{ParticipantIdentity, ParticipantSid};
use std::sync::Arc;
impl crate::room::participant::ParticipantInternalTrait for $x {
fn internal_events(&self) -> Arc<ParticipantEvents> {
self.shared.internal_events.clone()
}
}
impl crate::room::participant::ParticipantTrait for $x {
fn events(&self) -> Arc<ParticipantEvents> {
self.shared.events.clone()
@@ -126,10 +128,6 @@ macro_rules! impl_participant_trait {
fn metadata(&self) -> String {
self.shared.metadata.lock().clone()
}
fn update_info(&self, info: ParticipantInfo) {
self.shared.update_info(info);
}
}
};
}
crates/livekit-core/src/room/participant/remote_participant.rs
+61 -5
View File
@@ -1,16 +1,25 @@
use crate::events::participant::{TrackSubscribedEvent, TrackSubscriptionFailedEvent};
use crate::events::participant::{
TrackPublishedEvent, TrackSubscribedEvent, TrackSubscriptionFailedEvent,
};
use crate::events::TrackError;
use crate::room::id::TrackSid;
use crate::room::participant::{impl_participant_trait, ParticipantShared};
use crate::room::publication::{RemoteTrackPublication, TrackPublication, TrackPublicationTrait};
use crate::room::participant::{
impl_participant_trait, ParticipantInternalTrait, ParticipantShared,
};
use crate::room::publication::{
RemoteTrackPublication, TrackPublication, TrackPublicationInternalTrait, TrackPublicationTrait,
};
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::{TrackTrait, TrackKind};
use crate::room::track::{TrackKind, TrackTrait, TrackHandle};
use livekit_webrtc::media_stream::MediaStreamTrackHandle;
use std::collections::HashSet;
use std::time::Duration;
use tokio::time::{sleep, timeout};
use tracing::error;
use tracing::{info, error};
use super::ParticipantTrait;
const ADD_TRACK_TIMEOUT: Duration = Duration::from_secs(5);
@@ -80,6 +89,9 @@ impl RemoteParticipant {
_ => 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();
@@ -144,6 +156,50 @@ impl RemoteParticipant {
}
})
}
pub(crate) async fn update_info(self: Arc<Self>, info: ParticipantInfo) {
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());
} else {
let publication = RemoteTrackPublication::new(track.clone(), self.sid(), None);
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;
}
}
valid_tracks.insert(track.sid.into());
}
}
}
impl ParticipantInternalTrait for RemoteParticipant {
fn internal_events(&self) -> Arc<ParticipantEvents> {
self.shared.internal_events.clone()
}
}
impl_participant_trait!(RemoteParticipant);
crates/livekit-core/src/room/publication/mod.rs
+81 -16
View File
@@ -1,11 +1,22 @@
use crate::proto::TrackType;
use crate::proto::{TrackInfo, TrackSource as ProtoTrackSource};
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::utils::wrap_variants;
use parking_lot::Mutex;
use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
use std::sync::Arc;
use super::track::TrackDimension;
pub(crate) trait TrackPublicationInternalTrait {
fn update_track(&self, track: Option<TrackHandle>);
fn update_info(&self, info: TrackInfo);
}
pub trait TrackPublicationTrait {
fn name(&self) -> String;
fn sid(&self) -> TrackSid;
@@ -21,6 +32,43 @@ pub(super) struct TrackPublicationShared {
pub(super) kind: AtomicU8, // Casted to TrackKind
pub(super) source: AtomicU8, // Casted to TrackSource
pub(super) simulcasted: AtomicBool,
pub(super) dimension: Mutex<TrackDimension>,
pub(super) mime_type: Mutex<String>,
pub(super) participant: ParticipantSid, // TODO(theomonnom) Use WeakParticipant instead
}
impl TrackPublicationShared {
pub fn new(info: TrackInfo, participant: ParticipantSid, track: Option<TrackHandle>) -> Arc<Self> {
Arc::new(Self {
track: Mutex::new(track),
name: Mutex::new(info.name),
sid: Mutex::new(info.sid.into()),
kind: AtomicU8::new(TrackKind::from(TrackType::from_i32(info.r#type).unwrap()) as u8),
source: AtomicU8::new(TrackSource::from(
ProtoTrackSource::from_i32(info.source).unwrap(),
) as u8),
simulcasted: AtomicBool::new(info.simulcast),
dimension: Mutex::new(TrackDimension(info.width, info.height)),
mime_type: Mutex::new(info.mime_type),
participant,
})
}
pub fn update_info(&self, info: TrackInfo) {
*self.name.lock() = info.name;
*self.sid.lock() = info.sid.into();
self.kind.store(
TrackKind::from(TrackType::from_i32(info.r#type).unwrap()) as u8,
Ordering::SeqCst,
);
self.source.store(
TrackSource::from(ProtoTrackSource::from_i32(info.source).unwrap()) as u8,
Ordering::SeqCst,
);
self.simulcasted.store(info.simulcast, Ordering::SeqCst);
*self.dimension.lock() = TrackDimension(info.width, info.height);
*self.mime_type.lock() = info.mime_type;
}
}
#[derive(Clone)]
@@ -29,19 +77,9 @@ pub enum TrackPublication {
Remote(RemoteTrackPublication),
}
macro_rules! shared_getter {
($x:ident, $ret:ty) => {
fn $x(&self) -> $ret {
match self {
TrackPublication::Local(p) => p.$x(),
TrackPublication::Remote(p) => p.$x(),
}
}
};
}
impl TrackPublication {
pub fn track(&self) -> Option<TrackHandle> {
// Not calling Local/Remote function here, we don't need "cast"
match self {
TrackPublication::Local(p) => p.shared.track.lock().clone(),
TrackPublication::Remote(p) => p.shared.track.lock().clone(),
@@ -49,16 +87,37 @@ impl TrackPublication {
}
}
impl TrackPublicationInternalTrait for TrackPublication {
wrap_variants!(
[Local, Remote]
fnc!(update_track, (), [track: Option<TrackHandle>]);
fnc!(update_info, (), [info: TrackInfo]);
);
}
impl TrackPublicationTrait for TrackPublication {
shared_getter!(name, String);
shared_getter!(sid, TrackSid);
shared_getter!(kind, TrackKind);
shared_getter!(source, TrackSource);
shared_getter!(simulcasted, bool);
wrap_variants!(
[Local, Remote]
fnc!(sid, TrackSid, []);
fnc!(name, String, []);
fnc!(kind, TrackKind, []);
fnc!(source, TrackSource, []);
fnc!(simulcasted, 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()
@@ -104,6 +163,12 @@ pub struct RemoteTrackPublication {
}
impl RemoteTrackPublication {
pub fn new(info: TrackInfo, participant: ParticipantSid, track: Option<TrackHandle>) -> Self {
Self {
shared: TrackPublicationShared::new(info, participant, track),
}
}
pub fn track(&self) -> Option<RemoteTrackHandle> {
self.shared
.track
crates/livekit-core/src/room/track/mod.rs
+26 -1
View File
@@ -1,3 +1,4 @@
use crate::proto::{TrackSource as ProtoTrackSource, TrackType};
use crate::room::id::TrackSid;
use crate::room::track::local_audio_track::LocalAudioTrack;
use crate::room::track::local_video_track::LocalVideoTrack;
@@ -36,6 +37,16 @@ impl From<u8> for TrackKind {
}
}
impl From<TrackType> for TrackKind {
fn from(r#type: TrackType) -> Self {
match r#type {
TrackType::Audio => Self::Audio,
TrackType::Video => Self::Video,
_ => Self::Unknown,
}
}
}
#[derive(Debug)]
pub enum StreamState {
Unknown,
@@ -74,6 +85,20 @@ impl From<u8> for TrackSource {
}
}
impl From<ProtoTrackSource> for TrackSource {
fn from(source: ProtoTrackSource) -> Self {
match source {
ProtoTrackSource::Camera => Self::Camera,
ProtoTrackSource::Microphone => Self::Microphone,
ProtoTrackSource::ScreenShare => Self::Screenshare,
ProtoTrackSource::ScreenShareAudio => Self::ScreenshareAudio,
ProtoTrackSource::Unknown => Self::Unknown,
}
}
}
pub struct TrackDimension(pub u32, pub u32);
pub trait TrackTrait {
fn sid(&self) -> TrackSid;
fn name(&self) -> String;
@@ -103,7 +128,7 @@ impl TrackShared {
name: Mutex::new(name),
kind: AtomicU8::new(kind as u8),
stream_state: AtomicU8::new(StreamState::Active as u8),
rtc_track: rtc_track,
rtc_track,
}
}
crates/livekit-core/src/rtc_engine/mod.rs
+10 -7
View File
@@ -10,7 +10,7 @@ use prost::Message;
use serde::{Deserialize, Serialize};
use thiserror::Error;
use tokio::time::sleep;
use tracing::{debug, error, trace};
use tracing::{debug, error, info, trace};
use crate::{proto, signal_client};
use livekit_webrtc::data_channel::{DataChannel, DataChannelInit, DataSendError, DataState};
@@ -280,6 +280,7 @@ impl RTCEngine {
}
RTCEvent::ConnectionChange { state, target } => {
// Reconnect if we've been disconnected unexpectedly
trace!("Connection change, {:?} {:?}", state, target);
let subscriber_primary = engine_inner.join_response.lock().subscriber_primary;
let is_primary = subscriber_primary && target == SignalTarget::Subscriber;
@@ -328,10 +329,12 @@ impl RTCEngine {
target,
} => {
if target == SignalTarget::Subscriber {
let _ = emitter.send(EngineEvent::AddTrack {
rtp_receiver,
streams,
});
let _ = emitter
.send(EngineEvent::AddTrack {
rtp_receiver,
streams,
})
.await;
}
}
RTCEvent::Data { data, binary } => {
@@ -377,7 +380,7 @@ impl RTCEngine {
signal_response::Message::Offer(offer) => {
// Handle the subscriber offer & send an answer to livekit-server
// We always get an offer from the server when connecting
trace!("received offer from the publisher: {:?}", offer);
trace!("received offer for the subscriber: {:?}", offer);
let sdp = SessionDescription::from(offer.r#type.parse().unwrap(), &offer.sdp)?;
engine_inner
@@ -438,7 +441,7 @@ impl RTCEngine {
}
}
signal_response::Message::Update(update) => {
let _ = emitter.send(EngineEvent::ParticipantUpdate(update));
let _ = emitter.send(EngineEvent::ParticipantUpdate(update)).await;
}
_ => {}
}