[*] website can play audio

Author: heng30

lib/wrtc/data/README.md

@@ -0,0 +1,4 @@
+```bash
+ffmpeg -i $INPUT_FILE -an -c:v libx264 -bsf:v h264_mp4toannexb -b:v 2M -max_delay 0 -bf 0 output.h264
+ffmpeg -i $INPUT_FILE -c:a libopus -page_duration 20000 -vn output.ogg
+```

lib/wrtc/data/test.ogg

@@ -3,17 +3,25 @@ use std::{
     fs::File,
     io::BufReader,
     path::Path,
+    sync::{
+        Arc,
+        atomic::{AtomicI32, Ordering},
+    },
     time::{Duration, Instant},
 };
 use tokio::sync::broadcast::{self, Sender};
 use webrtc::media::io::h264_reader::H264Reader;
-use wrtc::{Event, PacketData, session::WebRTCServerSessionConfig, webrtc::WebRTCServer};
+use webrtc::media::io::ogg_reader::OggReader;
+use wrtc::{
+    Event, OPUS_SAMPLE_RATE, PacketData, session::WebRTCServerSessionConfig, webrtc::WebRTCServer,
+};
 
 #[tokio::main]
 async fn main() -> Result<()> {
     env_logger::init();
 
     let video_path = "./data/test.h264".to_string();
+    let audio_path = "./data/test.ogg".to_string();
     let config = WebRTCServerSessionConfig::default();
     let (packet_sender, _) = broadcast::channel(128);
     let (event_sender, mut event_receiver) = broadcast::channel(16);
@@ -22,34 +30,42 @@ async fn main() -> Result<()> {
         bail!("video file: '{video_path}' not exist");
     }
 
+    if !Path::new(&audio_path).exists() {
+        bail!("audio file: '{audio_path}' not exist");
+    }
+
     let packet_sender_clone = packet_sender.clone();
     tokio::spawn(async move {
-        let mut connections = 0;
+        let connections = Arc::new(AtomicI32::new(0));
+
         loop {
             tokio::select! {
                 ev = event_receiver.recv() => {
                     match ev {
                         Ok(Event::PeerConnected(_)) => {
-                            if connections == 0 {
-                                h264_streaming_thread(packet_sender_clone.clone(), video_path.clone());
+                            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());
                             }
 
-                            connections += 1;
-                            log::info!("connections count: {connections}");
+                            let count = connections.fetch_add(1, Ordering::Relaxed);
+                            log::info!("connections count: {}", count + 1);
                         }
                         Ok(Event::LocalClosed(addr)) => {
-                            if connections > 0 {
-                                connections  -= 1;
+                            if connections.fetch_sub(1, Ordering::Relaxed) == 0 {
+                                connections.store(0, Ordering::Relaxed);
                             }
+
                             log::info!("LocalClosed({addr})");
-                            log::info!("connections count: {connections}");
+                            log::info!("connections count: {}", connections.load(Ordering::Relaxed));
                         }
                         Ok(Event::PeerClosed(addr)) => {
-                            if connections > 0 {
-                                connections  -= 1;
+                            if connections.fetch_sub(1, Ordering::Relaxed) == 0 {
+                                connections.store(0, Ordering::Relaxed);
                             }
+
                             log::info!("PeerClosed({addr})");
-                            log::info!("connections count: {connections}");
+                            log::info!("connections count: {}", connections.load(Ordering::Relaxed));
                         }
                         _ => (),
                     }
@@ -75,9 +91,13 @@ async fn main() -> Result<()> {
     Ok(())
 }
 
-fn h264_streaming_thread(packet_sender: Sender<PacketData>, video_file: String) {
+fn h264_streaming_thread(
+    packet_sender: Sender<PacketData>,
+    video_file: String,
+    connections: Arc<AtomicI32>,
+) {
     tokio::spawn(async move {
-        loop {
+        'out: loop {
             let file = File::open(&video_file.clone()).unwrap();
             let reader = BufReader::new(file);
             let mut h264 = H264Reader::new(reader, 1_048_576);
@@ -106,6 +126,58 @@ fn h264_streaming_thread(packet_sender: Sender<PacketData>, video_file: String)
                 };
 
                 _ = ticker.tick().await;
+
+                if connections.load(Ordering::Relaxed) <= 0 {
+                    break 'out;
+                }
+            }
+
+            if connections.load(Ordering::Relaxed) <= 0 {
+                break;
+            }
+        }
+
+        log::info!("h264_streaming_thread exit...");
+    });
+}
+
+fn ogg_stream_thread(
+    packet_sender: Sender<PacketData>,
+    audio_file: String,
+    connections: Arc<AtomicI32>,
+) {
+    tokio::spawn(async move {
+        'out: loop {
+            let file = File::open(&audio_file).unwrap();
+            let reader = BufReader::new(file);
+            let (mut ogg, _) = OggReader::new(reader, true).unwrap();
+            let page_duration = Duration::from_millis(20);
+
+            let mut last_granule = 0;
+            let mut ticker = tokio::time::interval(page_duration);
+
+            while let Ok((page_data, page_header)) = ogg.parse_next_page() {
+                let sample_count = page_header.granule_position - last_granule;
+                last_granule = page_header.granule_position;
+                let sample_duration = Duration::from_millis(sample_count * 1000 / OPUS_SAMPLE_RATE);
+
+                if let Err(e) = packet_sender.send(PacketData::Audio {
+                    timestamp: Instant::now(),
+                    duration: sample_duration,
+                    data: page_data.freeze().into(),
+                }) {
+                    log::warn!("send audio data failed: {e}");
+                }
+
+                _ = ticker.tick().await;
+
+                if connections.load(Ordering::Relaxed) <= 0 {
+                    break 'out;
+                }
+            }
+
+            if connections.load(Ordering::Relaxed) <= 0 {
+                break;
             }
         }
     });

lib/wrtc/examples/whep_server_demo.rs

@@ -3,16 +3,18 @@ pub mod session;
 pub mod webrtc;
 pub mod whep;
 
+pub const OPUS_SAMPLE_RATE: u64 = 48000;
+
 #[derive(Clone)]
 pub enum PacketData {
     Video {
         timestamp: std::time::Instant,
-        data: bytes::BytesMut,
+        data: bytes::Bytes,
     },
     Audio {
         timestamp: std::time::Instant,
-        channels: u8,
-        data: bytes::BytesMut,
+        duration: std::time::Duration,
+        data: bytes::Bytes,
     },
 }
 

lib/wrtc/src/lib.rs

@@ -22,8 +22,6 @@ use webrtc::{
     track::track_local::{TrackLocal, track_local_static_sample::TrackLocalStaticSample},
 };
 
-const OPUS_SAMPLE_RATE: u64 = 48000;
-
 pub type Result<T> = std::result::Result<T, WebRTCError>;
 
 #[derive(Debug, Setters, Clone)]
@@ -160,20 +158,19 @@ pub async fn handle_whep(
 
                                     if let Err(err) = video_track
                                         .write_sample(&Sample {
-                                        data: data.into(),
+                                        data,
                                         duration: Duration::from_secs(1),
                                         ..Default::default()
                                     }).await {
                                         log::warn!("send video data error: {}", err);
                                     }
                                 }
-                                PacketData::Audio { timestamp: _timestamp, channels, data } => {
-                                    log::trace!("{:?}: sending audio data ({}) bytes with {channels} channels", _timestamp.elapsed(), data.len());
+                                PacketData::Audio { timestamp: _timestamp, duration, data } => {
+                                    log::trace!("{:?}: sending audio data ({}) bytes with {duration:.2?}", _timestamp.elapsed(), data.len());
 
-                                    let sample_duration = Duration::from_millis(data.len() as u64 * 1000 / (OPUS_SAMPLE_RATE * channels.max(1 ) as u64)) ;
                                     if let Err(err) = audio_track.write_sample(&Sample {
-                                        data: data.into(),
-                                        duration: sample_duration,
+                                        data,
+                                        duration,
                                         ..Default::default()
                                     }).await {
                                         log::warn!("send audio data error: {}", err);