diff --git a/Cargo.lock b/Cargo.lock index d8ed646..6adfcf7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -145,6 +145,16 @@ version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" +[[package]] +name = "errno" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" +dependencies = [ + "libc", + "windows-sys", +] + [[package]] name = "futures-core" version = "0.3.34" @@ -260,6 +270,15 @@ version = "0.2.189" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" +[[package]] +name = "lock_api" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "224399e74b87b5f3557511d98dff8b14089b3dadafcab6bb93eab67d3aace965" +dependencies = [ + "scopeguard", +] + [[package]] name = "log" version = "0.4.33" @@ -278,6 +297,17 @@ version = "2.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" +[[package]] +name = "mio" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "30d65c71f1ce40ab09135ce117d742b9f8a19ff91a41a8b57ed50bc2de59c427" +dependencies = [ + "libc", + "wasi", + "windows-sys", +] + [[package]] name = "ndk" version = "0.9.0" @@ -445,6 +475,29 @@ version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" +[[package]] +name = "parking_lot" +version = "0.12.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93857453250e3077bd71ff98b6a65ea6621a19bb0f559a85248955ac12c45a1a" +dependencies = [ + "lock_api", + "parking_lot_core", +] + +[[package]] +name = "parking_lot_core" +version = "0.9.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1" +dependencies = [ + "cfg-if", + "libc", + "redox_syscall", + "smallvec", + "windows-link", +] + [[package]] name = "pin-project-lite" version = "0.2.17" @@ -499,6 +552,15 @@ dependencies = [ "proc-macro2", ] +[[package]] +name = "redox_syscall" +version = "0.5.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" +dependencies = [ + "bitflags", +] + [[package]] name = "ringbuf" version = "0.5.1" @@ -534,6 +596,12 @@ dependencies = [ "winapi-util", ] +[[package]] +name = "scopeguard" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" + [[package]] name = "semver" version = "1.0.28" @@ -560,6 +628,16 @@ dependencies = [ "syn 3.0.3", ] +[[package]] +name = "signal-hook-registry" +version = "1.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b" +dependencies = [ + "errno", + "libc", +] + [[package]] name = "simd_cesu8" version = "1.2.0" @@ -588,6 +666,23 @@ version = "0.1.0" dependencies = [ "cpal", "ringbuf", + "tokio", +] + +[[package]] +name = "smallvec" +version = "1.15.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" + +[[package]] +name = "socket2" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" +dependencies = [ + "libc", + "windows-sys", ] [[package]] @@ -652,6 +747,34 @@ dependencies = [ "syn 3.0.3", ] +[[package]] +name = "tokio" +version = "1.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed" +dependencies = [ + "bytes", + "libc", + "mio", + "parking_lot", + "pin-project-lite", + "signal-hook-registry", + "socket2", + "tokio-macros", + "windows-sys", +] + +[[package]] +name = "tokio-macros" +version = "2.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + [[package]] name = "toml_datetime" version = "1.1.1+spec-1.1.0" @@ -698,6 +821,12 @@ dependencies = [ "winapi-util", ] +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + [[package]] name = "wasm-bindgen" version = "0.2.127" diff --git a/Cargo.toml b/Cargo.toml index 111bbbf..08ff223 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -6,3 +6,4 @@ edition = "2024" [dependencies] cpal = "0.18.1" ringbuf = "0.5.1" +tokio = { version = "1.53.1", features = ["full"] } diff --git a/src/main.rs b/src/main.rs index a081749..51c9824 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,11 +1,20 @@ use cpal::traits::{DeviceTrait, HostTrait, StreamTrait}; use ringbuf::{ HeapRb, - traits::{Consumer, Producer, Split}, + traits::{Consumer, Observer, Producer, Split}, }; -use std::sync::mpsc; +use std::sync::Arc; +use tokio::net::UdpSocket; + +#[tokio::main] +async fn main() { + let server_address = "127.0.0.1:8080"; + + let socket = Arc::new(UdpSocket::bind("0.0.0.0:0").await.unwrap()); + + let output_socket = socket.clone(); + let input_socket = socket.clone(); -fn main() { let host = cpal::default_host(); let input_device = host @@ -28,39 +37,33 @@ fn main() { let latency_ms = 150.0; let latency_frames = (latency_ms / 1000.0) * config.sample_rate as f32; let latency_samples = latency_frames as usize * config.channels as usize; - let rb = HeapRb::::new(latency_samples * 2); - let (mut prod, mut cons) = rb.split(); + + let ring_buffer_size = latency_samples * 2; + + let microphone_ring_buffer = HeapRb::::new(ring_buffer_size); + let (mut microphone_producer, mut microphone_consumer) = microphone_ring_buffer.split(); + + let speaker_ring_buffer = HeapRb::::new(ring_buffer_size); + let (mut speaker_producer, mut speaker_consumer) = speaker_ring_buffer.split(); for _ in 0..latency_samples { - prod.try_push(0.0).unwrap(); + microphone_producer.try_push(0.0).unwrap(); + speaker_producer.try_push(0.0).unwrap(); } let input_data_fn = move |data: &[f32], _: &cpal::InputCallbackInfo| { - let mut output_behind = false; for &sample in data { - if prod.try_push(sample).is_err() { - output_behind = true; - } - } - if output_behind { - eprintln!("output behind"); + if microphone_producer.try_push(sample).is_err() {} } }; let output_data_fn = move |data: &mut [f32], _: &cpal::OutputCallbackInfo| { - let mut input_behind = false; for sample in data { - *sample = match cons.try_pop() { + *sample = match speaker_consumer.try_pop() { Some(s) => s, - None => { - input_behind = true; - 0.0 - } + None => 0.0, }; } - if input_behind { - eprintln!("input behind") - } }; let err_fn = |err| eprintln!("stream error: {}", err); @@ -75,5 +78,49 @@ fn main() { input_stream.play().expect("failed to play input stream"); output_stream.play().expect("failed to play output stream"); - std::thread::sleep(std::time::Duration::from_secs(5)); + tokio::spawn(async move { + let frames_needed = ((20.0 / 1000.0) * config.sample_rate as f32) as usize; + let mut frame_buffer = vec![0.0; frames_needed]; + + loop { + if microphone_consumer.occupied_len() >= frames_needed { + microphone_consumer.pop_slice(&mut frame_buffer); + + let mut payload = Vec::with_capacity(frames_needed * 4); + for sample in &frame_buffer { + payload.extend_from_slice(&sample.to_le_bytes()); + } + + let _ = output_socket.send_to(&payload, server_address).await; + } else { + tokio::time::sleep(std::time::Duration::from_millis(2)).await; + } + } + }); + + tokio::spawn(async move { + let mut receive_buffer = vec![0u8; 4096]; + + loop { + match input_socket.recv_from(&mut receive_buffer).await { + Ok((len, _)) => { + let audio_packet = &receive_buffer[..len]; + + let mut decoded_samples = Vec::with_capacity(audio_packet.len() / 4); + + for chunk in audio_packet.chunks_exact(4) { + let bytes: [u8; 4] = chunk.try_into().unwrap(); + + decoded_samples.push(f32::from_le_bytes(bytes)); + } + + speaker_producer.push_slice(&decoded_samples); + } + Err(e) => eprintln!("failed to receive udp packet: {}", e), + } + } + }); + + tokio::signal::ctrl_c().await.expect("failed to close"); + println!("shutting down"); }