refactor(rtc-manager): remove async

Author: lambiengcode

crates/rtc-manager/src/entities/publisher.rs

@@ -1,17 +1,20 @@
-use std::sync::{
-    atomic::{AtomicU8, Ordering},
-    Arc,
-};
 use std::time::{Duration, Instant};
+use std::{
+    sync::{
+        atomic::{AtomicU8, Ordering},
+        Arc,
+    },
+    thread,
+};
 
 use dashmap::DashMap;
 use parking_lot::RwLock;
+use std::sync::mpsc::{self, Receiver};
 use str0m::{
     change::SdpOffer,
-    media::{Direction, KeyframeRequestKind, MediaData, MediaKind, Mid},
+    media::{Direction, KeyframeRequestKind, MediaData, MediaKind, Mid, Rid},
     Event, IceConnectionState, Rtc,
 };
-use tokio::sync::mpsc::{self, UnboundedReceiver};
 use tokio_util::sync::CancellationToken;
 
 use crate::{
@@ -67,10 +70,8 @@ impl Publisher {
         // Create str0m RTC instance
         let rtc = Rtc::builder().build();
 
-        // let (track_event_sender, track_event_receiver) = mpsc::unbounded_channel();
-
         // Create event channel for runtime -> publisher
-        let (event_tx, event_rx) = mpsc::unbounded_channel();
+        let (event_tx, event_rx) = mpsc::sync_channel(1);
 
         let publisher = Arc::new(Self {
             participant_id,
@@ -116,10 +117,8 @@ impl Publisher {
 
         // Start the RTC event loop (events from runtime)
         let publisher_clone = Arc::clone(&publisher);
-        tokio::spawn(async move {
-            publisher_clone
-                .run_event_loop(event_rx, joined_handler)
-                .await;
+        thread::spawn(move || {
+            publisher_clone.run_event_loop(event_rx, joined_handler);
         });
 
         Ok(publisher)
@@ -211,19 +210,19 @@ impl Publisher {
         self.tracks.clear();
     }
 
-    async fn run_event_loop<J>(self: Arc<Self>, mut rx: UnboundedReceiver<Event>, joined_handler: J)
+    fn run_event_loop<J>(self: Arc<Self>, rx: Receiver<Event>, joined_handler: J)
     where
         J: JoinedHandler + Clone,
     {
-        while let Some(event) = rx.recv().await {
-            self.handle_rtc_event(event, joined_handler.clone()).await;
+        while let Ok(event) = rx.recv() {
+            self.handle_rtc_event(event, joined_handler.clone());
             if self.cancel_token.is_cancelled() {
                 break;
             }
         }
     }
 
-    async fn handle_rtc_event<J>(&self, event: Event, joined_handler: J)
+    fn handle_rtc_event<J>(&self, event: Event, joined_handler: J)
     where
         J: JoinedHandler,
     {
@@ -250,7 +249,7 @@ impl Publisher {
                 self.request_keyframe_throttled(&data);
 
                 // Forward media data to subscribers
-                self.forward_media_to_subscribers(data).await;
+                self.forward_media_to_subscribers(data);
             }
             Event::RtpPacket(packet) => {
                 // Forward RTP packet to subscribers
@@ -262,7 +261,11 @@ impl Publisher {
         }
     }
 
-    async fn forward_media_to_subscribers(&self, media_data: MediaData) {
+    fn forward_media_to_subscribers(&self, media_data: MediaData) {
+        if media_data.rid.is_some() && media_data.rid != Some(Rid::from("h")) {
+            return;
+        }
+
         // For each subscriber, write the incoming media into its corresponding send-only mid
         for entry in self.subscribers.iter() {
             let subscriber = entry.value();

crates/rtc-manager/src/entities/room.rs

@@ -2,7 +2,6 @@ use std::sync::Arc;
 
 use dashmap::DashMap;
 use str0m::{net::Protocol, Candidate};
-use tracing::info;
 
 use crate::{
     entities::{publisher::Publisher, subscriber::Subscriber},
@@ -43,8 +42,6 @@ impl Room {
     {
         let participant_id = params.participant_id.clone();
 
-        info!("participant joining {}", participant_id.clone());
-
         // Create publisher
         let publisher = Publisher::new(
             participant_id.clone(),
@@ -58,25 +55,16 @@ impl Room {
             params.joined_handler,
         )?;
 
-        info!("publisher created");
-
         // Add publisher to room
         self.publishers
             .insert(participant_id.clone(), publisher.clone());
 
-        info!("publisher added to room");
-
-        info!("connection type: {:?}", params.connection_type);
-
         // Handle SDP based on connection type
         match params.connection_type {
             ConnectionType::SFU => {
-                info!("handle_offer: {}", params.sdp.len());
                 // For SFU mode, handle the offer and create an answer
                 let answer_sdp = publisher.handle_offer(params.sdp)?;
 
-                info!("answer_sdp: {}", answer_sdp.len());
-
                 Ok(Some(JoinRoomResponse {
                     sdp: answer_sdp,
                     is_recording: false,
@@ -160,13 +148,9 @@ impl Room {
             .get(participant_id)
             .ok_or(RtcError::PublisherNotFound)?;
 
-        info!("add_publisher_candidate: {}", participant_id);
-
         // Convert IceCandidate to str0m Candidate
         let str0m_candidate = self.convert_ice_candidate_to_str0m(candidate)?;
 
-        info!("str0m_candidate: {:?}", str0m_candidate);
-
         // Add remote (peer) candidate to publisher's RTC instance
         {
             let mut rtc = publisher.rtc.write();

crates/rtc-manager/src/entities/subscriber.rs

@@ -1,16 +1,19 @@
-use std::sync::{
-    atomic::{AtomicU8, Ordering},
-    Arc,
+use std::{
+    sync::{
+        atomic::{AtomicU8, Ordering},
+        Arc,
+    },
+    thread,
 };
 
 use dashmap::DashMap;
 use parking_lot::RwLock;
+use std::sync::mpsc::{self, Receiver};
 use str0m::{
     change::{SdpAnswer, SdpPendingOffer},
     media::{Direction, MediaKind, Mid},
     Event, IceConnectionState, Rtc,
 };
-use tokio::sync::mpsc::{self, UnboundedReceiver};
 use tokio_util::sync::CancellationToken;
 
 use crate::{
@@ -66,7 +69,7 @@ impl Subscriber {
         let rtc = Rtc::builder().build();
 
         // Create event channel for runtime -> subscriber
-        let (event_tx, event_rx) = mpsc::unbounded_channel();
+        let (event_tx, event_rx) = mpsc::sync_channel(1);
 
         let subscriber = Arc::new(Self {
             participant_id,
@@ -93,10 +96,9 @@ impl Subscriber {
             .ok();
 
         // Announce host candidate to the client via callback
-        if let (Some(cb), Some(host)) = (
-            Some(ice_handler),
-            RtcUdpRuntime::global().host_candidate(),
-        ) {
+        if let (Some(cb), Some(host)) =
+            (Some(ice_handler), RtcUdpRuntime::global().host_candidate())
+        {
             let ice = crate::utils::ice_utils::IceUtils::convert_from_str0m_candidate(
                 &host,
                 Some("0".to_string()),
@@ -107,8 +109,8 @@ impl Subscriber {
 
         // Start the RTC event loop
         let subscriber_clone = Arc::clone(&subscriber);
-        tokio::spawn(async move {
-            subscriber_clone.run_event_loop(event_rx).await;
+        thread::spawn(move || {
+            subscriber_clone.run_event_loop(event_rx);
         });
 
         Ok(subscriber)
@@ -220,16 +222,16 @@ impl Subscriber {
         self.tracks.clear();
     }
 
-    async fn run_event_loop(self: Arc<Self>, mut rx: UnboundedReceiver<Event>) {
-        while let Some(event) = rx.recv().await {
-            self.handle_rtc_event(event).await;
+    fn run_event_loop(self: Arc<Self>, rx: Receiver<Event>) {
+        while let Ok(event) = rx.recv() {
+            self.handle_rtc_event(event);
             if self.cancel_token.is_cancelled() {
                 break;
             }
         }
     }
 
-    async fn handle_rtc_event(&self, event: Event) {
+    fn handle_rtc_event(&self, event: Event) {
         match event {
             Event::Connected => {
                 tracing::info!(
@@ -268,7 +270,7 @@ impl Subscriber {
         }
     }
 
-    pub async fn receive_media_data(&self, _data: &[u8]) -> Result<(), RtcError> {
+    pub fn receive_media_data(&self, _data: &[u8]) -> Result<(), RtcError> {
         // TODO: Process incoming media data from publisher
         // This is where the subscriber receives data controlled by the publisher
         Ok(())

crates/rtc-manager/src/utils/udp_runtime.rs

@@ -6,8 +6,8 @@ use std::time::{Duration, Instant};
 use dashmap::DashMap;
 use once_cell::sync::OnceCell;
 use parking_lot::RwLock;
+use std::sync::mpsc::{self};
 use systemstat::{Platform, System};
-use tokio::sync::mpsc;
 use tokio_util::sync::CancellationToken;
 
 use str0m::net::Receive;
@@ -20,7 +20,7 @@ use crate::models::rtc_dto::RtcManagerConfig;
 pub struct RtcRegistration {
     pub id: String,
     pub rtc: Arc<RwLock<Rtc>>,
-    pub event_tx: mpsc::UnboundedSender<Event>,
+    pub event_tx: mpsc::SyncSender<Event>,
 }
 
 pub struct RtcUdpRuntime {
@@ -89,7 +89,7 @@ impl RtcUdpRuntime {
         &self,
         id: String,
         rtc: Arc<RwLock<Rtc>>,
-        event_tx: mpsc::UnboundedSender<Event>,
+        event_tx: mpsc::SyncSender<Event>,
     ) -> Result<(), RtcError> {
         // Add a host candidate that matches the runtime socket
         let addr = self

sfu/src/application/sfu_grpc_service.rs

@@ -57,6 +57,7 @@ impl SfuService for SfuGrpcService {
         let ice_handler = GrpcPublisherIceHandler {
             dispatcher: Arc::clone(&self.dispatcher_grpc_client),
             client_id: req.client_id.clone(),
+            tokio_handle: tokio::runtime::Handle::current(),
         };
 
         let joined_handler = GrpcJoinedHandler {
@@ -65,6 +66,7 @@ impl SfuService for SfuGrpcService {
             room_id: req.room_id.clone(),
             client_id: req.client_id.clone(),
             node_id: self.node_id.clone(),
+            tokio_handle: tokio::runtime::Handle::current(),
         };
 
         let rtc_manager = self.rtc_manager.clone();
@@ -115,12 +117,14 @@ impl SfuService for SfuGrpcService {
             dispatcher: Arc::clone(&self.dispatcher_grpc_client),
             client_id: req.client_id.clone(),
             target_id: req.target_id.clone(),
+            tokio_handle: tokio::runtime::Handle::current(),
         };
 
         let renegotiation_handler = GrpcRenegotiationHandler {
             dispatcher: Arc::clone(&self.dispatcher_grpc_client),
             client_id: req.client_id.clone(),
             target_id: req.target_id.clone(),
+            tokio_handle: tokio::runtime::Handle::current(),
         };
 
         let rtc_manager = self.rtc_manager.clone();
@@ -434,14 +438,15 @@ impl SfuService for SfuGrpcService {
 pub struct GrpcPublisherIceHandler {
     pub dispatcher: Arc<Mutex<DispatcherGrpcClient>>,
     pub client_id: String,
+    pub tokio_handle: tokio::runtime::Handle,
 }
 
 impl IceCandidateHandler for GrpcPublisherIceHandler {
     fn handle_candidate(&self, candidate: IceCandidate) {
         let dispatcher = Arc::clone(&self.dispatcher);
         let client_id = self.client_id.clone();
 
-        tokio::spawn(async move {
+        self.tokio_handle.spawn(async move {
             let dispatcher = dispatcher.lock().await;
 
             let _ = dispatcher
@@ -463,14 +468,16 @@ pub struct GrpcSubscriberIceHandler {
     pub dispatcher: Arc<Mutex<DispatcherGrpcClient>>,
     pub client_id: String,
     pub target_id: String,
+    pub tokio_handle: tokio::runtime::Handle,
 }
 
 impl IceCandidateHandler for GrpcSubscriberIceHandler {
     fn handle_candidate(&self, candidate: IceCandidate) {
         let dispatcher = Arc::clone(&self.dispatcher);
         let client_id = self.client_id.clone();
         let target_id = self.target_id.clone();
-        tokio::spawn(async move {
+
+        self.tokio_handle.spawn(async move {
             let dispatcher = dispatcher.lock().await;
 
             let _ = dispatcher
@@ -495,6 +502,7 @@ pub struct GrpcJoinedHandler {
     pub room_id: String,
     pub client_id: String,
     pub node_id: String,
+    pub tokio_handle: tokio::runtime::Handle,
 }
 
 impl JoinedHandler for GrpcJoinedHandler {
@@ -505,9 +513,8 @@ impl JoinedHandler for GrpcJoinedHandler {
         let client_id = self.client_id.clone();
         let node_id = self.node_id.clone();
 
-        tokio::spawn(async move {
+        self.tokio_handle.spawn(async move {
             let dispatcher = dispatcher.lock().await;
-
             let _ = dispatcher
                 .new_user_joined(NewUserJoinedRequest {
                     participant_id,
@@ -526,6 +533,7 @@ pub struct GrpcRenegotiationHandler {
     pub dispatcher: Arc<Mutex<DispatcherGrpcClient>>,
     pub client_id: String,
     pub target_id: String,
+    pub tokio_handle: tokio::runtime::Handle,
 }
 
 impl RenegotiationHandler for GrpcRenegotiationHandler {
@@ -534,7 +542,7 @@ impl RenegotiationHandler for GrpcRenegotiationHandler {
         let client_id = self.client_id.clone();
         let target_id = self.target_id.clone();
 
-        tokio::spawn(async move {
+        self.tokio_handle.spawn(async move {
             let dispatcher = dispatcher.lock().await;
             let _ = dispatcher
                 .subscriber_renegotiate(SubscriberRenegotiateRequest {