[*] website whep client

Author: heng30

Cargo.lock

@@ -10,6 +10,7 @@ log.workspace = true
 rand.workspace = true
 http.workspace = true
 bytes.workspace = true
+rubato.workspace = true
 anyhow.workspace = true
 webrtc.workspace = true
 bytesio.workspace = true
@@ -19,7 +20,6 @@ byteorder.workspace = true
 thiserror.workspace = true
 derive_setters.workspace = true
 tokio = { workspace = true, features = ["full"] }
-cutil = { workspace = true, features = ["crypto"] }
 serde = { workspace = true, features = ["derive"] }
 
 [dev-dependencies]

lib/wrtc/Cargo.toml

@@ -1,54 +1,3 @@
-# WRTC Examples
-
-This directory contains example programs demonstrating various features of the wrtc library.
-
-## Opus Codec Demo
-
-The `opus_demo.rs` example demonstrates how to use the Opus audio codec for encoding and decoding audio data.
-
-### Features
-
-- Reads WAV audio files
-- Encodes audio data using Opus codec
-- Decodes encoded audio back to PCM
-- Writes decoded audio to new WAV file
-- Shows compression ratio statistics
-- Comprehensive logging with env_logger
-
-### Usage
-
-```bash
-# Run with default logging (info level)
-cargo run --example opus_demo
-
-# Run with debug logging for detailed frame information
-RUST_LOG=debug cargo run --example opus_demo
-
-# Run with only warnings and errors
-RUST_LOG=warn cargo run --example opus_demo
-
-# Run with custom log filtering
-RUST_LOG=opus_demo=debug cargo run --example opus_demo
-```
-
-### What it does
-
-1. **Reads** `data/test.wav` (48kHz, stereo, 5 seconds)
-2. **Encodes** audio to Opus packets (250 frames)
-3. **Decodes** Opus packets back to audio
-4. **Saves** result to `/tmp/opus-coder.wav`
-5. **Shows** compression statistics (~31:1 compression ratio)
-
-### Dependencies
-
-- `hound` - WAV file I/O
-- `env_logger` - Structured logging
-- `log` - Logging facade
-- `wrtc::opus` - Opus codec implementation
-
-### Log Levels
-
-- **INFO**: Basic progress information (default)
-- **DEBUG**: Detailed frame-by-frame encoding/decoding info
-- **WARN**: Error messages for failed operations
-- **ERROR**: Critical errors that stop execution
+### How to test?
+- run: `RUST_LOG=debug cargo run --example whep_server_demo` or `RUST_LOG=debug cargo run --example whep_server_demo`
+- open web broswer and visit `http://localhost:9090`

lib/wrtc/data/test-44100.wav

@@ -6,8 +6,8 @@ use wrtc::opus::{OpusCoder, OpusCoderError};
 fn main() -> Result<(), Box<dyn std::error::Error>> {
     env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")).init();
 
-    let input_path = Path::new("data/test.wav");
-    let output_path = "/tmp/opus-coder.wav";
+    let input_path = Path::new("data/test-44100.wav");
+    let output_path = "/tmp/opus-coder-44100.wav";
 
     info!("Opus Codec Demo");
     info!("===============");
@@ -39,7 +39,15 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     let mut opus_encoder = OpusCoder::new(spec.sample_rate, channels)?;
     let mut opus_decoder = OpusCoder::new(spec.sample_rate, channels)?;
-    info!("  Frame size: {} samples", opus_encoder.frame_size());
+    info!(
+        "  Internal frame size: {} samples (48kHz)",
+        opus_encoder.frame_size()
+    );
+    info!(
+        "  Input frame size: {} samples ({}Hz)",
+        (spec.sample_rate as usize * 20) / 1000,
+        spec.sample_rate
+    );
 
     // 3. Encode audio data
     info!("Encoding audio with Opus...");
@@ -52,9 +60,15 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     info!("  Decoded {} samples", decoded_audio.len());
 
     // 5. Write decoded audio to WAV file
-    info!("Writing decoded audio to WAV file...");
-    write_wav_file(output_path, &decoded_audio, spec)?;
+    info!("Writing decoded audio to WAV file at 48kHz...");
+
+    // Opus always outputs at 48kHz, so we save at 48kHz regardless of input sample rate
+    let mut output_spec = spec;
+    output_spec.sample_rate = 48000;
+
+    write_wav_file(output_path, &decoded_audio, output_spec)?;
     info!("  Output written to: {}", output_path);
+    info!("  Output sample rate: 48000 Hz (Opus native rate)");
 
     info!("Demo completed successfully!");
 
@@ -92,7 +106,7 @@ fn encode_audio(
     audio_data: &[f32],
 ) -> Result<Vec<Vec<u8>>, OpusCoderError> {
     let mut encoded_packets = Vec::new();
-    let samples_per_frame = opus_coder.samples_per_frame();
+    let samples_per_frame = opus_coder.input_samples_per_frame();
     let total_frames = (audio_data.len() + samples_per_frame - 1) / samples_per_frame;
 
     debug!(

lib/wrtc/examples/README.md

@@ -1,8 +1,8 @@
 use anyhow::{Result, bail};
+use hound::WavReader;
 use image::{ImageBuffer, Rgb};
 use std::{
     fs::File,
-    io::BufReader,
     path::Path,
     sync::{
         Arc,
@@ -12,9 +12,9 @@ use std::{
 };
 use tokio::sync::broadcast::{self, Sender};
 use video_encoder::{EncodedFrame, VideoEncoderConfig};
-use webrtc::media::io::ogg_reader::OggReader;
 use wrtc::{
-    Event, OPUS_SAMPLE_RATE, PacketData, session::WebRTCServerSessionConfig, webrtc::WebRTCServer,
+    Event, PacketData, common::auth::Auth, opus::OpusCoder, session::WebRTCServerSessionConfig,
+    webrtc::WebRTCServer,
 };
 
 const IMG_WIDTH: u32 = 1920;
@@ -24,7 +24,7 @@ const IMG_HEIGHT: u32 = 1080;
 async fn main() -> Result<()> {
     env_logger::init();
 
-    let audio_path = "./data/test.ogg".to_string();
+    let audio_path = "./data/test-44100.wav".to_string();
     let config = WebRTCServerSessionConfig::default();
     let (packet_sender, _) = broadcast::channel(128);
     let (event_sender, mut event_receiver) = broadcast::channel(16);
@@ -43,7 +43,7 @@ async fn main() -> Result<()> {
                         Ok(Event::PeerConnected(_)) => {
                             if connections.load(Ordering::Relaxed) == 0 {
                                 h264_streaming_thread(packet_sender_clone.clone(), connections.clone());
-                                ogg_stream_thread(packet_sender_clone.clone(), audio_path.clone(), connections.clone());
+                                wav_stream_thread(packet_sender_clone.clone(), audio_path.clone(), connections.clone());
                             }
 
                             let count = connections.fetch_add(1, Ordering::Relaxed);
@@ -79,7 +79,7 @@ async fn main() -> Result<()> {
     let mut server = WebRTCServer::new(
         config,
         "0.0.0.0:9090".to_string(),
-        None,
+        Some(Auth::new("123".to_string())),
         packet_sender,
         event_sender,
     );
@@ -158,35 +158,83 @@ fn h264_streaming_thread(packet_sender: Sender<PacketData>, connections: Arc<Ato
     });
 }
 
-fn ogg_stream_thread(
+fn wav_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 reader = WavReader::new(file).unwrap();
+            let spec = reader.spec();
 
-            let mut last_granule = 0;
-            let mut ticker = tokio::time::interval(page_duration);
+            let channels = if spec.channels == 1 {
+                audiopus::Channels::Mono
+            } else if spec.channels == 2 {
+                audiopus::Channels::Stereo
+            } else {
+                log::error!(
+                    "Only mono and stereo audio are supported, got {} channels",
+                    spec.channels
+                );
+                break;
+            };
 
-            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);
+            let mut opus_coder = OpusCoder::new(spec.sample_rate, channels)
+                .expect("Failed to initialize Opus coder");
 
-                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}");
+            let samples: Vec<f32> = reader
+                .samples::<i16>()
+                .map(|s| match s {
+                    Ok(sample) => sample as f32 / 32768.0,
+                    Err(e) => {
+                        log::warn!("Failed to read sample: {}", e);
+                        0.0
+                    }
+                })
+                .collect();
+
+            log::trace!(
+                "Loaded WAV: {}Hz, {} channels, {} samples, {:.2}s",
+                spec.sample_rate,
+                spec.channels,
+                samples.len(),
+                samples.len() as f32 / (spec.sample_rate as f32 * spec.channels as f32)
+            );
+
+            let samples_per_frame = opus_coder.input_samples_per_frame();
+            let frame_duration_ms = 20;
+            let frame_duration = Duration::from_millis(frame_duration_ms);
+            let mut ticker = tokio::time::interval(frame_duration);
+
+            for (frame_idx, chunk) in samples.chunks(samples_per_frame).enumerate() {
+                let mut frame = vec![0.0f32; samples_per_frame];
+                frame[..chunk.len()].copy_from_slice(chunk);
+
+                match opus_coder.encode(&frame) {
+                    Ok(opus_data) => {
+                        if let Err(e) = packet_sender.send(PacketData::Audio {
+                            timestamp: Instant::now(),
+                            duration: frame_duration,
+                            data: opus_data.into(),
+                        }) {
+                            log::warn!("send audio data failed: {e}");
+                        }
+                    }
+                    Err(e) => {
+                        log::warn!("Encoding frame {} failed: {}", frame_idx + 1, e);
+                        if let Err(e) = packet_sender.send(PacketData::Audio {
+                            timestamp: Instant::now(),
+                            duration: frame_duration,
+                            data: vec![].into(),
+                        }) {
+                            log::warn!("send empty audio data failed: {e}");
+                        }
+                    }
                 }
 
-                _ = ticker.tick().await;
+                ticker.tick().await;
 
                 if connections.load(Ordering::Relaxed) <= 0 {
                     break 'out;
@@ -197,6 +245,8 @@ fn ogg_stream_thread(
                 break;
             }
         }
+
+        log::info!("wav_streaming_thread exit...");
     });
 }
 

lib/wrtc/examples/opus_demo.rs

@@ -13,7 +13,8 @@ use tokio::sync::broadcast::{self, Sender};
 use webrtc::media::io::h264_reader::H264Reader;
 use webrtc::media::io::ogg_reader::OggReader;
 use wrtc::{
-    Event, OPUS_SAMPLE_RATE, PacketData, session::WebRTCServerSessionConfig, webrtc::WebRTCServer,
+    Event, PacketData, opus::OPUS_SAMPLE_RATE, session::WebRTCServerSessionConfig,
+    webrtc::WebRTCServer,
 };
 
 #[tokio::main]

lib/wrtc/examples/whep_server2_demo.rs

@@ -1,6 +1,5 @@
 use crate::scanf;
 use indexmap::IndexMap;
-use serde::Deserialize;
 
 #[derive(Debug, thiserror::Error)]
 pub enum AuthError {
@@ -14,16 +13,6 @@ pub enum AuthError {
     InvalidTokenFormat,
 }
 
-#[derive(Debug, Deserialize, Clone, Default)]
-pub enum AuthAlgorithm {
-    #[default]
-    #[serde(rename = "simple")]
-    Simple,
-
-    #[serde(rename = "md5")]
-    Md5,
-}
-
 pub enum SecretCarrier {
     Query(String),
     Bearer(String),
@@ -69,31 +58,21 @@ pub fn get_secret(carrier: &SecretCarrier) -> Result<String, AuthError> {
 
 #[derive(Debug, Clone)]
 pub struct Auth {
-    key: String,
-    algorithm: AuthAlgorithm,
-    password: String,
+    token: String,
 }
 
 impl Auth {
-    pub fn new(key: String, password: String, algorithm: AuthAlgorithm) -> Self {
-        Self {
-            key,
-            algorithm,
-            password,
-        }
+    pub fn new(token: String) -> Self {
+        Self { token }
     }
 
-    pub fn authenticate(
-        &self,
-        stream_name: &String,
-        secret: &Option<SecretCarrier>,
-    ) -> Result<(), AuthError> {
+    pub fn authenticate(&self, secret: &Option<SecretCarrier>) -> Result<(), AuthError> {
         let mut auth_err_reason: String = String::from("there is no token str found.");
         let mut err = AuthError::NoTokenFound;
 
         if let Some(secret_value) = secret {
             let token = get_secret(secret_value)?;
-            if self.check(stream_name, token.as_str()) {
+            if self.check(token.as_str()) {
                 return Ok(());
             }
             auth_err_reason = format!("token is not correct: {token}");
@@ -104,13 +83,7 @@ impl Auth {
         return Err(err);
     }
 
-    fn check(&self, stream_name: &String, auth_str: &str) -> bool {
-        match self.algorithm {
-            AuthAlgorithm::Simple => self.password == auth_str,
-            AuthAlgorithm::Md5 => {
-                let raw_data = format!("{}{}", self.key, stream_name);
-                auth_str == cutil::crypto::md5(&raw_data).to_lowercase()
-            }
-        }
+    fn check(&self, token: &str) -> bool {
+        self.token == token
     }
 }