update .

Author: lambiengcode

crates/webrtc-manager/src/entities/str0m_room.rs

@@ -3,9 +3,11 @@ use std::sync::Arc;
 use dashmap::DashMap;
 use parking_lot::RwLock;
 use str0m::{
-    Candidate,
+    Candidate, Input,
     change::{SdpAnswer, SdpOffer},
+    media::MediaData,
 };
+use tracing::{debug, error, info};
 
 use crate::{
     entities::media::Media,
@@ -77,6 +79,22 @@ impl Str0mRoom {
                     (callback)(is_migrate).await;
                 });
             });
+
+            // Set media data callback to forward to subscribers
+            let room_clone = self.clone();
+            let participant_id_clone = participant_id.clone();
+            session_write.set_media_data_callback(move |media_data| {
+                let room_clone = room_clone.clone();
+                let participant_id = participant_id_clone.clone();
+                tokio::spawn(async move {
+                    if let Err(e) = room_clone
+                        .forward_media_data(&participant_id, media_data)
+                        .await
+                    {
+                        error!("Failed to forward media data: {:?}", e);
+                    }
+                });
+            });
         }
 
         // Add the session to publishers
@@ -85,19 +103,22 @@ impl Str0mRoom {
 
         // Handle SDP negotiation based on connection type
         if params.connection_type == ConnectionType::SFU {
+            println!("[Str0mRoom] params.sdp: {:?}", params.sdp.len());
             // For SFU, we need to accept the offer and create an answer
-            let offer: SdpOffer =
-                serde_json::from_str(&params.sdp).map_err(|_| WebRTCError::FailedToCreateOffer)?;
-
+            let offer: SdpOffer = SdpOffer::from_sdp_string(&params.sdp)
+                .map_err(|_| WebRTCError::FailedToCreateOffer)?;
+            println!("===> parse offer");
             let mut session_write = session.write();
             let answer = session_write.accept_offer(offer)?;
 
+            println!("===> accept offer");
+
             // Convert answer to string
-            let answer_json =
-                serde_json::to_string(&answer).map_err(|_| WebRTCError::FailedToCreateAnswer)?;
+            // let answer_json =
+            //     serde_json::to_string(&answer).map_err(|_| WebRTCError::FailedToCreateAnswer)?;
 
             return Ok(Some(JoinRoomResponse {
-                sdp: answer_json,
+                sdp: answer.to_sdp_string(),
                 is_recording: false,
             }));
         } else {
@@ -147,10 +168,10 @@ impl Str0mRoom {
                 let subscriber_session = self.session_manager.create_session(
                     participant_id.clone(),
                     ConnectionType::SFU,
-                    true,                   // is_video_enabled
-                    true,                   // is_audio_enabled
-                    false,                  // is_e2ee_enabled
-                    StreamingProtocol::HLS, // Use HLS as default for subscribers
+                    true,  // is_video_enabled
+                    true,  // is_audio_enabled
+                    false, // is_e2ee_enabled
+                    StreamingProtocol::SFU,
                 );
 
                 // Set up callbacks for subscriber
@@ -192,6 +213,40 @@ impl Str0mRoom {
         }
     }
 
+    /// Forward media data from a publisher to all its subscribers
+    pub async fn forward_media_data(
+        &self,
+        publisher_id: &str,
+        media_data: MediaData,
+    ) -> Result<(), WebRTCError> {
+        let prefix = format!("p_{publisher_id}_");
+
+        // Find all subscribers for this publisher
+        for entry in self.subscribers.iter() {
+            if entry.key().starts_with(&prefix) {
+                let subscriber = entry.value();
+                let mut session_write = subscriber.write();
+
+                // For now, we'll just log the media data forwarding
+                // In a real implementation, you would need to properly handle
+                // the media data according to str0m's API
+                debug!(
+                    "Forwarding media data from publisher {} to subscriber {}: mid={:?}, data_len={}",
+                    publisher_id,
+                    entry.key(),
+                    media_data.mid,
+                    media_data.data.len()
+                );
+
+                // TODO: Implement proper media data forwarding using str0m's API
+                // This would typically involve creating the appropriate media tracks
+                // and writing the data to them
+            }
+        }
+
+        Ok(())
+    }
+
     pub fn subscribe_hls_live_stream(
         &self,
         params: SubscribeHlsLiveStreamParams,

crates/webrtc-manager/src/lib.rs

@@ -4,5 +4,6 @@ pub mod models;
 pub mod str0m_network_manager;
 pub mod str0m_session;
 pub mod str0m_webrtc_manager;
-pub mod utils;
 pub mod webrtc_manager;
+
+pub use webrtc_manager::{JoinRoomRequest, NetworkManager, WebRTCManager};

crates/webrtc-manager/src/models/params.rs

@@ -41,7 +41,7 @@ pub struct JoinRoomParams {
     pub is_ipv6_supported: bool,
 }
 
-#[derive(Serialize)]
+#[derive(Serialize, Debug)]
 #[serde(rename_all = "camelCase")]
 pub struct JoinRoomResponse {
     pub sdp: String,

crates/webrtc-manager/src/str0m_session.rs

@@ -32,6 +32,7 @@ pub struct Str0mSession {
     pub on_candidate: Option<Box<dyn Fn(IceCandidate) + Send + Sync>>,
     pub on_negotiation_needed: Option<Box<dyn Fn(String) + Send + Sync>>,
     pub on_connected: Option<Box<dyn Fn() + Send + Sync>>,
+    pub on_media_data: Option<Box<dyn Fn(MediaData) + Send + Sync>>,
 }
 
 impl Str0mSession {
@@ -68,6 +69,7 @@ impl Str0mSession {
             on_candidate: None,
             on_negotiation_needed: None,
             on_connected: None,
+            on_media_data: None,
         }
     }
 
@@ -194,8 +196,11 @@ impl Str0mSession {
             data.data.len()
         );
 
-        // Here you would typically forward the media data to subscribers
-        // For now, we'll just log it
+        // Forward media data to subscribers if callback is set
+        if let Some(callback) = &self.on_media_data {
+            callback(data);
+        }
+
         Ok(())
     }
 
@@ -223,6 +228,14 @@ impl Str0mSession {
         self.on_connected = Some(Box::new(callback));
     }
 
+    /// Set the media data callback for forwarding to subscribers
+    pub fn set_media_data_callback<F>(&mut self, callback: F)
+    where
+        F: Fn(MediaData) + Send + Sync + 'static,
+    {
+        self.on_media_data = Some(Box::new(callback));
+    }
+
     /// Check if the session is connected
     pub fn is_connected(&self) -> bool {
         self.is_connected

crates/webrtc-manager/src/str0m_webrtc_manager.rs

@@ -89,6 +89,8 @@ impl Str0mWebRTCManager {
             room.join_room(params, room_id).await?
         };
 
+        println!("[Str0mWebRTCManager] res: {:?}", res);
+
         Ok(res)
     }
 

crates/webrtc-manager/src/webrtc_manager.rs

@@ -1,7 +1,48 @@
-use crate::str0m_webrtc_manager::{JoinRoomReq, Str0mWebRTCManager};
+use std::sync::Arc;
+use tokio::sync::Mutex;
+use tracing::info;
 
-// Re-export the str0m implementation as the main WebRTC manager
-pub type WebRTCManager = Str0mWebRTCManager;
+use crate::{
+    str0m_network_manager::Str0mNetworkManager,
+    str0m_webrtc_manager::{JoinRoomReq, Str0mWebRTCManager},
+};
 
-// Re-export the join room request type
+pub type WebRTCManager = Str0mWebRTCManager;
 pub type JoinRoomRequest = JoinRoomReq;
+
+pub struct NetworkManager {
+    network_manager: Arc<Mutex<Str0mNetworkManager>>,
+    webrtc_manager: Arc<Str0mWebRTCManager>,
+}
+
+impl NetworkManager {
+    pub fn new(bind_addr: std::net::SocketAddr) -> Result<Self, crate::errors::WebRTCError> {
+        let network_manager = Str0mNetworkManager::new(bind_addr)?;
+        let webrtc_manager = Str0mWebRTCManager::new();
+
+        Ok(Self {
+            network_manager: Arc::new(Mutex::new(network_manager)),
+            webrtc_manager: Arc::new(webrtc_manager),
+        })
+    }
+
+    pub fn get_webrtc_manager(&self) -> Arc<Str0mWebRTCManager> {
+        self.webrtc_manager.clone()
+    }
+
+    pub async fn start_network_loop(&self) {
+        let network_manager = self.network_manager.clone();
+
+        tokio::spawn(async move {
+            info!("Starting network loop...");
+            loop {
+                let mut manager = network_manager.lock().await;
+                if let Err(e) = manager.run_network_loop() {
+                    tracing::error!("Network loop error: {:?}", e);
+                }
+                // Small delay to prevent busy waiting
+                tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
+            }
+        });
+    }
+}

justfile

@@ -1,7 +1,7 @@
 signalling:
 	cargo run --bin signalling
 sfu:
-	cargo run --bin sfu --release
+	cargo run --bin sfu
 build-proto:
 	cargo build -p waterbus-proto 
 build-signalling: