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

Wip #1

Merged
filip merged 16 commits from wip into main 2026-10-30 20:44:18 +01:00
Conversation 0 Commits 16 Files Changed 12
+54 -62
5 changed files with 54 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 9ee645c144 - Show all commits
client/src/audio_streams.rs
+23 -23
View File
@@ -41,11 +41,8 @@ where
for sample in data { for sample in data {
let mut mixed_value: f32 = 0.0; let mut mixed_value: f32 = 0.0;
for (_key, consumer) in streams.iter_mut() { for consumer in streams.values_mut() {
mixed_value += match consumer.try_pop() { mixed_value += consumer.try_pop().unwrap_or(0.0)
Some(s) => s,
None => 0.0,
}
} }
*sample = mixed_value.to_sample::<T>(); *sample = mixed_value.to_sample::<T>();
@@ -53,9 +50,9 @@ where
}; };
let err_fn = |err| eprintln!("stream error: {}", err); let err_fn = |err| eprintln!("stream error: {}", err);
return output_device output_device
.build_output_stream(output_device_config, output_data_fn, err_fn, None) .build_output_stream(output_device_config, output_data_fn, err_fn, None)
.expect("building output stream failed"); .expect("building output stream failed")
} }
pub async fn audio_stream_receiver( pub async fn audio_stream_receiver(
@@ -187,26 +184,29 @@ pub async fn audio_stream_receiver(
.insert((stream_info.client_id, stream_info.stream_id), decoder); .insert((stream_info.client_id, stream_info.stream_id), decoder);
} }
if !active_input_audio_streams_resamplers.contains_key(&(client_id, stream_id)) { active_input_audio_streams_resamplers
let resampler = Fft::<f32>::new( .entry((client_id, stream_id))
stream_info.sample_rate as usize, .or_insert_with(|| {
output_device_config.sample_rate() as usize, Fft::<f32>::new(
input_frames, stream_info.sample_rate as usize,
stream_info.channels as usize, output_device_config.sample_rate() as usize,
FixedSync::Both, input_frames,
) stream_info.channels as usize,
.unwrap(); FixedSync::Both,
active_input_audio_streams_resamplers.insert((client_id, stream_id), resampler); )
} .unwrap()
});
if !active_input_audio_streams_producers.contains_key(&(client_id, stream_id)) { if let std::collections::hash_map::Entry::Vacant(e) =
active_input_audio_streams_producers.entry((client_id, stream_id))
{
let ring_buffer_frames = (100.0 / 1000.0) * stream_info.sample_rate as f32; let ring_buffer_frames = (100.0 / 1000.0) * stream_info.sample_rate as f32;
let ring_buffer_samples = let ring_buffer_samples =
ring_buffer_frames as usize * stream_info.channels as usize; ring_buffer_frames as usize * stream_info.channels as usize;
let ring_buffer = HeapRb::<f32>::new(ring_buffer_samples); let ring_buffer = HeapRb::<f32>::new(ring_buffer_samples);
let (producer, consumer) = ring_buffer.split(); let (producer, consumer) = ring_buffer.split();
active_input_audio_streams_producers.insert((client_id, stream_id), producer); e.insert(producer);
active_input_audio_streams_consumers active_input_audio_streams_consumers
.lock() .lock()
.unwrap() .unwrap()
@@ -294,9 +294,9 @@ where
}; };
let err_fn = |err| eprintln!("stream error: {}", err); let err_fn = |err| eprintln!("stream error: {}", err);
return input_device input_device
.build_input_stream(input_device_config, input_data_fn, err_fn, None) .build_input_stream(input_device_config, input_data_fn, err_fn, None)
.expect("building input stream failed"); .expect("building input stream failed")
} }
pub async fn audio_stream_spawner( pub async fn audio_stream_spawner(
@@ -317,7 +317,7 @@ pub async fn audio_stream_spawner(
}; };
if connected { if connected {
let _ = global_event_tx.send(ClientEvent::NewAudioStreamRequest(AudioStreamInfo { let _ = global_event_tx.send(ClientEvent::NewAudioStreamRequest(AudioStreamInfo {
client_id: client_id, client_id,
stream_id: current_stream_id, stream_id: current_stream_id,
channels: force_channels, channels: force_channels,
sample_rate: SAMPLE_RATE_48K, sample_rate: SAMPLE_RATE_48K,
client/src/main.rs
+2 -2
View File
@@ -72,14 +72,14 @@ async fn main() {
tokio::spawn(audio_stream_receiver( tokio::spawn(audio_stream_receiver(
output_device, output_device,
output_device_config.into(), output_device_config,
active_input_audio_streams_info.clone(), active_input_audio_streams_info.clone(),
udp_socket.clone(), udp_socket.clone(),
)); ));
tokio::spawn(audio_stream_spawner( tokio::spawn(audio_stream_spawner(
input_device, input_device,
input_device_config.into(), input_device_config,
global_event_tx.clone(), global_event_tx.clone(),
global_event_tx.subscribe(), global_event_tx.subscribe(),
udp_socket.clone(), udp_socket.clone(),
client/src/tcp_connection.rs
+2 -2
View File
@@ -94,7 +94,7 @@ pub async fn tcp_connection(
}) })
.expect("failed to parse initial request"); .expect("failed to parse initial request");
let _ = framed_tcp_stream.send(Bytes::from(request_json)) framed_tcp_stream.send(Bytes::from(request_json))
.await .await
.expect("failed to send initial message"); .expect("failed to send initial message");
@@ -110,7 +110,7 @@ pub async fn tcp_connection(
} }
}).expect("failed to parse initial request"); }).expect("failed to parse initial request");
let _ = framed_tcp_stream.send(Bytes::from(request_json)) framed_tcp_stream.send(Bytes::from(request_json))
.await .await
.expect("failed to send initial message"); .expect("failed to send initial message");
} }
client/src/tui.rs
+16 -21
View File
@@ -75,7 +75,7 @@ impl BlankScreen {
self.handle_events(); self.handle_events();
} }
return TuiScreen::Exit; TuiScreen::Exit
} }
fn draw(&mut self, frame: &mut Frame) { fn draw(&mut self, frame: &mut Frame) {
@@ -85,10 +85,7 @@ impl BlankScreen {
fn handle_events(&mut self) { fn handle_events(&mut self) {
match event::read().expect("failed to read event") { match event::read().expect("failed to read event") {
Event::Key(key_event) if key_event.kind == KeyEventKind::Press => { Event::Key(key_event) if key_event.kind == KeyEventKind::Press => {
match key_event.code { if let KeyCode::Char('q') = key_event.code { self.exit = true }
KeyCode::Char('q') => self.exit = true,
_ => {}
}
} }
_ => {} _ => {}
} }
@@ -121,11 +118,10 @@ impl DeviceSelectScreen {
let mut input_devices: Vec<(Device, SupportedStreamConfig)> = Vec::new(); let mut input_devices: Vec<(Device, SupportedStreamConfig)> = Vec::new();
if let Some(default_input_device) = host.default_input_device() { if let Some(default_input_device) = host.default_input_device()
if let Ok(default_config) = default_input_device.default_input_config() { && let Ok(default_config) = default_input_device.default_input_config() {
input_devices.push((default_input_device.clone(), default_config)); input_devices.push((default_input_device.clone(), default_config));
} }
}
if let Ok(possible_input_devices) = host.input_devices() { if let Ok(possible_input_devices) = host.input_devices() {
for possible_input_device in possible_input_devices { for possible_input_device in possible_input_devices {
@@ -137,16 +133,15 @@ impl DeviceSelectScreen {
let mut output_devices: Vec<(Device, SupportedStreamConfig)> = Vec::new(); let mut output_devices: Vec<(Device, SupportedStreamConfig)> = Vec::new();
if let Some(default_output_device) = host.default_output_device() { if let Some(default_output_device) = host.default_output_device()
if let Ok(default_config) = default_output_device.default_output_config() { && let Ok(default_config) = default_output_device.default_output_config() {
output_devices.push((default_output_device.clone(), default_config.into())); output_devices.push((default_output_device.clone(), default_config));
} }
}
if let Ok(possible_output_devices) = host.output_devices() { if let Ok(possible_output_devices) = host.output_devices() {
for possible_output_device in possible_output_devices { for possible_output_device in possible_output_devices {
if let Ok(default_config) = possible_output_device.default_output_config() { if let Ok(default_config) = possible_output_device.default_output_config() {
output_devices.push((possible_output_device.clone(), default_config.into())); output_devices.push((possible_output_device.clone(), default_config));
} }
} }
} }
@@ -162,13 +157,13 @@ impl DeviceSelectScreen {
exit_screen: TuiScreen::Exit, exit_screen: TuiScreen::Exit,
input_device_state: input_state, input_device_state: input_state,
output_device_state: output_state, output_device_state: output_state,
input_devices: input_devices, input_devices,
output_devices: output_devices, output_devices,
selected_selection: 0, selected_selection: 0,
audio_control: audio_control, audio_control,
global_event_tx: global_event_tx, global_event_tx,
client_info: client_info, client_info,
} }
} }
@@ -180,7 +175,7 @@ impl DeviceSelectScreen {
self.handle_events(); self.handle_events();
} }
return self.exit_screen; self.exit_screen
} }
fn draw(&mut self, frame: &mut Frame) { fn draw(&mut self, frame: &mut Frame) {
@@ -344,14 +339,14 @@ impl DeviceSelectScreen {
tokio::spawn(audio_stream_receiver( tokio::spawn(audio_stream_receiver(
output_device.clone(), output_device.clone(),
output_config.clone(), *output_config,
self.audio_control.active_input_audio_streams_info.clone(), self.audio_control.active_input_audio_streams_info.clone(),
self.audio_control.udp_socket.clone(), self.audio_control.udp_socket.clone(),
)); ));
tokio::spawn(audio_stream_spawner( tokio::spawn(audio_stream_spawner(
input_device.clone(), input_device.clone(),
input_config.clone(), *input_config,
self.global_event_tx.clone(), self.global_event_tx.clone(),
self.global_event_tx.subscribe(), self.global_event_tx.subscribe(),
self.audio_control.udp_socket.clone(), self.audio_control.udp_socket.clone(),
server/src/main.rs
+11 -14
View File
@@ -35,13 +35,13 @@ type State = Arc<RwLock<HashMap<u32, Client>>>;
fn scrape_all_client_streams_info(clients: &HashMap<u32, Client>) -> Vec<AudioStreamInfo> { fn scrape_all_client_streams_info(clients: &HashMap<u32, Client>) -> Vec<AudioStreamInfo> {
let mut all_audio_streams_info: Vec<AudioStreamInfo> = Vec::new(); let mut all_audio_streams_info: Vec<AudioStreamInfo> = Vec::new();
for (_, data) in clients { for data in clients.values() {
if let Some(audio_streams_info) = &data.audio_streams { if let Some(audio_streams_info) = &data.audio_streams {
all_audio_streams_info.extend(audio_streams_info.iter().cloned()); all_audio_streams_info.extend(audio_streams_info.iter().cloned());
} }
} }
return all_audio_streams_info; all_audio_streams_info
} }
#[tokio::main] #[tokio::main]
@@ -84,11 +84,10 @@ async fn main() {
} }
for (&other_client_id, other_client) in clients.iter() { for (&other_client_id, other_client) in clients.iter() {
if other_client_id != client_id { if other_client_id != client_id
if let Some(destination_udp) = other_client.udp_addr { && let Some(destination_udp) = other_client.udp_addr {
let _ = udp_socket.send_to(&buf[..len], destination_udp).await; let _ = udp_socket.send_to(&buf[..len], destination_udp).await;
} }
}
} }
} }
}); });
@@ -136,7 +135,7 @@ async fn main() {
}, },
}).expect("failed to serialize TcpRequestData::NewAudioStreamAccepted"); }).expect("failed to serialize TcpRequestData::NewAudioStreamAccepted");
let _ = framed_tcp_stream.send(Bytes::from(response_json)) framed_tcp_stream.send(Bytes::from(response_json))
.await .await
.expect("failed to send message"); .expect("failed to send message");
@@ -145,9 +144,9 @@ async fn main() {
TcpRequestData::NewAudioStream { audio_stream_info } => { TcpRequestData::NewAudioStream { audio_stream_info } => {
if let Some(client) = tcp_state.write().await.get_mut(&current_client_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(audio_streams) = &mut client.audio_streams {
audio_streams.push(audio_stream_info.clone()); audio_streams.push(audio_stream_info);
} else { } else {
client.audio_streams = Some(vec![audio_stream_info.clone()]); client.audio_streams = Some(vec![audio_stream_info]);
} }
} }
@@ -163,13 +162,11 @@ async fn main() {
let _ = global_event_tx_clone.send(ServerEvent::SendAllOtherStreams(current_client_id)); let _ = global_event_tx_clone.send(ServerEvent::SendAllOtherStreams(current_client_id));
} }
TcpRequestData::DropAudioStream { audio_stream_id } => { TcpRequestData::DropAudioStream { audio_stream_id } => {
if let Some(client) = tcp_state.write().await.get_mut(&current_client_id) { if let Some(client) = tcp_state.write().await.get_mut(&current_client_id)
if let Some(audio_streams) = &mut client.audio_streams { && let Some(audio_streams) = &mut client.audio_streams
if let Some(idx) = audio_streams.iter().position(|s| s.stream_id == audio_stream_id) { && let Some(idx) = audio_streams.iter().position(|s| s.stream_id == audio_stream_id) {
audio_streams.remove(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));
} }
}, },
@@ -205,7 +202,7 @@ async fn main() {
let response_json = serde_json::to_string(&TcpResponse { let response_json = serde_json::to_string(&TcpResponse {
status: ResponseStatus::Ok, status: ResponseStatus::Ok,
data: shared::TcpResponseData::UserDisconnected { data: shared::TcpResponseData::UserDisconnected {
user_id: user_id user_id
}, },
}).expect("failed to serialize ServerEvent::UserDisconnected"); }).expect("failed to serialize ServerEvent::UserDisconnected");