diff --git a/Cargo.lock b/Cargo.lock index ff05fc7..94a57d8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -283,6 +283,7 @@ dependencies = [ "clap", "cpal", "crossterm", + "futures", "gag", "rand 0.10.2", "ratatui", @@ -292,6 +293,7 @@ dependencies = [ "serde_json", "shared", "tokio", + "tokio-util", ] [[package]] @@ -645,12 +647,71 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb" +[[package]] +name = "futures" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a31d2a3fbaaeb2af2368bbdd904aa8e812d3c04a1ee10d3171f52d556e5d0a3" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-channel" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b1f9e3d69d39e4862ffed03ed071a76f9a13ba1d9109d355b0f0aa6b15e393c4" +dependencies = [ + "futures-core", + "futures-sink", +] + [[package]] name = "futures-core" version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" +[[package]] +name = "futures-executor" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "031b47cf1a3c6cc8bc2fc76cd437f521619387907d469316e7c0bc278f1f5432" +dependencies = [ + "futures-core", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-io" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53c0fa8157de1303bfffdaa1cc2a673bfffb60102f76b0ef4441659124373fed" + +[[package]] +name = "futures-macro" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9fb9654ba8355388abeb8dcb4fc62f511300867002afc858860463bdd9fe0c44" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + +[[package]] +name = "futures-sink" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1944426bf7d03f1d14f708785e4b33efd750b36d48a157b836b3efc15ede8e1d" + [[package]] name = "futures-task" version = "0.3.34" @@ -663,8 +724,13 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc" dependencies = [ + "futures-channel", "futures-core", + "futures-io", + "futures-macro", + "futures-sink", "futures-task", + "memchr", "pin-project-lite", "slab", ] @@ -1756,11 +1822,13 @@ name = "server" version = "0.1.0" dependencies = [ "clap", + "futures", "rand 0.10.2", "serde", "serde_json", "shared", "tokio", + "tokio-util", ] [[package]] @@ -2106,6 +2174,20 @@ dependencies = [ "syn 3.0.3", ] +[[package]] +name = "tokio-util" +version = "0.7.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "494815d09bf52b5548659851081238f0ca39ff638363907596da739561c62c52" +dependencies = [ + "bytes", + "futures-core", + "futures-sink", + "libc", + "pin-project-lite", + "tokio", +] + [[package]] name = "toml_datetime" version = "1.1.1+spec-1.1.0" diff --git a/client/Cargo.toml b/client/Cargo.toml index 6cbd3d6..f053389 100644 --- a/client/Cargo.toml +++ b/client/Cargo.toml @@ -16,3 +16,5 @@ ropus = "0.12.18" ratatui = "0.30.2" crossterm = "0.29.0" gag = "1.0.0" +tokio-util = { version = "0.7.19", features = ["codec"]} +futures = "0.3.34" diff --git a/client/src/tcp_connection.rs b/client/src/tcp_connection.rs index 5d6f104..159dd72 100644 --- a/client/src/tcp_connection.rs +++ b/client/src/tcp_connection.rs @@ -1,12 +1,16 @@ +use futures::{SinkExt, StreamExt}; use ringbuf::{HeapRb, traits::Split}; use ropus::Decoder; use shared::{ResponseStatus, TcpRequest, TcpResponse, TcpResponseData}; use std::sync::{Arc, Mutex, RwLock}; use tokio::{ - io::{AsyncReadExt, AsyncWriteExt}, net::TcpStream, sync::broadcast::{Receiver, Sender}, }; +use tokio_util::{ + bytes::Bytes, + codec::{Framed, LengthDelimitedCodec}, +}; use crate::models::{AudioControl, ClientEvent, ClientInfo}; @@ -21,99 +25,103 @@ pub async fn tcp_connection( info.server_address.clone() }; - let mut tcp_stream = TcpStream::connect(server_address) + let tcp_stream = TcpStream::connect(server_address) .await .expect("failed to connect using tcp"); - let mut tcp_buf = [0u8; 4096]; + let mut framed_tcp_stream = Framed::new(tcp_stream, LengthDelimitedCodec::new()); + loop { tokio::select! { - socket_result = tcp_stream.read(&mut tcp_buf) => { - - match socket_result { - Ok(0) => { + frame_result = framed_tcp_stream.next() => { + match frame_result { + None => { println!("server shut down"); break; } - Ok(n) => { - let ssss = String::from_utf8(tcp_buf[0..n].to_vec()).unwrap(); - println!("json data: {}", ssss); + Some(Ok(bytes)) => { + match serde_json::from_slice::(&bytes) { + Ok(response) => { + println!("connection Ok: {}", response.status == ResponseStatus::Ok); + match response.data { + TcpResponseData::NewConnection { client_id } => { + let mut client_info_var = client_info.write().unwrap(); + client_info_var.client_id = client_id; + client_info_var.connected = true; + println!("fetched client id: {}", client_id); + } + TcpResponseData::UserConnected { user_id } => { + println!("new user connected: {}", user_id); + } + TcpResponseData::UserDisconnected { user_id } => { + println!("user disconnected: {}", user_id); + } + TcpResponseData::AudioStreamsFullUpdate { audio_stream_info } => { + let mut producer_map = audio_control.active_input_audio_streams_producers.lock().unwrap(); + producer_map.clear(); + let mut consumer_map = audio_control.active_input_audio_streams_consumers.lock().unwrap(); + consumer_map.clear(); - match serde_json::from_slice::(&tcp_buf[0..n]) { - Ok(response) => { - println!("connection Ok: {}", response.status == ResponseStatus::Ok); - match response.data { - TcpResponseData::NewConnection { client_id } => { - let mut client_info_var = client_info.write().unwrap(); - client_info_var.client_id = client_id; - client_info_var.connected = true; - println!("fetched client id: {}", client_id); - } - TcpResponseData::UserConnected { user_id } => { - println!("new user connected: {}", user_id); - } - TcpResponseData::UserDisconnected { user_id } => { - println!("user disconnected: {}", user_id); - } - TcpResponseData::AudioStreamsFullUpdate { audio_stream_info } => { - let mut producer_map = audio_control.active_input_audio_streams_producers.lock().unwrap(); - producer_map.clear(); - let mut consumer_map = audio_control.active_input_audio_streams_consumers.lock().unwrap(); - consumer_map.clear(); + let mut encoder_map = audio_control.active_input_audio_streams_encoders.write().unwrap(); + encoder_map.clear(); - let mut encoder_map = audio_control.active_input_audio_streams_encoders.write().unwrap(); - encoder_map.clear(); + let mut info_map = audio_control.active_input_audio_streams_info.write().unwrap(); + info_map.clear(); - let mut info_map = audio_control.active_input_audio_streams_info.write().unwrap(); - info_map.clear(); + for audio_stream in audio_stream_info { + let ring_buffer_ms = 150.0; + let ring_buffer_frames = (ring_buffer_ms / 1000.0) * audio_stream.sample_rate as f32; + let ring_buffer_samples = ring_buffer_frames as usize * audio_stream.channels as usize; + let ring_buffer_size = ring_buffer_samples; + let ring_buffer = HeapRb::::new(ring_buffer_size); + let (producer, consumer) = ring_buffer.split(); - for audio_stream in audio_stream_info { - let ring_buffer_ms = 150.0; - let ring_buffer_frames = (ring_buffer_ms / 1000.0) * audio_stream.sample_rate as f32; - let ring_buffer_samples = ring_buffer_frames as usize * audio_stream.channels as usize; - let ring_buffer_size = ring_buffer_samples; - let ring_buffer = HeapRb::::new(ring_buffer_size); - let (producer, consumer) = ring_buffer.split(); + producer_map.insert( + (audio_stream.client_id, audio_stream.stream_id), + producer, + ); + consumer_map.insert( + (audio_stream.client_id, audio_stream.stream_id), + consumer, + ); - producer_map.insert( - (audio_stream.client_id, audio_stream.stream_id), - producer, - ); - consumer_map.insert( - (audio_stream.client_id, audio_stream.stream_id), - consumer, - ); + let channels: ropus::Channels = { + if audio_stream.channels == 1 { + ropus::Channels::Mono + } else { + ropus::Channels::Stereo + } + }; - let channels: ropus::Channels = { - if audio_stream.channels == 1 { - ropus::Channels::Mono - } else { - ropus::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), + ); - 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, - ); + info_map.insert( + (audio_stream.client_id, audio_stream.stream_id), + audio_stream, + ); + } + println!("audio streams update"); + } + TcpResponseData::NewAudioStreamAccepted { audio_stream_id } => { + let _ = global_event_tx.send(ClientEvent::NewAudioStreamAccepted(audio_stream_id)); } - println!("audio streams update"); - } - TcpResponseData::NewAudioStreamAccepted { audio_stream_id } => { - let _ = global_event_tx.send(ClientEvent::NewAudioStreamAccepted(audio_stream_id)); } } + Err(e) => { + eprintln!("parsing error: {}", e); + break; + } } - Err(e) => eprintln!("failed to parse tcp packet: {}", e),} - }, - Err(e) => eprintln!("failed to receive tcp packet: {}", e), + } + Some(Err(e)) => { + eprintln!("socket read error: {}", e); + break; + } } } event_result = global_event_rx.recv() => { @@ -121,34 +129,31 @@ pub async fn tcp_connection( Ok(event) => { match event { ClientEvent::NewConnection => { - let _ = tcp_stream - .write_all( - serde_json::to_string(&TcpRequest { - client_id: 0, - data: shared::TcpRequestData::NewConnection, - }) - .expect("failed to parse initial request") - .as_bytes(), - ) - .await - .expect("failed to send initial message"); + let request_json = serde_json::to_string(&TcpRequest { + client_id: 0, + data: shared::TcpRequestData::NewConnection, + }) + .expect("failed to parse initial request"); + + let _ = framed_tcp_stream.send(Bytes::from(request_json)) + .await + .expect("failed to send initial message"); + } ClientEvent::NewAudioStreamRequest(audio_stream_info) => { println!("new audio stream request"); let session_client_id = client_info.read().unwrap().client_id; - let _ = tcp_stream - .write_all( - serde_json::to_string(&TcpRequest { - client_id: session_client_id, - data: shared::TcpRequestData::NewAudioStream { - audio_stream_info - }, - }) - .expect("failed to parse initial request") - .as_bytes(), - ) - .await - .expect("failed to send initial message"); + + let request_json = serde_json::to_string(&TcpRequest { + client_id: session_client_id, + data: shared::TcpRequestData::NewAudioStream { + audio_stream_info + } + }).expect("failed to parse initial request"); + + let _ = framed_tcp_stream.send(Bytes::from(request_json)) + .await + .expect("failed to send initial message"); } _ => {} } diff --git a/server/Cargo.toml b/server/Cargo.toml index c478328..d138e57 100644 --- a/server/Cargo.toml +++ b/server/Cargo.toml @@ -10,3 +10,5 @@ serde_json = "1.0.151" tokio = { version = "1.53.1", features = ["full"] } rand = "0.10.2" clap = { version = "4.6.6", features = ["derive"] } +tokio-util = { version = "0.7.19", features = ["codec"]} +futures = "0.3.34" diff --git a/server/src/main.rs b/server/src/main.rs index a151ef3..07ccd54 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -1,12 +1,14 @@ use clap::Parser; +use futures::{SinkExt, StreamExt}; use shared::{AudioStreamInfo, ResponseStatus, TcpRequest, TcpRequestData, TcpResponse}; use std::sync::Arc; use std::{collections::HashMap, net::SocketAddr}; -use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::{ net::{TcpListener, UdpSocket}, sync::{RwLock, broadcast}, }; +use tokio_util::bytes::Bytes; +use tokio_util::codec::{Framed, LengthDelimitedCodec}; #[derive(Parser)] #[command(version, about)] @@ -93,8 +95,9 @@ async fn main() { }); loop { - let (mut tcp_stream, tcp_address) = - tcp_listener.accept().await.expect("failed to accept tcp"); + let (tcp_stream, tcp_address) = tcp_listener.accept().await.expect("failed to accept tcp"); + + let mut framed_tcp_stream = Framed::new(tcp_stream, LengthDelimitedCodec::new()); let tcp_state = state.clone(); @@ -103,88 +106,81 @@ async fn main() { tokio::spawn(async move { let current_client_id: u32 = rand::random(); - let mut tcp_buf = [0; 4096]; loop { tokio::select! { - socket_result = tcp_stream.read(&mut tcp_buf) => { - match socket_result { - Ok(0) => { + frame_result = framed_tcp_stream.next() => { + match frame_result { + None => { let _ = global_event_tx_clone.send(ServerEvent::UserDisconnected(current_client_id)); println!("client disconnected: {}", tcp_address); break; } - Ok(n) => match serde_json::from_slice::(&tcp_buf[0..n]) { - Ok(request) => match request.data { - TcpRequestData::NewConnection => { - tcp_state.write().await.insert( - current_client_id, - Client { - udp_addr: None, - audio_streams: None, - }, - ); + Some(Ok(bytes)) => { + match serde_json::from_slice::(&bytes) { + Ok(request) => match request.data { + TcpRequestData::NewConnection => { + tcp_state.write().await.insert( + current_client_id, + Client { + udp_addr: None, + audio_streams: None, + }, + ); - println!("new client: {}, from: {}", current_client_id, tcp_address); - let _ = tcp_stream - .write_all( - serde_json::to_string(&TcpResponse { - status: ResponseStatus::Ok, - data: shared::TcpResponseData::NewConnection { - client_id: current_client_id, - }, - }) - .expect("failed to parse response json TcpRequestData::NewConnection") - .as_bytes(), - ) - .await; + println!("new client: {}, from: {}", current_client_id, tcp_address); - let _ = global_event_tx_clone.send(ServerEvent::UserConnected(current_client_id)); - } - TcpRequestData::NewAudioStream { audio_stream_info } => { - if let Some(client) = tcp_state.write().await.get_mut(¤t_client_id) { - if let Some(audio_streams) = &mut client.audio_streams { - audio_streams.push(audio_stream_info); - } else { - client.audio_streams = Some(vec![audio_stream_info]); - } + let response_json = serde_json::to_string(&TcpResponse { + status: ResponseStatus::Ok, + data: shared::TcpResponseData::NewConnection { + client_id: current_client_id, + }, + }).expect("failed to serialize TcpRequestData::NewAudioStreamAccepted"); + + let _ = framed_tcp_stream.send(Bytes::from(response_json)) + .await + .expect("failed to send message"); + + let _ = global_event_tx_clone.send(ServerEvent::UserConnected(current_client_id)); } - let _ = tcp_stream - .write_all( - serde_json::to_string(&TcpResponse { - status: ResponseStatus::Ok, - data: shared::TcpResponseData::NewAudioStreamAccepted { - audio_stream_id: audio_stream_info.stream_id, - }, - }) - .expect("failed to parse response json TcpRequestData::NewConnection") - .as_bytes(), - ) - .await; + TcpRequestData::NewAudioStream { audio_stream_info } => { + if let Some(client) = tcp_state.write().await.get_mut(¤t_client_id) { + if let Some(audio_streams) = &mut client.audio_streams { + audio_streams.push(audio_stream_info.clone()); + } else { + client.audio_streams = Some(vec![audio_stream_info.clone()]); + } + } - let _ = global_event_tx_clone.send(ServerEvent::SendAllOtherStreams(current_client_id)); - } - TcpRequestData::DropAudioStream { audio_stream_id } => { - if let Some(client) = tcp_state.write().await.get_mut(¤t_client_id) { - if let Some(audio_streams) = &mut client.audio_streams { - let mut maybe_idx: Option = None; - for (i, audio_stream) in audio_streams.iter_mut().enumerate() { - if audio_stream.stream_id == audio_stream_id { - maybe_idx = Some(i); + let response_json = serde_json::to_string(&TcpResponse { + status: ResponseStatus::Ok, + data: shared::TcpResponseData::NewAudioStreamAccepted { + audio_stream_id: audio_stream_info.stream_id, + }, + }).expect("failed to serialize TcpRequestData::NewAudioStreamAccepted"); + + let _ = framed_tcp_stream.send(Bytes::from(response_json)).await; + + let _ = global_event_tx_clone.send(ServerEvent::SendAllOtherStreams(current_client_id)); + } + TcpRequestData::DropAudioStream { audio_stream_id } => { + if let Some(client) = tcp_state.write().await.get_mut(¤t_client_id) { + if let Some(audio_streams) = &mut client.audio_streams { + if let Some(idx) = audio_streams.iter().position(|s| s.stream_id == audio_stream_id) { + audio_streams.remove(idx); } } - if let Some(idx) = maybe_idx { - audio_streams.remove(idx); - } } + let _ = global_event_tx_clone.send(ServerEvent::SendAllOtherStreams(current_client_id)); } - let _ = global_event_tx_clone.send(ServerEvent::SendAllOtherStreams(current_client_id)); - } - _ => eprintln!("i dont support other requests yet"), - }, - Err(e) => eprintln!("failed to parse tcp packet: {}", e), - }, - Err(e) => eprintln!("failed to receive tcp packet: {}", e), + }, + Err(e) => eprintln!("failed to parse tcp packet: {}", e), + } + } + Some(Err(e)) => { + eprintln!("socket read error: {}", e); + break; + } } } event_result = global_event_rx.recv() => { @@ -193,36 +189,28 @@ async fn main() { match event { ServerEvent::UserConnected(new_user_id) => { if new_user_id != current_client_id { - let _ = tcp_stream - .write_all( - serde_json::to_string(&TcpResponse { - status: ResponseStatus::Ok, - data: shared::TcpResponseData::UserConnected { - user_id: new_user_id - }, - }) - .expect("failed to parse response json ServerEvent::UserConnected") - .as_bytes(), - ) - .await; + let response_json = serde_json::to_string(&TcpResponse { + status: ResponseStatus::Ok, + data: shared::TcpResponseData::UserConnected { + user_id: new_user_id + }, + }).expect("failed to serialize ServerEvent::UserConnected"); + + let _ = framed_tcp_stream.send(Bytes::from(response_json)).await; let _ = global_event_tx_clone.send(ServerEvent::SendAllOtherStreams(current_client_id)); } } ServerEvent::UserDisconnected(user_id) => { if user_id != current_client_id { - let _ = tcp_stream - .write_all( - serde_json::to_string(&TcpResponse { - status: ResponseStatus::Ok, - data: shared::TcpResponseData::UserDisconnected { - user_id: user_id - }, - }) - .expect("failed to parse response json ServerEvent::UserConnected") - .as_bytes(), - ) - .await; + let response_json = serde_json::to_string(&TcpResponse { + status: ResponseStatus::Ok, + data: shared::TcpResponseData::UserDisconnected { + user_id: user_id + }, + }).expect("failed to serialize ServerEvent::UserDisconnected"); + + let _ = framed_tcp_stream.send(Bytes::from(response_json)).await; let _ = global_event_tx_clone.send(ServerEvent::SendAllOtherStreams(current_client_id)); } @@ -231,38 +219,30 @@ async fn main() { if current_client_id != id { let clients = tcp_state.read().await; - let _ = tcp_stream - .write_all( - serde_json::to_string(&TcpResponse { - status: ResponseStatus::Ok, - data: shared::TcpResponseData::AudioStreamsFullUpdate { - audio_stream_info: scrape_all_client_streams_info(&clients), - }, - }) - .expect("failed to parse response json TcpRequestData::NewConnection") - .as_bytes(), - ) - .await; + let response_json = serde_json::to_string(&TcpResponse { + status: ResponseStatus::Ok, + data: shared::TcpResponseData::AudioStreamsFullUpdate { + audio_stream_info: scrape_all_client_streams_info(&clients), + }, + }).expect("failed to serialize ServerEvent::SendAllOtherStreams"); + + let _ = framed_tcp_stream.send(Bytes::from(response_json)).await; } } ServerEvent::SendAllStreams => { let clients = tcp_state.read().await; - let _ = tcp_stream - .write_all( - serde_json::to_string(&TcpResponse { - status: ResponseStatus::Ok, - data: shared::TcpResponseData::AudioStreamsFullUpdate { - audio_stream_info: scrape_all_client_streams_info(&clients), - }, - }) - .expect("failed to parse response json TcpRequestData::NewConnection") - .as_bytes(), - ) - .await; + let response_json = serde_json::to_string(&TcpResponse { + status: ResponseStatus::Ok, + data: shared::TcpResponseData::AudioStreamsFullUpdate { + audio_stream_info: scrape_all_client_streams_info(&clients), + }, + }).expect("failed to serialize ServerEvent::SendAllStreams"); + + let _ = framed_tcp_stream.send(Bytes::from(response_json)).await; } _ => { - println!("lkdahjsf"); + println!("unhandled event"); } } }