filip/smalltalk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

tcp framing

Browse Source
This commit is contained in:
filip
2026-09-11 23:16:51 +02:00
parent 6b47abcd89
commit 472482f12e
5 changed files with 288 additions and 217 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+82
View File
@@ -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"
client/Cargo.toml
+2
View File
@@ -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"
client/src/tcp_connection.rs
+103 -98
View File
@@ -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::<TcpResponse>(&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::<TcpResponse>(&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::<f32>::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::<f32>::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");
}
_ => {}
}
server/Cargo.toml
+2
View File
@@ -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"
server/src/main.rs
+99 -119
View File
@@ -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::<TcpRequest>(&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::<TcpRequest>(&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(&current_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(&current_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(&current_client_id) {
if let Some(audio_streams) = &mut client.audio_streams {
let mut maybe_idx: Option<usize> = 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(&current_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");
}
}
}