refactor(callbacks): .

Author: lambiengcode

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

@@ -17,8 +17,8 @@ use tokio_util::sync::CancellationToken;
 use crate::{
     errors::RtcError,
     models::{
+        callbacks::{IceCandidateHandler, JoinedHandler},
         connection_type::ConnectionType,
-        params::{IceCandidateCallback, JoinedCallback},
         streaming_protocol::StreamingProtocol,
     },
 };
@@ -40,26 +40,30 @@ pub struct Publisher {
     pub streaming_protocol: StreamingProtocol,
     // pub track_event_sender: Option<mpsc::UnboundedSender<TrackSubscribedMessage>>,
     // pub track_event_receiver: Option<mpsc::UnboundedReceiver<TrackSubscribedMessage>>,
-    pub ice_candidate_callback: Option<IceCandidateCallback>,
-    pub joined_callback: Option<JoinedCallback>,
+    // pub ice_handler: Option<I>,
+    // pub joined_handler: Option<J>,
     // Map incoming mid -> kind, populated on MediaAdded
     pub incoming_mid_kind: Arc<DashMap<Mid, MediaKind>>,
     // Throttle keyframe requests per mid
     pub last_kf_req: Arc<DashMap<Mid, Instant>>,
 }
 
 impl Publisher {
-    pub async fn new(
+    pub fn new<I, J>(
         participant_id: String,
         room_id: String,
         connection_type: ConnectionType,
         is_video_enabled: bool,
         is_audio_enabled: bool,
         is_e2ee_enabled: bool,
         streaming_protocol: StreamingProtocol,
-        ice_candidate_callback: IceCandidateCallback,
-        joined_callback: JoinedCallback,
-    ) -> Result<Arc<Self>, RtcError> {
+        ice_handler: I,
+        joined_handler: J,
+    ) -> Result<Arc<Self>, RtcError>
+    where
+        I: IceCandidateHandler,
+        J: JoinedHandler + Clone,
+    {
         // Create str0m RTC instance
         let rtc = Rtc::builder().build();
 
@@ -82,8 +86,8 @@ impl Publisher {
             streaming_protocol,
             // track_event_sender: Some(track_event_sender),
             // track_event_receiver: Some(track_event_receiver),
-            ice_candidate_callback: Some(ice_candidate_callback),
-            joined_callback: Some(joined_callback),
+            // ice_handler: Some(ice_handler),
+            // joined_handler: Some(joined_handler),
             incoming_mid_kind: Arc::new(DashMap::new()),
             last_kf_req: Arc::new(DashMap::new()),
         });
@@ -99,22 +103,23 @@ impl Publisher {
             .ok();
 
         // Announce host candidate to the client via callback
-        if let (Some(cb), Some(host)) = (
-            publisher.ice_candidate_callback.as_ref(),
-            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()),
                 Some(0),
             );
-            (cb)(ice).await;
+            cb.handle_candidate(ice);
         }
 
         // 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).await;
+            publisher_clone
+                .run_event_loop(event_rx, joined_handler)
+                .await;
         });
 
         Ok(publisher)
@@ -206,22 +211,26 @@ impl Publisher {
         self.tracks.clear();
     }
 
-    async fn run_event_loop(self: Arc<Self>, mut rx: UnboundedReceiver<Event>) {
+    async fn run_event_loop<J>(self: Arc<Self>, mut rx: UnboundedReceiver<Event>, joined_handler: J)
+    where
+        J: JoinedHandler + Clone,
+    {
         while let Some(event) = rx.recv().await {
-            self.handle_rtc_event(event).await;
+            self.handle_rtc_event(event, joined_handler.clone()).await;
             if self.cancel_token.is_cancelled() {
                 break;
             }
         }
     }
 
-    async fn handle_rtc_event(&self, event: Event) {
+    async fn handle_rtc_event<J>(&self, event: Event, joined_handler: J)
+    where
+        J: JoinedHandler,
+    {
         match event {
             Event::Connected => {
                 tracing::info!("Publisher {} connected", self.participant_id);
-                if let Some(callback) = &self.joined_callback {
-                    (callback)(true).await;
-                }
+                joined_handler.handle_joined(true);
             }
             Event::IceConnectionStateChange(state) => {
                 tracing::debug!("ICE connection state changed: {:?}", state);

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

@@ -8,11 +8,11 @@ use crate::{
     entities::{publisher::Publisher, subscriber::Subscriber},
     errors::RtcError,
     models::{
+        callbacks::{IceCandidateHandler, JoinedHandler, RenegotiationHandler},
         connection_type::ConnectionType,
-        params::{
-            IceCandidate, JoinRoomParams, JoinRoomResponse, RtcManagerConfigs,
-            SubscribeHlsLiveStreamParams, SubscribeHlsLiveStreamResponse, SubscribeParams,
-            SubscribeResponse,
+        rtc_dto::{
+            IceCandidate, JoinRoomParameters, JoinRoomResponse, SubscribeHlsLiveStreamParams,
+            SubscribeHlsLiveStreamResponse, SubscribeParameters, SubscribeResponse,
         },
     },
 };
@@ -22,24 +22,25 @@ pub struct Room {
     pub room_id: String,
     publishers: Arc<DashMap<String, Arc<Publisher>>>,
     subscribers: Arc<DashMap<String, Arc<Subscriber>>>,
-    config: RtcManagerConfigs,
 }
 
 impl Room {
-    pub fn new(room_id: String, config: RtcManagerConfigs) -> Self {
+    pub fn new(room_id: String) -> Self {
         Self {
             room_id,
             publishers: Arc::new(DashMap::new()),
             subscribers: Arc::new(DashMap::new()),
-            config,
         }
     }
 
-    pub async fn join_room(
+    pub fn join_room<I, J>(
         &mut self,
-        params: JoinRoomParams,
-        _room_id: &str,
-    ) -> Result<Option<JoinRoomResponse>, RtcError> {
+        params: JoinRoomParameters<I, J>,
+    ) -> Result<Option<JoinRoomResponse>, RtcError>
+    where
+        I: IceCandidateHandler,
+        J: JoinedHandler + Clone,
+    {
         let participant_id = params.participant_id.clone();
 
         info!("participant joining {}", participant_id.clone());
@@ -53,10 +54,9 @@ impl Room {
             params.is_audio_enabled,
             params.is_e2ee_enabled,
             params.streaming_protocol,
-            params.on_candidate,
-            params.callback,
-        )
-        .await?;
+            params.ice_handler,
+            params.joined_handler,
+        )?;
 
         info!("publisher created");
 
@@ -90,10 +90,14 @@ impl Room {
         }
     }
 
-    pub async fn subscribe(
+    pub fn subscribe<I, R>(
         &mut self,
-        params: SubscribeParams,
-    ) -> Result<SubscribeResponse, RtcError> {
+        params: SubscribeParameters<I, R>,
+    ) -> Result<SubscribeResponse, RtcError>
+    where
+        I: IceCandidateHandler,
+        R: RenegotiationHandler,
+    {
         let target_id = params.target_id.clone();
         let participant_id = params.participant_id.clone();
 
@@ -108,10 +112,9 @@ impl Room {
         let subscriber = Subscriber::new(
             participant_id.clone(),
             target_id.clone(),
-            params.on_candidate,
-            params.on_negotiation_needed,
-        )
-        .await?;
+            params.ice_handler,
+            params.renegotiation_handler,
+        )?;
 
         // Add subscriber to publisher's subscriber list
         publisher.add_subscriber(participant_id.clone(), subscriber.clone());
@@ -132,7 +135,7 @@ impl Room {
             is_screen_sharing: false,        // TODO: Get from publisher state
             is_hand_raising: false,          // TODO: Get from publisher state
             is_e2ee_enabled: false,          // TODO: Get from publisher state
-            video_codec: "H264".to_string(),  // TODO: Get from publisher
+            video_codec: "H264".to_string(), // TODO: Get from publisher
             screen_track_id: "".to_string(), // TODO: Get from publisher if screen sharing
         })
     }

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

@@ -16,7 +16,7 @@ use tokio_util::sync::CancellationToken;
 use crate::{
     errors::RtcError,
     models::{
-        params::{IceCandidateCallback, RenegotiationCallback},
+        callbacks::{IceCandidateHandler, RenegotiationHandler},
         quality::TrackQuality,
     },
 };
@@ -46,18 +46,22 @@ pub struct Subscriber {
     pub client_requested_quality: Arc<RwLock<Option<TrackQuality>>>,
     pub send_video_mid: Arc<RwLock<Option<Mid>>>,
     pub send_audio_mid: Arc<RwLock<Option<Mid>>>,
-    pub ice_candidate_callback: Option<IceCandidateCallback>,
-    pub renegotiation_callback: Option<RenegotiationCallback>,
+    // pub ice_handler: Option<I>,
+    // pub renegotiation_handler: Option<R>,
     pending: Arc<RwLock<Option<SdpPendingOffer>>>,
 }
 
 impl Subscriber {
-    pub async fn new(
+    pub fn new<I, R>(
         participant_id: String,
         target_id: String,
-        ice_candidate_callback: IceCandidateCallback,
-        renegotiation_callback: RenegotiationCallback,
-    ) -> Result<Arc<Self>, RtcError> {
+        ice_handler: I,
+        _renegotiation_handler: R,
+    ) -> Result<Arc<Self>, RtcError>
+    where
+        I: IceCandidateHandler,
+        R: RenegotiationHandler,
+    {
         // Create str0m RTC instance
         let rtc = Rtc::builder().build();
 
@@ -75,8 +79,6 @@ impl Subscriber {
             client_requested_quality: Arc::new(RwLock::new(None)),
             send_video_mid: Arc::new(RwLock::new(None)),
             send_audio_mid: Arc::new(RwLock::new(None)),
-            ice_candidate_callback: Some(ice_candidate_callback),
-            renegotiation_callback: Some(renegotiation_callback),
             pending: Arc::new(RwLock::new(None)),
         });
 
@@ -92,15 +94,15 @@ impl Subscriber {
 
         // Announce host candidate to the client via callback
         if let (Some(cb), Some(host)) = (
-            subscriber.ice_candidate_callback.as_ref(),
+            Some(ice_handler),
             RtcUdpRuntime::global().host_candidate(),
         ) {
             let ice = crate::utils::ice_utils::IceUtils::convert_from_str0m_candidate(
                 &host,
                 Some("0".to_string()),
                 Some(0),
             );
-            (cb)(ice).await;
+            cb.handle_candidate(ice);
         }
 
         // Start the RTC event loop

crates/rtc-manager/src/models/callbacks.rs

@@ -0,0 +1,13 @@
+use crate::models::rtc_dto::IceCandidate;
+
+pub trait IceCandidateHandler: Send + Sync + 'static {
+    fn handle_candidate(&self, candidate: IceCandidate);
+}
+
+pub trait JoinedHandler: Send + Sync + 'static {
+    fn handle_joined(&self, is_migrate: bool);
+}
+
+pub trait RenegotiationHandler: Send + Sync + 'static {
+    fn handle_renegotiation(&self, sdp: String);
+}

crates/rtc-manager/src/models/mod.rs

@@ -1,5 +1,6 @@
+pub mod callbacks;
 pub mod connection_type;
-pub mod params;
 pub mod quality;
+pub mod rtc_dto;
 pub mod rtp_forward_info;
 pub mod streaming_protocol;

crates/rtc-manager/src/models/params.rs crates/rtc-manager/src/models/rtc_dto.rscrates/rtc-manager/src/models/params.rs renamed to crates/rtc-manager/src/models/rtc_dto.rs

@@ -1,19 +1,15 @@
-use std::{pin::Pin, sync::Arc, future::Future};
 use parking_lot::RwLock;
 use serde::Serialize;
+use std::sync::Arc;
 
-use crate::models::streaming_protocol::StreamingProtocol;
 use super::connection_type::ConnectionType;
-
-pub type IceCandidateCallback =
-    Arc<dyn Fn(IceCandidate) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync>;
-pub type RenegotiationCallback =
-    Arc<dyn Fn(String) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync>;
-pub type JoinedCallback =
-    Arc<dyn Fn(bool) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync>;
+use crate::models::{
+    callbacks::{IceCandidateHandler, JoinedHandler, RenegotiationHandler},
+    streaming_protocol::StreamingProtocol,
+};
 
 #[derive(Debug, Clone)]
-pub struct RtcManagerConfigs {
+pub struct RtcManagerConfig {
     pub public_ip: String,
     pub port_min: u16,
     pub port_max: u16,
@@ -25,17 +21,22 @@ pub struct WClient {
     pub room_id: String,
 }
 
-#[derive(Clone)]
-pub struct JoinRoomParams {
-    pub sdp: String,
+pub struct JoinRoomParameters<I, J>
+where
+    I: IceCandidateHandler,
+    J: JoinedHandler,
+{
+    pub client_id: String,
     pub participant_id: String,
+    pub room_id: String,
+    pub sdp: String,
     pub is_video_enabled: bool,
     pub is_audio_enabled: bool,
     pub is_e2ee_enabled: bool,
     pub total_tracks: u8,
     pub connection_type: ConnectionType,
-    pub callback: JoinedCallback,
-    pub on_candidate: IceCandidateCallback,
+    pub joined_handler: J,
+    pub ice_handler: I,
     pub streaming_protocol: StreamingProtocol,
     pub is_ipv6_supported: bool,
 }
@@ -48,11 +49,17 @@ pub struct JoinRoomResponse {
 }
 
 #[derive(Clone)]
-pub struct SubscribeParams {
+pub struct SubscribeParameters<I, R>
+where
+    I: IceCandidateHandler,
+    R: RenegotiationHandler,
+{
+    pub client_id: String,
+    pub room_id: String,
     pub target_id: String,
     pub participant_id: String,
-    pub on_negotiation_needed: RenegotiationCallback,
-    pub on_candidate: IceCandidateCallback,
+    pub renegotiation_handler: R,
+    pub ice_handler: I,
     pub is_ipv6_supported: bool,
 }