From 65759699eb17906d89515c2e81169f520e55f3c1 Mon Sep 17 00:00:00 2001 From: Filip Kojro Date: Sun, 23 Aug 2026 18:27:35 +0200 Subject: [PATCH] adding opus encoder --- Cargo.lock | 52 +++++++++++++++++ client/Cargo.toml | 1 + client/src/main.rs | 143 +++++++++++++++++++++++++++++++-------------- 3 files changed, 152 insertions(+), 44 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 4640785..edb406e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -74,6 +74,17 @@ dependencies = [ "windows-sys", ] +[[package]] +name = "audiopus_sys" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "62314a1546a2064e033665d658e88c620a62904be945f8147e6b16c3db9f8651" +dependencies = [ + "cmake", + "log", + "pkg-config", +] + [[package]] name = "autocfg" version = "1.5.1" @@ -107,6 +118,16 @@ version = "1.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" +[[package]] +name = "cc" +version = "1.4.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ad534f4357a5264cce5019c989cf66a4f0dc4e0d1b1d15f8aacec0ff7360273" +dependencies = [ + "find-msvc-tools", + "shlex", +] + [[package]] name = "cfg-if" version = "1.0.4" @@ -170,6 +191,7 @@ version = "0.1.0" dependencies = [ "clap", "cpal", + "opus", "rand", "ringbuf", "serde", @@ -178,6 +200,15 @@ dependencies = [ "tokio", ] +[[package]] +name = "cmake" +version = "0.1.58" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0f78a02292a74a88ac736019ab962ece0bc380e3f977bf72e376c5d78ff0678" +dependencies = [ + "cc", +] + [[package]] name = "colorchoice" version = "1.0.5" @@ -285,6 +316,12 @@ dependencies = [ "windows-sys", ] +[[package]] +name = "find-msvc-tools" +version = "0.1.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d45db016d36b838f563236e9193d0ee6ce38f3f68b6c94e914b4929c96bbb890" + [[package]] name = "futures-core" version = "0.3.34" @@ -641,6 +678,15 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" +[[package]] +name = "opus" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4d3809943dff6fbad5f0484449ea26bdb9cb7d8efdf26ed50d3c7f227f69eb5c" +dependencies = [ + "audiopus_sys", +] + [[package]] name = "parking_lot" version = "0.12.5" @@ -859,6 +905,12 @@ dependencies = [ "serde", ] +[[package]] +name = "shlex" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" + [[package]] name = "signal-hook-registry" version = "1.4.8" diff --git a/client/Cargo.toml b/client/Cargo.toml index 55608a6..1e5223c 100644 --- a/client/Cargo.toml +++ b/client/Cargo.toml @@ -12,3 +12,4 @@ serde_json = "1.0.151" tokio = { version = "1.53.1", features = ["full"] } rand = "0.10.2" clap = { version = "4.6.6", features = ["derive"] } +opus = "0.3.1" diff --git a/client/src/main.rs b/client/src/main.rs index 24ef13c..424a8c4 100644 --- a/client/src/main.rs +++ b/client/src/main.rs @@ -3,6 +3,7 @@ use cpal::{ Device, traits::{DeviceTrait, HostTrait, StreamTrait}, }; +use opus::{Decoder, Encoder}; use ringbuf::{ HeapRb, consumer, traits::{Consumer, Observer, Producer, Split}, @@ -40,6 +41,8 @@ type InputAudioStreamsProducers = type InputAudioStreamsConsumers = Arc>, false, true>>>>; +type InputAudioStreamsEncoders = Arc>>>; + type InputAudioStreamsInfo = Arc>>; struct ClientInfo { @@ -64,11 +67,16 @@ async fn main() { Arc::new(Mutex::new(HashMap::new())); let active_input_audio_streams_consumers: InputAudioStreamsConsumers = Arc::new(Mutex::new(HashMap::new())); + + let active_input_audio_streams_encoders: InputAudioStreamsEncoders = + Arc::new(RwLock::new(HashMap::new())); + let active_input_audio_streams_info: InputAudioStreamsInfo = Arc::new(RwLock::new(HashMap::new())); let active_input_audio_streams_producers_tcp = active_input_audio_streams_producers.clone(); let active_input_audio_streams_consumers_tcp = active_input_audio_streams_consumers.clone(); + let active_input_audio_streams_encoders_tcp = active_input_audio_streams_encoders.clone(); let active_input_audio_streams_info_tcp = active_input_audio_streams_info.clone(); let server_address = { @@ -125,6 +133,9 @@ async fn main() { let mut consumer_map = active_input_audio_streams_consumers_tcp.lock().unwrap(); consumer_map.clear(); + let mut encoder_map = active_input_audio_streams_encoders_tcp.write().unwrap(); + encoder_map.clear(); + let mut info_map = active_input_audio_streams_info_tcp.write().unwrap(); info_map.clear(); @@ -145,6 +156,21 @@ async fn main() { (audio_stream.client_id, audio_stream.stream_id), consumer, ); + + let channels: opus::Channels = { + if audio_stream.channels == 1 { + opus::Channels::Mono + } else { + opus::Channels::Stereo + } + }; + + let decoder = Decoder::new(audio_stream.sample_rate, channels).expect("failed to creat decoder"); + encoder_map.insert( + (audio_stream.client_id, audio_stream.stream_id), + Mutex::new(decoder), + ); + info_map.insert( (audio_stream.client_id, audio_stream.stream_id), audio_stream, @@ -263,10 +289,14 @@ async fn main() { let active_input_audio_streams_producers_udp = active_input_audio_streams_producers.clone(); let active_input_audio_streams_info_udp = active_input_audio_streams_info.clone(); let speaker_config_clone = speaker_config.clone(); + let active_input_audio_streams_encoders_clone = active_input_audio_streams_encoders.clone(); tokio::spawn(async move { let mut receive_buffer = vec![0u8; 4096]; + let mut decoded_samples = [0.0f32; 1024]; + let mut decoded_samples_converted = [0.0f32; 1024]; + loop { match input_socket.recv_from(&mut receive_buffer).await { Ok((len, _)) => { @@ -282,55 +312,65 @@ async fn main() { let audio_packet = &receive_buffer[5..len]; - let mut decoded_samples = Vec::with_capacity(audio_packet.len() / 4); + let stream_info = { + if let Some(&info) = active_input_audio_streams_info_udp + .read() + .unwrap() + .get(&(client_id, stream_id)) + { + info + } else { + continue; + } + }; - for chunk in audio_packet.chunks_exact(4) { - let bytes: [u8; 4] = chunk.try_into().unwrap(); + let decoders_map = active_input_audio_streams_encoders_clone.read().unwrap(); - decoded_samples.push(f32::from_le_bytes(bytes)); - } + let decoder_mutex = decoders_map + .get(&(client_id, stream_id)) + .expect("failed to get encoder mutex"); + + let mut decoder = decoder_mutex.lock().expect("failed to get encoder"); + + let decoded_len = decoder + .decode_float(audio_packet, &mut decoded_samples, false) + .expect("failed to decode opus"); let mut active_streams_map = active_input_audio_streams_producers_udp.lock().unwrap(); if let Some(producer) = active_streams_map.get_mut(&(client_id, stream_id)) { - if let Some(&stream_info) = active_input_audio_streams_info_udp - .read() - .unwrap() - .get(&(client_id, stream_id)) - { - if stream_info.channels == speaker_config_clone.channels { - producer.push_slice(&decoded_samples); - } else { - let n = stream_info.channels; - let mut step = 0; - decoded_samples.retain(|_| { - if step == 0 { - step = n - 1; - true - } else { - step -= 1; - false - } - }); - if speaker_config_clone.channels == 1 { - producer.push_slice(&decoded_samples); - } else { - let mut out = vec![ - 0f32; - decoded_samples.len() - * speaker_config_clone.channels as usize - ]; - for (item, chunk) in - decoded_samples.iter().zip(out.chunks_exact_mut( - speaker_config_clone.channels as usize, - )) - { - chunk.fill(*item); - } - producer.push_slice(&out); - } + if stream_info.channels == speaker_config_clone.channels { + producer.push_slice( + &decoded_samples[..decoded_len * stream_info.channels as usize], + ); + } else if stream_info.channels > speaker_config_clone.channels { + for (i, frame) in decoded_samples + [..decoded_len * stream_info.channels as usize] + .chunks_exact(2) + .enumerate() + { + decoded_samples_converted[i] = (frame[0] + frame[1]) / 2.0; } + + producer.push_slice( + &decoded_samples_converted + [..decoded_len * stream_info.channels as usize], + ); + } else { + for (i, frame) in decoded_samples_converted + [..decoded_len * stream_info.channels as usize] + .chunks_exact_mut(2) + .enumerate() + { + frame[0] = decoded_samples[i]; + frame[1] = decoded_samples[i]; + } + + producer.push_slice( + &decoded_samples_converted + [..decoded_len * stream_info.channels as usize], + ); } } @@ -437,6 +477,19 @@ async fn audio_stream_spawner( let frames_needed = ((20.0 / 1000.0) * config.sample_rate as f32) as usize; let mut frame_buffer = vec![0.0; frames_needed]; + let channels: opus::Channels = { + if config.channels == 1 { + opus::Channels::Mono + } else { + opus::Channels::Stereo + } + }; + + let mut opus_encoder = Encoder::new(config.sample_rate, channels, opus::Application::Voip) + .expect("falied to create opus encoder"); + + let mut encoded_frame_buffer = [0u8; 1024]; + loop { if consumer.occupied_len() >= frames_needed { consumer.pop_slice(&mut frame_buffer); @@ -451,9 +504,11 @@ async fn audio_stream_spawner( payload.extend(client_id.to_be_bytes()); payload.extend(current_stream_id.to_be_bytes()); - for sample in &frame_buffer { - payload.extend_from_slice(&sample.to_le_bytes()); - } + let encoded_len = opus_encoder + .encode_float(&frame_buffer.as_slice(), &mut encoded_frame_buffer) + .expect("failed to encode frame"); + + payload.extend_from_slice(&encoded_frame_buffer[..encoded_len]); let _ = udp_socket.send_to(&payload, server_address).await; } else {