From 4c5828b9c04dde821e74ae8fe0b852d0df1cf358 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Monnom?= Date: Tue, 10 Jan 2023 22:57:27 +0100 Subject: [PATCH] added prelude & cleaned imports (#27) * wip * wip * wip * wip * wip * wip --- examples/simple_room/src/:w | 15 ++ examples/simple_room/src/app.rs | 8 +- examples/simple_room/src/events.rs | 3 +- examples/simple_room/src/video_renderer.rs | 6 +- livekit-webrtc/src/data_channel.rs | 2 +- livekit-webrtc/src/lib.rs | 2 + livekit-webrtc/src/prelude.rs | 18 ++ livekit/src/lib.rs | 7 +- livekit/src/prelude.rs | 26 +++ livekit/src/room/mod.rs | 24 +- .../src/room/participant/local_participant.rs | 21 +- livekit/src/room/participant/mod.rs | 26 +-- .../room/participant/remote_participant.rs | 26 +-- livekit/src/room/publication/mod.rs | 37 ++-- livekit/src/room/room_session.rs | 15 +- livekit/src/room/track/audio_track.rs | 4 +- livekit/src/room/track/local_audio_track.rs | 2 +- livekit/src/room/track/local_track.rs | 4 +- livekit/src/room/track/local_video_track.rs | 2 +- livekit/src/room/track/mod.rs | 47 ++-- livekit/src/room/track/remote_audio_track.rs | 4 +- livekit/src/room/track/remote_track.rs | 12 +- livekit/src/room/track/remote_video_track.rs | 5 +- livekit/src/room/track/video_track.rs | 4 +- livekit/src/rtc_engine/lk_runtime.rs | 6 +- livekit/src/rtc_engine/mod.rs | 49 ++-- livekit/src/rtc_engine/pc_transport.rs | 17 +- livekit/src/rtc_engine/rtc_events.rs | 39 ++-- livekit/src/rtc_engine/rtc_session.rs | 209 +++++++++--------- livekit/src/signal_client/mod.rs | 20 +- livekit/src/signal_client/signal_stream.rs | 14 +- 31 files changed, 345 insertions(+), 329 deletions(-) create mode 100644 examples/simple_room/src/:w create mode 100644 livekit-webrtc/src/prelude.rs create mode 100644 livekit/src/prelude.rs diff --git a/examples/simple_room/src/:w b/examples/simple_room/src/:w new file mode 100644 index 0000000..64239fc --- /dev/null +++ b/examples/simple_room/src/:w @@ -0,0 +1,15 @@ +use livekit::prelude::*; +use livekit::{RoomResult, SimulateScenario}; + +#[derive(Debug)] +pub enum AsyncCmd { + RoomConnect { url: String, token: String }, + RoomDisconnect, + SimulateScenario { scenario: SimulateScenario }, +} + +#[derive(Debug)] +pub enum UiCmd { + ConnectResult { result: RoomResult<()> }, + RoomEvent { event: RoomEvent }, +} diff --git a/examples/simple_room/src/app.rs b/examples/simple_room/src/app.rs index 1333701..ec8638e 100644 --- a/examples/simple_room/src/app.rs +++ b/examples/simple_room/src/app.rs @@ -3,12 +3,8 @@ use crate::video_renderer::VideoRenderer; use crate::{events::AsyncCmd, video_grid::VideoGrid}; use egui::{Rounding, Stroke}; use egui_wgpu::WgpuConfiguration; -use livekit::room::id::{ParticipantSid, TrackSid}; -use livekit::room::participant::ParticipantTrait; -use livekit::room::room_session::ConnectionState; -use livekit::room::track::remote_track::RemoteTrackHandle; -use livekit::room::track::TrackTrait; -use livekit::room::{Room, RoomEvent, RoomEvents, SimulateScenario}; +use livekit::prelude::*; +use livekit::SimulateScenario; use parking_lot::Mutex; use std::collections::HashMap; use std::sync::{ diff --git a/examples/simple_room/src/events.rs b/examples/simple_room/src/events.rs index 8798b94..64239fc 100644 --- a/examples/simple_room/src/events.rs +++ b/examples/simple_room/src/events.rs @@ -1,4 +1,5 @@ -use livekit::room::{RoomEvent, RoomResult, SimulateScenario}; +use livekit::prelude::*; +use livekit::{RoomResult, SimulateScenario}; #[derive(Debug)] pub enum AsyncCmd { diff --git a/examples/simple_room/src/video_renderer.rs b/examples/simple_room/src/video_renderer.rs index 4956a3b..587423e 100644 --- a/examples/simple_room/src/video_renderer.rs +++ b/examples/simple_room/src/video_renderer.rs @@ -1,7 +1,5 @@ -use livekit::webrtc::media_stream::VideoTrack; -use livekit::webrtc::video_frame_buffer::PlanarYuv8Buffer; -use livekit::webrtc::video_frame_buffer::PlanarYuvBuffer; -use livekit::webrtc::video_frame_buffer::VideoFrameBufferTrait; +use livekit::prelude::*; +use livekit::webrtc::video_frame_buffer::*; use livekit::webrtc::yuv_helper; use std::convert::TryInto; use std::num::NonZeroU32; diff --git a/livekit-webrtc/src/data_channel.rs b/livekit-webrtc/src/data_channel.rs index 409283c..5395b03 100644 --- a/livekit-webrtc/src/data_channel.rs +++ b/livekit-webrtc/src/data_channel.rs @@ -1,6 +1,6 @@ use std::error::Error; use std::fmt::{Debug, Display, Formatter}; -use std::sync::{Arc, Mutex}; +use std::sync::Mutex; use cxx::UniquePtr; use log::trace; diff --git a/livekit-webrtc/src/lib.rs b/livekit-webrtc/src/lib.rs index 951b6ac..f5372a8 100644 --- a/livekit-webrtc/src/lib.rs +++ b/livekit-webrtc/src/lib.rs @@ -10,3 +10,5 @@ pub mod video_frame; pub mod video_frame_buffer; pub mod webrtc; pub mod yuv_helper; + +pub mod prelude; diff --git a/livekit-webrtc/src/prelude.rs b/livekit-webrtc/src/prelude.rs new file mode 100644 index 0000000..c78e396 --- /dev/null +++ b/livekit-webrtc/src/prelude.rs @@ -0,0 +1,18 @@ +pub use crate::data_channel::{DataChannel, DataChannelInit, DataState}; +pub use crate::jsep::{IceCandidate, SessionDescription}; +pub use crate::media_stream::{ + AudioTrack, MediaStream, MediaStreamTrackHandle, MediaStreamTrackTrait, VideoTrack, +}; +pub use crate::peer_connection::{ + IceConnectionState, IceGatheringState, PeerConnection, PeerConnectionState, + RTCOfferAnswerOptions, SignalingState, +}; +pub use crate::peer_connection_factory::{ + ContinualGatheringPolicy, ICEServer, IceTransportsType, PeerConnectionFactory, RTCConfiguration, +}; +pub use crate::rtc_error::RTCError; +pub use crate::rtp_receiver::RtpReceiver; +pub use crate::rtp_transceiver::RtpTransceiver; +pub use crate::video_frame::{VideoFrame, VideoRotation}; +pub use crate::video_frame_buffer::*; +pub use crate::webrtc::RTCRuntime; diff --git a/livekit/src/lib.rs b/livekit/src/lib.rs index 048b964..d6f131a 100644 --- a/livekit/src/lib.rs +++ b/livekit/src/lib.rs @@ -6,8 +6,13 @@ pub mod proto { mod rtc_engine; mod signal_client; +mod room; -pub mod room; pub mod webrtc { pub use livekit_webrtc::*; } + +pub use room::*; + +/// `use livekit::prelude::*;` to import livekit types +pub mod prelude; diff --git a/livekit/src/prelude.rs b/livekit/src/prelude.rs new file mode 100644 index 0000000..a93c996 --- /dev/null +++ b/livekit/src/prelude.rs @@ -0,0 +1,26 @@ +pub use crate::participant::{ + LocalParticipant, Participant, ParticipantEvent, ParticipantTrait, RemoteParticipant, +}; + +pub use crate::{ConnectionState, Room, RoomError, RoomEvent, RoomEvents, RoomSession}; + +pub use crate::publication::{ + LocalTrackPublication, RemoteTrackPublication, TrackPublication, TrackPublicationTrait, +}; + +pub use crate::track::{ + AudioTrackHandle, LocalAudioTrack, LocalTrackHandle, LocalVideoTrack, RemoteAudioTrack, + RemoteTrackHandle, RemoteVideoTrack, StreamState, TrackEvent, TrackHandle, TrackKind, + TrackSource, TrackTrait, VideoTrackHandle, +}; + +pub use crate::id::*; + +pub use crate::webrtc::{ + data_channel::DataChannel, + media_stream::{ + AudioTrack, MediaStream, MediaStreamTrackHandle, MediaStreamTrackTrait, VideoTrack, + }, + rtp_receiver::RtpReceiver, + rtp_transceiver::RtpTransceiver, +}; diff --git a/livekit/src/room/mod.rs b/livekit/src/room/mod.rs index e4b4a8d..db4c611 100644 --- a/livekit/src/room/mod.rs +++ b/livekit/src/room/mod.rs @@ -1,12 +1,6 @@ -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::prelude::*; +use crate::participant::ConnectionQuality; +use crate::proto; use crate::rtc_engine::EngineError; use std::fmt::Debug; use std::sync::Arc; @@ -21,6 +15,8 @@ pub mod publication; pub mod room_session; pub mod track; +pub use room_session::*; + pub type RoomEvents = mpsc::UnboundedReceiver; pub type RoomEmitter = mpsc::UnboundedSender; pub type RoomResult = Result; @@ -33,12 +29,6 @@ pub enum RoomError { Internal(String), } -#[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), @@ -62,7 +52,7 @@ pub enum RoomEvent { participant: Arc, }, TrackSubscriptionFailed { - error: TrackError, + error: track::TrackError, sid: TrackSid, participant: Arc, }, @@ -83,7 +73,7 @@ pub enum RoomEvent { }, DataReceived { payload: Arc>, - kind: data_packet::Kind, + kind: proto::data_packet::Kind, participant: Arc, }, ConnectionStateChanged(ConnectionState), diff --git a/livekit/src/room/participant/local_participant.rs b/livekit/src/room/participant/local_participant.rs index e2f2d5d..8527be7 100644 --- a/livekit/src/room/participant/local_participant.rs +++ b/livekit/src/room/participant/local_participant.rs @@ -1,12 +1,9 @@ -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 super::{ + impl_participant_trait, ConnectionQuality, ParticipantInternalTrait, ParticipantShared, }; -use crate::room::publication::TrackPublication; -use crate::room::RoomError; +use crate::prelude::*; +use crate::proto; +use crate::publication::TrackPublication; use crate::rtc_engine::RTCEngine; use parking_lot::RwLockReadGuard; use std::collections::HashMap; @@ -37,11 +34,11 @@ impl LocalParticipant { pub async fn publish_data( &self, data: &[u8], - kind: data_packet::Kind, + kind: proto::data_packet::Kind, ) -> Result<(), RoomError> { - let data = DataPacket { + let data = proto::DataPacket { kind: kind as i32, - value: Some(data_packet::Value::User(UserPacket { + value: Some(proto::data_packet::Value::User(proto::UserPacket { participant_sid: self.sid().to_string(), payload: data.to_vec(), destination_sids: vec![], @@ -56,7 +53,7 @@ impl LocalParticipant { } impl ParticipantInternalTrait for LocalParticipant { - fn update_info(self: &Arc, info: ParticipantInfo, _emit_events: bool) { + fn update_info(self: &Arc, info: proto::ParticipantInfo, _emit_events: bool) { self.shared.update_info(info); } diff --git a/livekit/src/room/participant/mod.rs b/livekit/src/room/participant/mod.rs index 2c2b71f..318a9aa 100644 --- a/livekit/src/room/participant/mod.rs +++ b/livekit/src/room/participant/mod.rs @@ -1,23 +1,19 @@ -use super::publication::RemoteTrackPublication; -use super::TrackError; +use crate::prelude::*; 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 crate::room::track::remote_track::RemoteTrackHandle; +use crate::track::TrackError; use livekit_utils::enum_dispatch; 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; +mod local_participant; +mod remote_participant; + +pub use local_participant::*; +pub use remote_participant::*; #[derive(Debug, Clone)] pub enum ParticipantEvent { @@ -41,7 +37,7 @@ pub enum ParticipantEvent { }, DataReceived { payload: Arc>, - kind: data_packet::Kind, + kind: proto::data_packet::Kind, }, SpeakingChanged { speaking: bool, @@ -120,7 +116,7 @@ impl ParticipantShared { } } - pub(crate) fn update_info(&self, info: ParticipantInfo) { + pub(crate) fn update_info(&self, info: proto::ParticipantInfo) { *self.sid.lock() = info.sid.into(); *self.identity.lock() = info.identity.into(); *self.name.lock() = info.name; @@ -154,7 +150,7 @@ pub(crate) trait ParticipantInternalTrait { 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, info: ParticipantInfo, emit_events: bool); + fn update_info(self: &Arc, info: proto::ParticipantInfo, emit_events: bool); } pub trait ParticipantTrait { @@ -180,7 +176,7 @@ pub enum Participant { impl Participant { enum_dispatch!( [Local, Remote] - fnc!(pub(crate), update_info, &Self, [info: ParticipantInfo, emit_events: bool], ()); + fnc!(pub(crate), update_info, &Self, [info: proto::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], ()); diff --git a/livekit/src/room/participant/remote_participant.rs b/livekit/src/room/participant/remote_participant.rs index b5a37c8..3206338 100644 --- a/livekit/src/room/participant/remote_participant.rs +++ b/livekit/src/room/participant/remote_participant.rs @@ -1,19 +1,11 @@ -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 super::{ + impl_participant_trait, ConnectionQuality, 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::{TrackKind, TrackTrait}; -use crate::room::TrackError; -use livekit_webrtc::media_stream::MediaStreamTrackHandle; +use crate::prelude::*; +use crate::proto; +use crate::publication::TrackPublicationInternalTrait; +use crate::track::TrackError; +use livekit_webrtc::prelude::*; use parking_lot::RwLockReadGuard; use std::collections::HashMap; use std::collections::HashSet; @@ -55,7 +47,7 @@ impl RemoteParticipant { /// 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>, kind: data_packet::Kind) { + pub(crate) fn on_data_received(&self, data: Arc>, kind: proto::data_packet::Kind) { self.shared .dispatcher .lock() @@ -172,7 +164,7 @@ impl RemoteParticipant { } impl ParticipantInternalTrait for RemoteParticipant { - fn update_info(self: &Arc, info: ParticipantInfo, emit_events: bool) { + fn update_info(self: &Arc, info: proto::ParticipantInfo, emit_events: bool) { self.shared.update_info(info.clone()); let mut valid_tracks = HashSet::::new(); diff --git a/livekit/src/room/publication/mod.rs b/livekit/src/room/publication/mod.rs index 0e8a60e..2ab3742 100644 --- a/livekit/src/room/publication/mod.rs +++ b/livekit/src/room/publication/mod.rs @@ -1,10 +1,5 @@ -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, TrackTrait}; +use crate::prelude::*; +use crate::proto; use livekit_utils::enum_dispatch; use livekit_utils::observer::Dispatcher; use parking_lot::Mutex; @@ -16,7 +11,7 @@ use super::track::{TrackDimension, TrackEvent}; pub(crate) trait TrackPublicationInternalTrait { fn update_track(&self, track: Option); - fn update_info(&self, info: TrackInfo); + fn update_info(&self, info: proto::TrackInfo); } pub trait TrackPublicationTrait { @@ -46,7 +41,7 @@ pub(super) struct TrackPublicationShared { impl TrackPublicationShared { pub fn new( - info: TrackInfo, + info: proto::TrackInfo, participant: ParticipantSid, track: Option, ) -> Arc { @@ -54,9 +49,11 @@ impl TrackPublicationShared { 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), + kind: AtomicU8::new( + TrackKind::from(proto::TrackType::from_i32(info.r#type).unwrap()) as u8, + ), source: AtomicU8::new(TrackSource::from( - ProtoTrackSource::from_i32(info.source).unwrap(), + proto::TrackSource::from_i32(info.source).unwrap(), ) as u8), simulcasted: AtomicBool::new(info.simulcast), dimension: Mutex::new(TrackDimension(info.width, info.height)), @@ -106,17 +103,17 @@ impl TrackPublicationShared { } } - pub fn update_info(&self, info: TrackInfo) { + pub fn update_info(&self, info: proto::TrackInfo) { *self.name.lock() = info.name; *self.sid.lock() = info.sid.into(); *self.dimension.lock() = TrackDimension(info.width, info.height); *self.mime_type.lock() = info.mime_type; self.kind.store( - TrackKind::from(TrackType::from_i32(info.r#type).unwrap()) as u8, + TrackKind::from(proto::TrackType::from_i32(info.r#type).unwrap()) as u8, Ordering::SeqCst, ); self.source.store( - TrackSource::from(ProtoTrackSource::from_i32(info.source).unwrap()) as u8, + TrackSource::from(proto::TrackSource::from_i32(info.source).unwrap()) as u8, Ordering::SeqCst, ); self.simulcasted.store(info.simulcast, Ordering::SeqCst); @@ -156,7 +153,7 @@ impl TrackPublicationInternalTrait for TrackPublication { enum_dispatch!( [Local, Remote] fnc!(update_track, &Self, [track: Option], ()); - fnc!(update_info, &Self, [info: TrackInfo], ()); + fnc!(update_info, &Self, [info: proto::TrackInfo], ()); ); } @@ -222,7 +219,7 @@ impl TrackPublicationInternalTrait for LocalTrackPublication { self.shared.update_track(track); } - fn update_info(&self, info: TrackInfo) { + fn update_info(&self, info: proto::TrackInfo) { self.shared.update_info(info); } } @@ -233,7 +230,11 @@ pub struct RemoteTrackPublication { } impl RemoteTrackPublication { - pub fn new(info: TrackInfo, participant: ParticipantSid, track: Option) -> Self { + pub fn new( + info: proto::TrackInfo, + participant: ParticipantSid, + track: Option, + ) -> Self { Self { shared: TrackPublicationShared::new(info, participant, track), } @@ -253,7 +254,7 @@ impl TrackPublicationInternalTrait for RemoteTrackPublication { self.shared.update_track(track); } - fn update_info(&self, info: TrackInfo) { + fn update_info(&self, info: proto::TrackInfo) { self.shared.update_info(info); } } diff --git a/livekit/src/room/room_session.rs b/livekit/src/room/room_session.rs index b8c811e..0b881b8 100644 --- a/livekit/src/room/room_session.rs +++ b/livekit/src/room/room_session.rs @@ -1,12 +1,9 @@ -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::participant::{ConnectionQuality, ParticipantInternalTrait}; +use crate::prelude::*; +use crate::proto; use crate::rtc_engine::{EngineEvent, EngineEvents, EngineResult, RTCEngine}; use crate::signal_client::SignalOptions; +use crate::{RoomEmitter, RoomError, RoomEvent, RoomResult, SimulateScenario}; use parking_lot::{Mutex, RwLock}; use std::collections::HashMap; use std::sync::atomic::{AtomicU8, Ordering}; @@ -365,7 +362,7 @@ impl SessionInner { 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 { + if pi.state == proto::participant_info::State::Disconnected as i32 { // Participant disconnected info!("Participant disconnected: {}", pi.sid); self.clone() @@ -394,7 +391,7 @@ impl SessionInner { /// Active speakers changed /// Update the participants & sort the active_speakers by audio_level #[instrument(level = Level::DEBUG)] - fn handle_speakers_changed(&self, speakers_info: Vec) { + fn handle_speakers_changed(&self, speakers_info: Vec) { let mut speakers = Vec::new(); for speaker in speakers_info { diff --git a/livekit/src/room/track/audio_track.rs b/livekit/src/room/track/audio_track.rs index cea9bfb..3ccd183 100644 --- a/livekit/src/room/track/audio_track.rs +++ b/livekit/src/room/track/audio_track.rs @@ -1,6 +1,4 @@ -use crate::room::track::local_audio_track::LocalAudioTrack; -use crate::room::track::remote_audio_track::RemoteAudioTrack; -use crate::room::track::TrackHandle; +use crate::prelude::*; use std::sync::Arc; #[derive(Clone)] diff --git a/livekit/src/room/track/local_audio_track.rs b/livekit/src/room/track/local_audio_track.rs index 057796a..860068c 100644 --- a/livekit/src/room/track/local_audio_track.rs +++ b/livekit/src/room/track/local_audio_track.rs @@ -1,4 +1,4 @@ -use crate::room::track::{impl_track_trait, TrackShared}; +use super::{impl_track_trait, TrackShared}; #[derive(Debug)] pub struct LocalAudioTrack { diff --git a/livekit/src/room/track/local_track.rs b/livekit/src/room/track/local_track.rs index 64cc127..0da4b5e 100644 --- a/livekit/src/room/track/local_track.rs +++ b/livekit/src/room/track/local_track.rs @@ -1,6 +1,4 @@ -use crate::room::track::local_audio_track::LocalAudioTrack; -use crate::room::track::local_video_track::LocalVideoTrack; -use crate::room::track::TrackHandle; +use crate::prelude::*; use std::sync::Arc; #[derive(Clone)] diff --git a/livekit/src/room/track/local_video_track.rs b/livekit/src/room/track/local_video_track.rs index c46b155..6fb5d52 100644 --- a/livekit/src/room/track/local_video_track.rs +++ b/livekit/src/room/track/local_video_track.rs @@ -1,4 +1,4 @@ -use crate::room::track::{impl_track_trait, TrackShared}; +use super::{impl_track_trait, TrackShared}; #[derive(Debug)] pub struct LocalVideoTrack { diff --git a/livekit/src/room/track/mod.rs b/livekit/src/room/track/mod.rs index b9eb468..89f7b64 100644 --- a/livekit/src/room/track/mod.rs +++ b/livekit/src/room/track/mod.rs @@ -1,15 +1,11 @@ -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; -use crate::room::track::remote_audio_track::RemoteAudioTrack; -use crate::room::track::remote_video_track::RemoteVideoTrack; +use crate::prelude::*; +use crate::proto; 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::{AtomicBool, AtomicU8, Ordering}; use std::sync::Arc; +use thiserror::Error; use tokio::sync::mpsc; pub mod audio_track; @@ -21,6 +17,21 @@ pub mod remote_track; pub mod remote_video_track; pub mod video_track; +pub use audio_track::*; +pub use local_audio_track::*; +pub use local_track::*; +pub use local_video_track::*; +pub use remote_audio_track::*; +pub use remote_track::*; +pub use remote_video_track::*; +pub use video_track::*; + +#[derive(Error, Debug, Clone)] +pub enum TrackError { + #[error("could not find published track with sid: {0}")] + TrackNotFound(String), +} + #[derive(Debug)] pub enum TrackKind { Unknown, @@ -38,11 +49,11 @@ impl From for TrackKind { } } -impl From for TrackKind { - fn from(r#type: TrackType) -> Self { +impl From for TrackKind { + fn from(r#type: proto::TrackType) -> Self { match r#type { - TrackType::Audio => Self::Audio, - TrackType::Video => Self::Video, + proto::TrackType::Audio => Self::Audio, + proto::TrackType::Video => Self::Video, _ => Self::Unknown, } } @@ -86,14 +97,14 @@ impl From for TrackSource { } } -impl From for TrackSource { - fn from(source: ProtoTrackSource) -> Self { +impl From for TrackSource { + fn from(source: proto::TrackSource) -> Self { match source { - ProtoTrackSource::Camera => Self::Camera, - ProtoTrackSource::Microphone => Self::Microphone, - ProtoTrackSource::ScreenShare => Self::Screenshare, - ProtoTrackSource::ScreenShareAudio => Self::ScreenshareAudio, - ProtoTrackSource::Unknown => Self::Unknown, + proto::TrackSource::Camera => Self::Camera, + proto::TrackSource::Microphone => Self::Microphone, + proto::TrackSource::ScreenShare => Self::Screenshare, + proto::TrackSource::ScreenShareAudio => Self::ScreenshareAudio, + proto::TrackSource::Unknown => Self::Unknown, } } } diff --git a/livekit/src/room/track/remote_audio_track.rs b/livekit/src/room/track/remote_audio_track.rs index 1ce12c6..d691a78 100644 --- a/livekit/src/room/track/remote_audio_track.rs +++ b/livekit/src/room/track/remote_audio_track.rs @@ -1,5 +1,5 @@ -use crate::room::track::{impl_track_trait, TrackShared}; -use livekit_webrtc::media_stream::{AudioTrack, MediaStreamTrackHandle}; +use super::{impl_track_trait, TrackShared}; +use crate::prelude::*; use std::sync::Arc; #[derive(Debug)] diff --git a/livekit/src/room/track/remote_track.rs b/livekit/src/room/track/remote_track.rs index 0ed2c09..d65ec15 100644 --- a/livekit/src/room/track/remote_track.rs +++ b/livekit/src/room/track/remote_track.rs @@ -1,13 +1,7 @@ -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, TrackEvent}; -use tokio::sync::mpsc; +use crate::prelude::*; use livekit_utils::enum_dispatch; - -use super::TrackTrait; +use std::sync::Arc; +use tokio::sync::mpsc; #[derive(Clone, Debug)] pub enum RemoteTrackHandle { diff --git a/livekit/src/room/track/remote_video_track.rs b/livekit/src/room/track/remote_video_track.rs index b3fc13f..428c403 100644 --- a/livekit/src/room/track/remote_video_track.rs +++ b/livekit/src/room/track/remote_video_track.rs @@ -1,8 +1,7 @@ -use livekit_webrtc::media_stream::{MediaStreamTrackHandle, VideoTrack}; +use super::{impl_track_trait, TrackShared}; +use crate::prelude::*; use std::sync::Arc; -use crate::room::track::{impl_track_trait, TrackShared}; - #[derive(Debug)] pub struct RemoteVideoTrack { shared: TrackShared, diff --git a/livekit/src/room/track/video_track.rs b/livekit/src/room/track/video_track.rs index 4294e5e..230582a 100644 --- a/livekit/src/room/track/video_track.rs +++ b/livekit/src/room/track/video_track.rs @@ -1,6 +1,4 @@ -use crate::room::track::local_video_track::LocalVideoTrack; -use crate::room::track::remote_video_track::RemoteVideoTrack; -use crate::room::track::TrackHandle; +use crate::prelude::*; use std::sync::Arc; #[derive(Clone)] diff --git a/livekit/src/rtc_engine/lk_runtime.rs b/livekit/src/rtc_engine/lk_runtime.rs index 5ff17c6..a8d3a4e 100644 --- a/livekit/src/rtc_engine/lk_runtime.rs +++ b/livekit/src/rtc_engine/lk_runtime.rs @@ -1,9 +1,7 @@ -use std::fmt::{Debug, Formatter}; - -use tracing::trace; - use livekit_webrtc::peer_connection_factory::PeerConnectionFactory; use livekit_webrtc::webrtc::RTCRuntime; +use std::fmt::{Debug, Formatter}; +use tracing::trace; /// SAFETY: The order of initialization and deletion is important for LKRuntime. /// See the C++ constructors & destructors of these fields diff --git a/livekit/src/rtc_engine/mod.rs b/livekit/src/rtc_engine/mod.rs index 7b3c055..ad2dd5b 100644 --- a/livekit/src/rtc_engine/mod.rs +++ b/livekit/src/rtc_engine/mod.rs @@ -1,30 +1,24 @@ +use crate::prelude::*; +use crate::proto; +use crate::rtc_engine::lk_runtime::LKRuntime; +use crate::rtc_engine::rtc_session::{RTCSession, SessionEvent, SessionEvents, SessionInfo}; +use crate::signal_client::{SignalError, SignalOptions}; use futures::future::BoxFuture; use futures::FutureExt; +use lazy_static::lazy_static; use livekit_webrtc::data_channel::DataSendError; use livekit_webrtc::jsep::SdpParseError; -use livekit_webrtc::media_stream::{MediaStream, MediaStreamTrackHandle}; -use livekit_webrtc::rtc_error::RTCError; -use livekit_webrtc::rtp_receiver::RtpReceiver; +use livekit_webrtc::prelude::*; use parking_lot::Mutex; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Weak}; use std::time::Duration; use thiserror::Error; use tokio::sync::RwLock as AsyncRwLock; +use tokio::sync::{mpsc, oneshot}; use tokio::task::JoinHandle; use tokio::time::{interval, Interval}; - -use lazy_static::lazy_static; -use tokio::sync::{mpsc, oneshot}; -use tracing::{error, info, trace, warn}; - -use crate::proto::{ - self as proto, data_packet, DataPacket, JoinResponse, ParticipantUpdate, SpeakerInfo, -}; -use crate::rtc_engine::lk_runtime::LKRuntime; -use crate::signal_client::{SignalError, SignalOptions}; - -use self::rtc_session::{RTCSession, SessionEvent, SessionEvents, SessionInfo}; +use tracing::{error, info, warn}; mod lk_runtime; mod pc_transport; @@ -80,10 +74,10 @@ pub enum EngineEvent { Data { participant_sid: String, payload: Vec, - kind: data_packet::Kind, + kind: proto::data_packet::Kind, }, SpeakersChanged { - speakers: Vec, + speakers: Vec, }, ConnectionQuality { updates: Vec, @@ -133,21 +127,20 @@ pub struct RTCEngine { impl RTCEngine { pub fn new() -> (Self, EngineEvents) { - let mut lk_runtime = None; - { + let lk_runtime = { let mut lk_runtime_ref = LK_RUNTIME.lock(); - lk_runtime = lk_runtime_ref.upgrade(); - - if lk_runtime.is_none() { + if let Some(lk_runtime) = lk_runtime_ref.upgrade() { + lk_runtime + } else { let new_runtime = Arc::new(LKRuntime::default()); *lk_runtime_ref = Arc::downgrade(&new_runtime); - lk_runtime = Some(new_runtime); + new_runtime } - } + }; let (engine_emitter, engine_events) = mpsc::channel(8); let inner = Arc::new(EngineInner { - lk_runtime: lk_runtime.unwrap(), + lk_runtime, session_info: Default::default(), running_handle: Default::default(), opened: Default::default(), @@ -178,8 +171,8 @@ impl RTCEngine { #[tracing::instrument(skip(data))] pub async fn publish_data( &self, - data: &DataPacket, - kind: data_packet::Kind, + data: &proto::DataPacket, + kind: proto::data_packet::Kind, ) -> EngineResult<()> { self.inner.wait_reconnection().await?; self.inner @@ -207,7 +200,7 @@ impl RTCEngine { Ok(()) } - pub fn join_response(&self) -> Option { + pub fn join_response(&self) -> Option { if let Some(info) = self.inner.session_info.lock().as_ref() { Some(info.join_response.clone()) } else { diff --git a/livekit/src/rtc_engine/pc_transport.rs b/livekit/src/rtc_engine/pc_transport.rs index 34b2fa2..9185f2d 100644 --- a/livekit/src/rtc_engine/pc_transport.rs +++ b/livekit/src/rtc_engine/pc_transport.rs @@ -1,18 +1,11 @@ +use crate::proto; +use livekit_webrtc::prelude::*; use std::fmt::{Debug, Formatter}; use std::future::Future; use std::pin::Pin; use std::time::Duration; - use tracing::{event, Level}; -use livekit_webrtc::jsep::{IceCandidate, SessionDescription}; -use livekit_webrtc::peer_connection::{ - IceConnectionState, PeerConnection, RTCOfferAnswerOptions, SignalingState, -}; -use livekit_webrtc::rtc_error::RTCError; - -use crate::proto::SignalTarget; - const NEGOTIATION_FREQUENCY: Duration = Duration::from_millis(150); pub type OnOfferHandler = Box< @@ -22,7 +15,7 @@ pub type OnOfferHandler = Box< >; pub struct PCTransport { - signal_target: SignalTarget, + signal_target: proto::SignalTarget, peer_connection: PeerConnection, pending_candidates: Vec, on_offer_handler: Option, @@ -37,7 +30,7 @@ impl Debug for PCTransport { } impl PCTransport { - pub fn new(peer_connection: PeerConnection, signal_target: SignalTarget) -> Self { + pub fn new(peer_connection: PeerConnection, signal_target: proto::SignalTarget) -> Self { Self { signal_target, peer_connection, @@ -58,7 +51,7 @@ impl PCTransport { &mut self.peer_connection } - pub fn signal_target(&self) -> SignalTarget { + pub fn signal_target(&self) -> proto::SignalTarget { self.signal_target.clone() } diff --git a/livekit/src/rtc_engine/rtc_events.rs b/livekit/src/rtc_engine/rtc_events.rs index dca1af6..e3a8c56 100644 --- a/livekit/src/rtc_engine/rtc_events.rs +++ b/livekit/src/rtc_engine/rtc_events.rs @@ -1,19 +1,15 @@ -use livekit_webrtc::data_channel::{DataChannel, OnMessageHandler}; -use livekit_webrtc::jsep::{IceCandidate, SessionDescription}; -use livekit_webrtc::media_stream::MediaStream; +use super::pc_transport::PCTransport; +use crate::proto; +use crate::rtc_engine::pc_transport::OnOfferHandler; +use livekit_webrtc::data_channel::OnMessageHandler; use livekit_webrtc::peer_connection::{ OnAddTrackHandler, OnConnectionChangeHandler, OnDataChannelHandler, OnIceCandidateErrorHandler, OnIceCandidateHandler, PeerConnectionState, }; -use livekit_webrtc::rtp_receiver::RtpReceiver; +use livekit_webrtc::prelude::*; use tokio::sync::mpsc; use tracing::error; -use crate::proto::SignalTarget; -use crate::rtc_engine::pc_transport::OnOfferHandler; - -use super::pc_transport::PCTransport; - pub type RTCEmitter = mpsc::UnboundedSender; pub type RTCEvents = mpsc::UnboundedReceiver; @@ -21,25 +17,25 @@ pub type RTCEvents = mpsc::UnboundedReceiver; pub enum RTCEvent { IceCandidate { ice_candidate: IceCandidate, - target: SignalTarget, + target: proto::SignalTarget, }, ConnectionChange { state: PeerConnectionState, - target: SignalTarget, + target: proto::SignalTarget, }, DataChannel { data_channel: DataChannel, - target: SignalTarget, + target: proto::SignalTarget, }, // TODO (theomonnom): Move Offer to PCTransport Offer { offer: SessionDescription, - target: SignalTarget, + target: proto::SignalTarget, }, AddTrack { rtp_receiver: RtpReceiver, streams: Vec, - target: SignalTarget, + target: proto::SignalTarget, }, Data { data: Vec, @@ -50,13 +46,16 @@ pub enum RTCEvent { /// Handlers used to forward events to a channel /// Every callback here is called on the signaling thread -fn on_connection_change(target: SignalTarget, emitter: RTCEmitter) -> OnConnectionChangeHandler { +fn on_connection_change( + target: proto::SignalTarget, + emitter: RTCEmitter, +) -> OnConnectionChangeHandler { Box::new(move |state| { let _ = emitter.send(RTCEvent::ConnectionChange { state, target }); }) } -fn on_ice_candidate(target: SignalTarget, emitter: RTCEmitter) -> OnIceCandidateHandler { +fn on_ice_candidate(target: proto::SignalTarget, emitter: RTCEmitter) -> OnIceCandidateHandler { Box::new(move |ice_candidate| { let _ = emitter.send(RTCEvent::IceCandidate { ice_candidate, @@ -65,7 +64,7 @@ fn on_ice_candidate(target: SignalTarget, emitter: RTCEmitter) -> OnIceCandidate }) } -fn on_offer(target: SignalTarget, emitter: RTCEmitter) -> OnOfferHandler { +fn on_offer(target: proto::SignalTarget, emitter: RTCEmitter) -> OnOfferHandler { Box::new(move |offer| { let _ = emitter.send(RTCEvent::Offer { offer, target }); @@ -73,7 +72,7 @@ fn on_offer(target: SignalTarget, emitter: RTCEmitter) -> OnOfferHandler { }) } -fn on_data_channel(target: SignalTarget, emitter: RTCEmitter) -> OnDataChannelHandler { +fn on_data_channel(target: proto::SignalTarget, emitter: RTCEmitter) -> OnDataChannelHandler { Box::new(move |mut data_channel| { data_channel.on_message(on_message(emitter.clone())); @@ -84,7 +83,7 @@ fn on_data_channel(target: SignalTarget, emitter: RTCEmitter) -> OnDataChannelHa }) } -fn on_add_track(target: SignalTarget, emitter: RTCEmitter) -> OnAddTrackHandler { +fn on_add_track(target: proto::SignalTarget, emitter: RTCEmitter) -> OnAddTrackHandler { Box::new(move |rtp_receiver, streams| { let _ = emitter.send(RTCEvent::AddTrack { rtp_receiver, @@ -95,7 +94,7 @@ fn on_add_track(target: SignalTarget, emitter: RTCEmitter) -> OnAddTrackHandler } fn on_ice_candidate_error( - target: SignalTarget, + target: proto::SignalTarget, _emitter: RTCEmitter, ) -> OnIceCandidateErrorHandler { Box::new(move |address, port, url, error_code, error_text| { diff --git a/livekit/src/rtc_engine/rtc_session.rs b/livekit/src/rtc_engine/rtc_session.rs index 8bafab2..b64ea13 100644 --- a/livekit/src/rtc_engine/rtc_session.rs +++ b/livekit/src/rtc_engine/rtc_session.rs @@ -1,38 +1,21 @@ -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 super::{rtc_events, EngineError, EngineResult, SimulateScenario}; 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}; +use crate::{proto, signal_client}; +use livekit_webrtc::prelude::*; +use parking_lot::Mutex; +use prost::Message; +use serde::{Deserialize, Serialize}; +use std::convert::TryInto; +use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::{mpsc, watch, Mutex as AsyncMutex}; +use tokio::task::JoinHandle; +use tokio::time::sleep; +use tracing::{debug, error, trace, warn}; pub const MAX_ICE_CONNECT_TIMEOUT: Duration = Duration::from_secs(15); pub const LOSSY_DC_LABEL: &str = "_lossy"; @@ -65,7 +48,7 @@ pub enum SessionEvent { // TODO(theomonnom): Move entirely the reconnection logic on mod.rs Close { source: String, - reason: DisconnectReason, + reason: proto::DisconnectReason, can_reconnect: bool, full_reconnect: bool, retry_now: bool, @@ -110,7 +93,7 @@ pub struct SessionInfo { pub url: String, pub token: String, pub options: SignalOptions, - pub join_response: JoinResponse, + pub join_response: proto::JoinResponse, } /// Fields shared with rtc_task and signal_task @@ -171,14 +154,14 @@ impl RTCSession { lk_runtime .pc_factory .create_peer_connection(rtc_config.clone())?, - SignalTarget::Publisher, + proto::SignalTarget::Publisher, ); let mut subscriber_pc = PCTransport::new( lk_runtime .pc_factory .create_peer_connection(rtc_config.clone())?, - SignalTarget::Subscriber, + proto::SignalTarget::Subscriber, ); let mut lossy_dc = publisher_pc.peer_connection().create_data_channel( @@ -257,8 +240,8 @@ impl RTCSession { pub async fn publish_data( &self, - data: &DataPacket, - kind: data_packet::Kind, + data: &proto::DataPacket, + kind: proto::data_packet::Kind, ) -> Result<(), EngineError> { self.inner.publish_data(data, kind).await } @@ -301,7 +284,7 @@ impl RTCSession { &self.inner.signal_client } - pub fn data_channel(&self, kind: data_packet::Kind) -> &DataChannel { + pub fn data_channel(&self, kind: proto::data_packet::Kind) -> &DataChannel { &self.inner.data_channel(kind) } } @@ -347,7 +330,13 @@ impl SessionInner { } } SignalEvent::Close => { - self.on_session_disconnected("SignalClient closed", DisconnectReason::UnknownReason, true, false, false); + self.on_session_disconnected( + "SignalClient closed", + proto::DisconnectReason::UnknownReason, + true, + false, + false + ); } } } else { @@ -362,9 +351,9 @@ impl SessionInner { } } - async fn on_signal_event(&self, event: signal_response::Message) -> EngineResult<()> { + async fn on_signal_event(&self, event: proto::signal_response::Message) -> EngineResult<()> { match event { - signal_response::Message::Answer(answer) => { + proto::signal_response::Message::Answer(answer) => { trace!("received publisher answer: {:?}", answer); let answer = SessionDescription::from(answer.r#type.parse().unwrap(), &answer.sdp)?; self.publisher_pc @@ -373,7 +362,7 @@ impl SessionInner { .set_remote_description(answer) .await?; } - signal_response::Message::Offer(offer) => { + proto::signal_response::Message::Offer(offer) => { trace!("received subscriber offer: {:?}", offer); let offer = SessionDescription::from(offer.r#type.parse().unwrap(), &offer.sdp)?; let answer = self @@ -384,14 +373,16 @@ impl SessionInner { .await?; self.signal_client - .send(signal_request::Message::Answer(proto::SessionDescription { - r#type: "answer".to_string(), - sdp: answer.to_string(), - })) + .send(proto::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(); + proto::signal_response::Message::Trickle(trickle) => { + let target = proto::SignalTarget::from_i32(trickle.target).unwrap(); let ice_candidate = { let json = serde_json::from_str::(&trickle.candidate_init)?; IceCandidate::from(&json.sdpMid, json.sdpMLineIndex, &json.candidate)? @@ -399,7 +390,7 @@ impl SessionInner { trace!("received ice_candidate {:?} {:?}", target, ice_candidate); - if target == SignalTarget::Publisher { + if target == proto::SignalTarget::Publisher { self.publisher_pc .lock() .await @@ -413,7 +404,7 @@ impl SessionInner { .await?; } } - signal_response::Message::Leave(leave) => { + proto::signal_response::Message::Leave(leave) => { self.on_session_disconnected( "received leave", leave.reason(), @@ -422,17 +413,17 @@ impl SessionInner { true, ); } - signal_response::Message::Update(update) => { + proto::signal_response::Message::Update(update) => { let _ = self.emitter.send(SessionEvent::ParticipantUpdate { updates: update.participants, }); } - signal_response::Message::SpeakersChanged(speaker) => { + proto::signal_response::Message::SpeakersChanged(speaker) => { let _ = self.emitter.send(SessionEvent::SpeakersChanged { speakers: speaker.speakers, }); } - signal_response::Message::ConnectionQuality(quality) => { + proto::signal_response::Message::ConnectionQuality(quality) => { let _ = self.emitter.send(SessionEvent::ConnectionQuality { updates: quality.updates, }); @@ -450,20 +441,22 @@ impl SessionInner { 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, - })) + .send(proto::signal_request::Message::Trickle( + proto::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; + && target == proto::SignalTarget::Subscriber; if is_primary && state == PeerConnectionState::Connected { let old_state = self @@ -478,7 +471,7 @@ impl SessionInner { self.on_session_disconnected( "pc_state failed", - DisconnectReason::UnknownReason, + proto::DisconnectReason::UnknownReason, true, false, false, @@ -494,10 +487,12 @@ impl SessionInner { 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(), - })) + .send(proto::signal_request::Message::Offer( + proto::SessionDescription { + r#type: "offer".to_string(), + sdp: offer.to_string(), + }, + )) .await; } RTCEvent::AddTrack { @@ -522,16 +517,16 @@ impl SessionInner { ))?; } - let data = DataPacket::decode(&*data)?; + let data = proto::DataPacket::decode(&*data)?; match data.value.unwrap() { - Value::User(user) => { + proto::data_packet::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(), + kind: proto::data_packet::Kind::from_i32(data.kind).unwrap(), }); } - Value::Speaker(_) => {} + proto::data_packet::Value::Speaker(_) => {} } } } @@ -544,7 +539,7 @@ impl SessionInner { fn on_session_disconnected( &self, source: &str, - reason: DisconnectReason, + reason: proto::DisconnectReason, can_reconnect: bool, retry_now: bool, full_reconnect: bool, @@ -574,52 +569,64 @@ impl SessionInner { } SimulateScenario::Speaker => { self.signal_client - .send(signal_request::Message::Simulate(proto::SimulateScenario { - scenario: Some(proto::simulate_scenario::Scenario::SpeakerUpdate(3)), - })) + .send(proto::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)), - })) + .send(proto::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)), - })) + .send(proto::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)), - })) + .send(proto::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, + .send(proto::signal_request::Message::Simulate( + proto::SimulateScenario { + scenario: Some( + proto::simulate_scenario::Scenario::SwitchCandidateProtocol( + proto::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, + .send(proto::signal_request::Message::Simulate( + proto::SimulateScenario { + scenario: Some( + proto::simulate_scenario::Scenario::SwitchCandidateProtocol( + proto::CandidateProtocol::Tls as i32, + ), ), - ), - })) + }, + )) .await; } } @@ -628,8 +635,8 @@ impl SessionInner { #[tracing::instrument(skip(data))] async fn publish_data( &self, - data: &DataPacket, - kind: data_packet::Kind, + data: &proto::DataPacket, + kind: proto::data_packet::Kind, ) -> Result<(), EngineError> { self.ensure_publisher_connected(kind).await?; self.data_channel(kind) @@ -705,7 +712,7 @@ impl SessionInner { /// 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<()> { + async fn ensure_publisher_connected(&self, kind: proto::data_packet::Kind) -> EngineResult<()> { if !self.info.join_response.subscriber_primary { return Ok(()); } @@ -750,8 +757,8 @@ impl SessionInner { } } - fn data_channel(&self, kind: data_packet::Kind) -> &DataChannel { - if kind == data_packet::Kind::Reliable { + fn data_channel(&self, kind: proto::data_packet::Kind) -> &DataChannel { + if kind == proto::data_packet::Kind::Reliable { &self.reliable_dc } else { &self.lossy_dc diff --git a/livekit/src/signal_client/mod.rs b/livekit/src/signal_client/mod.rs index f4fac3d..e8776f8 100644 --- a/livekit/src/signal_client/mod.rs +++ b/livekit/src/signal_client/mod.rs @@ -1,13 +1,9 @@ -use std::fmt::Debug; - -use std::time::Duration; - -use crate::proto::{signal_request, signal_response, JoinResponse}; +use crate::proto; use crate::signal_client::signal_stream::SignalStream; -use livekit_webrtc::peer_connection_factory::{ - ContinualGatheringPolicy, ICEServer, IceTransportsType, RTCConfiguration, -}; +use livekit_webrtc::prelude::*; use parking_lot::RwLock; +use std::fmt::Debug; +use std::time::Duration; use thiserror::Error; use tokio::sync::mpsc; use tokio_tungstenite::tungstenite::Error as WsError; @@ -37,7 +33,7 @@ pub enum SignalError { #[derive(Debug)] pub enum SignalEvent { Open, - Signal(signal_response::Message), + Signal(proto::signal_response::Message), Close, } @@ -98,7 +94,7 @@ impl SignalClient { } #[instrument(level = Level::DEBUG)] - pub async fn send(&self, signal: signal_request::Message) { + pub async fn send(&self, signal: proto::signal_request::Message) { if let Some(stream) = self.stream.read().as_ref() { if stream.send(signal).await.is_ok() { return; @@ -118,8 +114,8 @@ impl SignalClient { } } -impl From for RTCConfiguration { - fn from(join_response: JoinResponse) -> Self { +impl From for RTCConfiguration { + fn from(join_response: proto::JoinResponse) -> Self { Self { ice_servers: { let mut servers = vec![]; diff --git a/livekit/src/signal_client/signal_stream.rs b/livekit/src/signal_client/signal_stream.rs index 1692397..76d6681 100644 --- a/livekit/src/signal_client/signal_stream.rs +++ b/livekit/src/signal_client/signal_stream.rs @@ -1,19 +1,17 @@ +use crate::proto; +use crate::signal_client::{SignalEmitter, SignalEvent, SignalOptions, SignalResult}; use futures_util::stream::{SplitSink, SplitStream}; use futures_util::{SinkExt, StreamExt}; 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; use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream}; use tracing::{event, Level}; -use crate::proto::{signal_request, SignalRequest, SignalResponse}; -use crate::signal_client::{SignalEmitter, SignalEvent, SignalOptions, SignalResult}; - pub const PROTOCOL_VERSION: u32 = 8; type WebSocket = WebSocketStream>; @@ -21,7 +19,7 @@ type WebSocket = WebSocketStream>; #[derive(Debug)] enum InternalMessage { Signal { - signal: signal_request::Message, + signal: proto::signal_request::Message, response_chn: oneshot::Sender>, }, Pong { @@ -107,7 +105,7 @@ impl SignalStream { /// Send a SignalRequest to the websocket /// It also waits for the message to be sent - pub async fn send(&self, signal: signal_request::Message) -> SignalResult<()> { + pub async fn send(&self, signal: proto::signal_request::Message) -> SignalResult<()> { let (send, recv) = oneshot::channel(); let msg = InternalMessage::Signal { signal, @@ -133,7 +131,7 @@ impl SignalStream { event!(Level::TRACE, "sending SignalRequest: {:?}", signal); let data = Message::Binary( - SignalRequest { + proto::SignalRequest { message: Some(signal), } .encode_to_vec(), @@ -178,7 +176,7 @@ impl SignalStream { while let Some(msg) = ws_reader.next().await { match msg { Ok(Message::Binary(data)) => { - let res = SignalResponse::decode(data.as_slice()) + let res = proto::SignalResponse::decode(data.as_slice()) .expect("failed to decode SignalResponse"); let msg = res.message.unwrap();