[*] fix: webrtc whep_server_demos

heng30 · heng30 · commit a77e512367d9 · 2025-12-08T11:03:18.000+08:00
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/lib/wrtc/Cargo.toml b/lib/wrtc/Cargo.toml
@@ -29,6 +29,7 @@ reqwest = { workspace = true, features = ["json"] }
 [dev-dependencies]
 image.workspace = true
 hound.workspace = true
+once_cell.workspace = true
 stunclient.workspace = true
 env_logger.workspace = true
 serde_json.workspace = true
diff --git a/lib/wrtc/examples/whep_server2_demo.rs b/lib/wrtc/examples/whep_server2_demo.rs
@@ -1,11 +1,13 @@
 use anyhow::{Result, bail};
 use hound::WavReader;
 use image::{ImageBuffer, Rgb};
+use once_cell::sync::Lazy;
 use std::{
+    collections::HashSet,
     fs::File,
     path::Path,
     sync::{
-        Arc,
+        Arc, Mutex,
         atomic::{AtomicI32, Ordering},
     },
     time::{Duration, Instant},
@@ -19,6 +21,7 @@ use wrtc::{
 
 const IMG_WIDTH: u32 = 1920;
 const IMG_HEIGHT: u32 = 1080;
+static CONNECTIONS: Lazy<Mutex<HashSet<String>>> = Lazy::new(|| Mutex::new(HashSet::default()));
 
 #[tokio::main]
 async fn main() -> Result<()> {
@@ -35,35 +38,32 @@ async fn main() -> Result<()> {
 
     let packet_sender_clone = packet_sender.clone();
     tokio::spawn(async move {
-        let connections = Arc::new(AtomicI32::new(0));
         loop {
             tokio::select! {
                 ev = event_receiver.recv() => {
                     match ev {
-                        Ok(Event::PeerConnected(_)) => {
-                            if connections.load(Ordering::Relaxed) == 0 {
-                                h264_streaming_thread(packet_sender_clone.clone(), connections.clone());
-                                wav_stream_thread(packet_sender_clone.clone(), audio_path.clone(), connections.clone());
+                        Ok(Event::PeerConnected(addr)) => {
+                            let mut connections = CONNECTIONS.lock().unwrap();
+                            if connections.is_empty(){
+                                h264_streaming_thread(packet_sender_clone.clone());
+                                wav_stream_thread(packet_sender_clone.clone(), audio_path.clone());
                             }
 
-                            let count = connections.fetch_add(1, Ordering::Relaxed);
-                            log::info!("connections count: {}", count + 1);
+                            connections.insert(addr);
+                            log::info!("connections count: {}", connections.len());
                         }
                         Ok(Event::LocalClosed(addr)) => {
-                            if connections.fetch_sub(1, Ordering::Relaxed) == 0 {
-                                connections.store(0, Ordering::Relaxed);
-                            }
-
                             log::info!("LocalClosed({addr})");
-                            log::info!("connections count: {}", connections.load(Ordering::Relaxed));
+
+                            let mut connections = CONNECTIONS.lock().unwrap();
+                            connections.remove(&addr);
+                            log::info!("connections count: {}", connections.len());
                         }
                         Ok(Event::PeerClosed(addr)) => {
-                            if connections.fetch_sub(1, Ordering::Relaxed) == 0 {
-                                connections.store(0, Ordering::Relaxed);
-                            }
-
                             log::info!("PeerClosed({addr})");
-                            log::info!("connections count: {}", connections.load(Ordering::Relaxed));
+                            let mut connections = CONNECTIONS.lock().unwrap();
+                            connections.remove(&addr);
+                            log::info!("connections count: {}", connections.len());
                         }
                         _ => (),
                     }
@@ -89,7 +89,7 @@ async fn main() -> Result<()> {
     Ok(())
 }
 
-fn h264_streaming_thread(packet_sender: Sender<PacketData>, connections: Arc<AtomicI32>) {
+fn h264_streaming_thread(packet_sender: Sender<PacketData>) {
     std::thread::spawn(move || {
         let fps = 25;
 
@@ -133,7 +133,7 @@ fn h264_streaming_thread(packet_sender: Sender<PacketData>, connections: Arc<Ato
 
                 std::thread::sleep(Duration::from_secs_f64(1 as f64 / fps as f64));
 
-                if connections.load(Ordering::Relaxed) <= 0 {
+                if CONNECTIONS.lock().unwrap().is_empty() {
                     break 'out;
                 }
             }
@@ -149,7 +149,7 @@ fn h264_streaming_thread(packet_sender: Sender<PacketData>, connections: Arc<Ato
                 log::warn!("Failed to flush encoder frame: {:?}", e);
             }
 
-            if connections.load(Ordering::Relaxed) <= 0 {
+            if CONNECTIONS.lock().unwrap().is_empty() {
                 break;
             }
         }
@@ -158,11 +158,7 @@ fn h264_streaming_thread(packet_sender: Sender<PacketData>, connections: Arc<Ato
     });
 }
 
-fn wav_stream_thread(
-    packet_sender: Sender<PacketData>,
-    audio_file: String,
-    connections: Arc<AtomicI32>,
-) {
+fn wav_stream_thread(packet_sender: Sender<PacketData>, audio_file: String) {
     tokio::spawn(async move {
         'out: loop {
             let file = File::open(&audio_file).unwrap();
@@ -236,12 +232,12 @@ fn wav_stream_thread(
 
                 ticker.tick().await;
 
-                if connections.load(Ordering::Relaxed) <= 0 {
+                if CONNECTIONS.lock().unwrap().is_empty() {
                     break 'out;
                 }
             }
 
-            if connections.load(Ordering::Relaxed) <= 0 {
+            if CONNECTIONS.lock().unwrap().is_empty() {
                 break;
             }
         }
diff --git a/lib/wrtc/examples/whep_server_demo.rs b/lib/wrtc/examples/whep_server_demo.rs
@@ -1,22 +1,22 @@
 use anyhow::{Result, bail};
+use once_cell::sync::Lazy;
 use std::{
+    collections::HashSet,
     fs::File,
     io::BufReader,
     path::Path,
-    sync::{
-        Arc,
-        atomic::{AtomicI32, Ordering},
-    },
+    sync::Mutex,
     time::{Duration, Instant},
 };
 use tokio::sync::broadcast::{self, Sender};
-use webrtc::media::io::h264_reader::H264Reader;
-use webrtc::media::io::ogg_reader::OggReader;
+use webrtc::media::io::{h264_reader::H264Reader, ogg_reader::OggReader};
 use wrtc::{
     Event, PacketData, opus::OPUS_SAMPLE_RATE, session::WebRTCServerSessionConfig,
     webrtc::WebRTCServer,
 };
 
+static CONNECTIONS: Lazy<Mutex<HashSet<String>>> = Lazy::new(|| Mutex::new(HashSet::default()));
+
 #[tokio::main]
 async fn main() -> Result<()> {
     env_logger::init();
@@ -37,36 +37,32 @@ async fn main() -> Result<()> {
 
     let packet_sender_clone = packet_sender.clone();
     tokio::spawn(async move {
-        let connections = Arc::new(AtomicI32::new(0));
-
         loop {
             tokio::select! {
                 ev = event_receiver.recv() => {
                     match ev {
-                        Ok(Event::PeerConnected(_)) => {
-                            if connections.load(Ordering::Relaxed) == 0 {
-                                h264_streaming_thread(packet_sender_clone.clone(), video_path.clone(), connections.clone());
-                                ogg_stream_thread(packet_sender_clone.clone(), audio_path.clone(), connections.clone());
+                        Ok(Event::PeerConnected(addr)) => {
+                            let mut connections = CONNECTIONS.lock().unwrap();
+                            if connections.is_empty(){
+                                h264_streaming_thread(packet_sender_clone.clone(), video_path.clone());
+                                ogg_stream_thread(packet_sender_clone.clone(), audio_path.clone());
                             }
 
-                            let count = connections.fetch_add(1, Ordering::Relaxed);
-                            log::info!("connections count: {}", count + 1);
+                            connections.insert(addr);
+                            log::info!("connections count: {}", connections.len());
                         }
                         Ok(Event::LocalClosed(addr)) => {
-                            if connections.fetch_sub(1, Ordering::Relaxed) == 0 {
-                                connections.store(0, Ordering::Relaxed);
-                            }
-
                             log::info!("LocalClosed({addr})");
-                            log::info!("connections count: {}", connections.load(Ordering::Relaxed));
+                            let mut connections = CONNECTIONS.lock().unwrap();
+                            connections.remove(&addr);
+                            log::info!("connections count: {}", connections.len());
                         }
                         Ok(Event::PeerClosed(addr)) => {
-                            if connections.fetch_sub(1, Ordering::Relaxed) == 0 {
-                                connections.store(0, Ordering::Relaxed);
-                            }
-
                             log::info!("PeerClosed({addr})");
-                            log::info!("connections count: {}", connections.load(Ordering::Relaxed));
+                            let mut connections = CONNECTIONS.lock().unwrap();
+                            connections.remove(&addr);
+                            log::info!("connections count: {}", connections.len());
+
                         }
                         _ => (),
                     }
@@ -92,11 +88,7 @@ async fn main() -> Result<()> {
     Ok(())
 }
 
-fn h264_streaming_thread(
-    packet_sender: Sender<PacketData>,
-    video_file: String,
-    connections: Arc<AtomicI32>,
-) {
+fn h264_streaming_thread(packet_sender: Sender<PacketData>, video_file: String) {
     tokio::spawn(async move {
         'out: loop {
             let file = File::open(&video_file.clone()).unwrap();
@@ -128,12 +120,12 @@ fn h264_streaming_thread(
 
                 _ = ticker.tick().await;
 
-                if connections.load(Ordering::Relaxed) <= 0 {
+                if CONNECTIONS.lock().unwrap().is_empty() {
                     break 'out;
                 }
             }
 
-            if connections.load(Ordering::Relaxed) <= 0 {
+            if CONNECTIONS.lock().unwrap().is_empty() {
                 break;
             }
         }
@@ -142,11 +134,7 @@ fn h264_streaming_thread(
     });
 }
 
-fn ogg_stream_thread(
-    packet_sender: Sender<PacketData>,
-    audio_file: String,
-    connections: Arc<AtomicI32>,
-) {
+fn ogg_stream_thread(packet_sender: Sender<PacketData>, audio_file: String) {
     tokio::spawn(async move {
         'out: loop {
             let file = File::open(&audio_file).unwrap();
@@ -172,12 +160,12 @@ fn ogg_stream_thread(
 
                 _ = ticker.tick().await;
 
-                if connections.load(Ordering::Relaxed) <= 0 {
+                if CONNECTIONS.lock().unwrap().is_empty() {
                     break 'out;
                 }
             }
 
-            if connections.load(Ordering::Relaxed) <= 0 {
+            if CONNECTIONS.lock().unwrap().is_empty() {
                 break;
             }
         }
diff --git a/lib/wrtc/src/session.rs b/lib/wrtc/src/session.rs
@@ -148,10 +148,7 @@ impl WebRTCServerSession {
             let ty = eles[1];
 
             if eles.len() < 2 {
-                log::warn!(
-                    "WebRTCServerSession::run the http path is not correct: {}",
-                    http_request.uri.path
-                );
+                log::warn!("the http path is not correct: {}", http_request.uri.path);
 
                 return Err(SessionError::HttpRequestPathError);
             }
@@ -247,10 +244,7 @@ impl WebRTCServerSession {
                 }
                 http_method_name::PATCH => (),
                 _ => {
-                    log::warn!(
-                        "WebRTCServerSession::unsupported method name: {}",
-                        http_request.method
-                    );
+                    log::warn!("unsupported method name: {}", http_request.method);
                 }
             }
 
@@ -271,7 +265,7 @@ impl WebRTCServerSession {
 
             sessions_unlock.remove(uuid);
         } else {
-            log::warn!("the session :{uuid}  is not exited.");
+            log::warn!("the session: [{uuid}] is not exited.");
         }
     }
 

Original file line number	Diff line number	Diff line change
`@@ -148,10 +148,7 @@ impl WebRTCServerSession {`
`148`	`148`	`let ty = eles[1];`
`149`	`149`
`150`	`150`	`if eles.len() < 2 {`
`151`		`- log::warn!(`
`152`		`- "WebRTCServerSession::run the http path is not correct: {}",`
`153`		`- http_request.uri.path`
`154`		`- );`
	`151`	`+ log::warn!("the http path is not correct: {}", http_request.uri.path);`
`155`	`152`
`156`	`153`	`return Err(SessionError::HttpRequestPathError);`
`157`	`154`	`}`
`@@ -247,10 +244,7 @@ impl WebRTCServerSession {`
`247`	`244`	`}`
`248`	`245`	`http_method_name::PATCH => (),`
`249`	`246`	`_ => {`
`250`		`- log::warn!(`
`251`		`- "WebRTCServerSession::unsupported method name: {}",`
`252`		`- http_request.method`
`253`		`- );`
	`247`	`+ log::warn!("unsupported method name: {}", http_request.method);`
`254`	`248`	`}`
`255`	`249`	`}`
`256`	`250`
`@@ -271,7 +265,7 @@ impl WebRTCServerSession {`
`271`	`265`
`272`	`266`	`sessions_unlock.remove(uuid);`
`273`	`267`	`} else {`
`274`		`- log::warn!("the session :{uuid} is not exited.");`
	`268`	`+ log::warn!("the session: [{uuid}] is not exited.");`
`275`	`269`	`}`
`276`	`270`	`}`
`277`	`271`