diff --git a/Cargo.lock b/Cargo.lock index 6adfcf7..2ddfaca 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -195,6 +195,12 @@ dependencies = [ "hashbrown", ] +[[package]] +name = "itoa" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" + [[package]] name = "jni" version = "0.22.4" @@ -608,6 +614,16 @@ version = "1.0.28" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8a7852d02fc848982e0c167ef163aaff9cd91dc640ba85e263cb1ce46fae51cd" +[[package]] +name = "serde" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba" +dependencies = [ + "serde_core", + "serde_derive", +] + [[package]] name = "serde_core" version = "1.0.229" @@ -628,6 +644,19 @@ dependencies = [ "syn 3.0.3", ] +[[package]] +name = "serde_json" +version = "1.0.151" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + [[package]] name = "signal-hook-registry" version = "1.4.8" @@ -666,6 +695,8 @@ version = "0.1.0" dependencies = [ "cpal", "ringbuf", + "serde", + "serde_json", "tokio", ] @@ -1018,3 +1049,9 @@ checksum = "23b97319f7b8343df12cc98938e5c3eb436064524c8d2b4e30a1d3a36eecdf81" dependencies = [ "memchr", ] + +[[package]] +name = "zmij" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" diff --git a/Cargo.toml b/Cargo.toml index 08ff223..bfa0c9e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -6,4 +6,6 @@ edition = "2024" [dependencies] cpal = "0.18.1" ringbuf = "0.5.1" +serde = { version = "1.0.229", features = ["derive"] } +serde_json = "1.0.151" tokio = { version = "1.53.1", features = ["full"] } diff --git a/src/main.rs b/src/main.rs index 51c9824..17c143e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -3,13 +3,53 @@ use ringbuf::{ HeapRb, traits::{Consumer, Observer, Producer, Split}, }; +use serde::{Deserialize, Serialize}; use std::sync::Arc; -use tokio::net::UdpSocket; +use tokio::net::{TcpStream, UdpSocket}; + +#[derive(Serialize, Deserialize, PartialEq)] +enum ResponseStatus { + Ok, + Err, +} + +#[derive(Serialize, Deserialize)] +struct ConectionResponse { + status: ResponseStatus, + client_id: u32, +} #[tokio::main] async fn main() { let server_address = "127.0.0.1:8080"; + let tcp_stream = TcpStream::connect(server_address) + .await + .expect("failed to connect using tcp"); + + tcp_stream.readable().await.expect("idk"); + let mut tcp_buf = [0; 4096]; + + let mut session_client_id: u32 = 0; + + match tcp_stream.try_read(&mut tcp_buf) { + Ok(n) => { + // let connection_response = str::from_utf8(&tcp_buf[0..n]).expect("failed to create str"); + + let connection_response: ConectionResponse = + serde_json::from_slice(&tcp_buf[0..n]).unwrap(); + + session_client_id = connection_response.client_id; + + println!( + "tcp response status: {}, clientId: {}", + connection_response.status == ResponseStatus::Ok, + connection_response.client_id + ); + } + Err(e) => eprintln!("failed to receive tcp packet: {}", e), + } + let socket = Arc::new(UdpSocket::bind("0.0.0.0:0").await.unwrap()); let output_socket = socket.clone(); @@ -34,11 +74,11 @@ async fn main() { println!("stream config: {:?}", config); - 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 ring_buffer_ms = 150.0; + let ring_buffer_frames = (ring_buffer_ms / 1000.0) * config.sample_rate as f32; + let ring_buffer_samples = ring_buffer_frames as usize * config.channels as usize; - let ring_buffer_size = latency_samples * 2; + let ring_buffer_size = ring_buffer_samples; let microphone_ring_buffer = HeapRb::::new(ring_buffer_size); let (mut microphone_producer, mut microphone_consumer) = microphone_ring_buffer.split(); @@ -46,11 +86,6 @@ async fn main() { 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 { - microphone_producer.try_push(0.0).unwrap(); - speaker_producer.try_push(0.0).unwrap(); - } - let input_data_fn = move |data: &[f32], _: &cpal::InputCallbackInfo| { for &sample in data { if microphone_producer.try_push(sample).is_err() {} @@ -87,6 +122,9 @@ async fn main() { microphone_consumer.pop_slice(&mut frame_buffer); let mut payload = Vec::with_capacity(frames_needed * 4); + + payload.extend(session_client_id.to_be_bytes()); + for sample in &frame_buffer { payload.extend_from_slice(&sample.to_le_bytes()); } @@ -104,7 +142,14 @@ async fn main() { loop { match input_socket.recv_from(&mut receive_buffer).await { Ok((len, _)) => { - let audio_packet = &receive_buffer[..len]; + if len < 4 { + continue; + } + + let id_bytes: [u8; 4] = receive_buffer[0..4].try_into().unwrap(); + let client_id = u32::from_be_bytes(id_bytes); + + let audio_packet = &receive_buffer[4..len]; let mut decoded_samples = Vec::with_capacity(audio_packet.len() / 4);