connecting to server and sending/receiving raw mic input

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"

Cargo.toml

@@ -6,3 +6,4 @@ edition = "2024"
 [dependencies]
 cpal = "0.18.1"
 ringbuf = "0.5.1"
+tokio = { version = "1.53.1", features = ["full"] }

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::<f32>::new(latency_samples * 2);
+
-    let (mut prod, mut cons) = rb.split();
+    let ring_buffer_size = latency_samples * 2;
+
+    let microphone_ring_buffer = HeapRb::<f32>::new(ring_buffer_size);
+    let (mut microphone_producer, mut microphone_consumer) = microphone_ring_buffer.split();
+
+    let speaker_ring_buffer = HeapRb::<f32>::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() {
+            if microphone_producer.try_push(sample).is_err() {}
-                output_behind = true;
-            }
-        }
-        if output_behind {
-            eprintln!("output behind");
         }
     };
 
     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 => {
+                None => 0.0,
-                    input_behind = true;
-                    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");
 }