From 8555cc1a7664cecd8b34c234a48d272c3307cc7b Mon Sep 17 00:00:00 2001 From: Filip Kojro Date: Wed, 16 Sep 2026 16:07:05 +0200 Subject: [PATCH 01/16] converting audio streams to specified channels (dont work for every combination) --- client/src/audio_streams.rs | 65 ++++++++++++++++++++++++++----------- client/src/main.rs | 1 + client/src/tui.rs | 7 ++-- 3 files changed, 51 insertions(+), 22 deletions(-) diff --git a/client/src/audio_streams.rs b/client/src/audio_streams.rs index c917509..1d85176 100644 --- a/client/src/audio_streams.rs +++ b/client/src/audio_streams.rs @@ -122,7 +122,7 @@ pub async fn audio_stream_receiver( let mut receive_buffer = vec![0u8; 4096]; - let mut decoded_samples = [0.0f32; 1024]; + let mut decoded_samples = [0.0f32; 2048]; let mut decoded_samples_converted = [0.0f32; 2048]; loop { @@ -240,6 +240,7 @@ pub async fn audio_stream_spawner( mut global_event_rx: Receiver, udp_socket: Arc, client_info: Arc>, + force_channels: u16, ) { let current_stream_id: u8 = rand::random(); @@ -252,7 +253,7 @@ pub async fn audio_stream_spawner( let _ = global_event_tx.send(ClientEvent::NewAudioStreamRequest(AudioStreamInfo { client_id: client_id, stream_id: current_stream_id, - channels: input_device_config.channels(), + channels: force_channels, sample_rate: input_device_config.sample_rate(), })); break; @@ -274,8 +275,7 @@ pub async fn audio_stream_spawner( let ring_buffer_ms = 80.0; let ring_buffer_frames = (ring_buffer_ms / 1000.0) * input_device_config.sample_rate() as f32; - let ring_buffer_samples = ring_buffer_frames as usize * input_device_config.channels() as usize; - let ring_buffer_size = ring_buffer_samples; + let ring_buffer_size = ring_buffer_frames as usize * input_device_config.channels() as usize; let ring_buffer = HeapRb::::new(ring_buffer_size); let (producer, mut consumer) = ring_buffer.split(); @@ -318,32 +318,63 @@ pub async fn audio_stream_spawner( input_stream.play().expect("failed to play input stream"); - let frames_needed = ((20.0 / 1000.0) * input_device_config.sample_rate() as f32) as usize; - let mut frame_buffer = vec![0.0; frames_needed]; + let samples_needed = ((20.0 / 1000.0) + * input_device_config.sample_rate() as f32 + * input_device_config.channels() as f32) as usize; - let channels: ropus::Channels = { - if input_device_config.channels() == 2 { - ropus::Channels::Stereo - } else { - ropus::Channels::Mono + let output_samples_needed = ((20.0 / 1000.0) + * input_device_config.sample_rate() as f32 + * force_channels as f32) as usize; + + let mut frame_buffer = vec![0f32; samples_needed]; + let mut output_frame_buffer = vec![0f32; output_samples_needed]; + + let encoder_channels: ropus::Channels = match force_channels { + 1 => ropus::Channels::Mono, + 2 => ropus::Channels::Stereo, + _ => { + eprintln!("unsupported by encoder channel count"); + return; } }; let mut opus_encoder = Encoder::builder( input_device_config.sample_rate(), - channels, + encoder_channels, ropus::Application::Voip, ) .build() .unwrap(); - let mut encoded_frame_buffer = [0u8; 1024]; + let mut encoded_frame_buffer = [0u8; 2048]; loop { - if consumer.occupied_len() >= frames_needed { + if consumer.occupied_len() >= samples_needed { consumer.pop_slice(&mut frame_buffer); - let mut payload = Vec::with_capacity(4 + 1 + frames_needed * 4); + if force_channels == input_device_config.channels() { + output_frame_buffer.copy_from_slice(frame_buffer.as_slice()); + } else if force_channels < input_device_config.channels() { + let div = (input_device_config.channels() / force_channels) as usize; + + output_frame_buffer.fill(0.0); + + for (i, input_sample) in frame_buffer.iter().enumerate() { + output_frame_buffer[i / div] += *input_sample / div as f32; + } + } else { + let div = (force_channels / input_device_config.channels()) as usize; + + for (i, output_sample) in output_frame_buffer.iter_mut().enumerate() { + *output_sample = frame_buffer[i / div]; + } + } + + let encoded_len = opus_encoder + .encode_float(&output_frame_buffer.as_slice(), &mut encoded_frame_buffer) + .expect("failed to encode frame"); + + let mut payload = Vec::::with_capacity(4 + 1 + encoded_len * 4); let (server_address, client_id) = { let info = client_info.read().unwrap(); @@ -353,10 +384,6 @@ pub async fn audio_stream_spawner( payload.extend(client_id.to_be_bytes()); payload.extend(current_stream_id.to_be_bytes()); - let encoded_len = opus_encoder - .encode_float(&frame_buffer.as_slice(), &mut encoded_frame_buffer) - .expect("failed to encode frame"); - payload.extend_from_slice(&encoded_frame_buffer[..encoded_len]); let _ = udp_socket.send_to(&payload, server_address).await; diff --git a/client/src/main.rs b/client/src/main.rs index a2c591a..2ed1d5a 100644 --- a/client/src/main.rs +++ b/client/src/main.rs @@ -101,6 +101,7 @@ async fn main() { global_event_tx.subscribe(), udp_socket.clone(), client_info.clone(), + 1, )); tokio::signal::ctrl_c().await.expect("failed to close"); diff --git a/client/src/tui.rs b/client/src/tui.rs index fad8a91..f955928 100644 --- a/client/src/tui.rs +++ b/client/src/tui.rs @@ -1,5 +1,5 @@ use cpal::{ - Device, SAMPLE_RATE_48K, SampleFormat, SupportedStreamConfig, + Device, SAMPLE_RATE_48K, SupportedStreamConfig, traits::{DeviceTrait, HostTrait}, }; use crossterm::event::{self, Event, KeyCode, KeyEvent, KeyEventKind}; @@ -264,7 +264,7 @@ impl DeviceSelectScreen { for (device, config) in self.input_devices.iter() { input_device_rows.push(Row::new(vec![ device.description().unwrap().name().to_string(), - config.sample_format().to_string(), + config.channels().to_string(), ])); } let input_device_table = Table::new(input_device_rows, widths) @@ -299,7 +299,7 @@ impl DeviceSelectScreen { for (device, config) in self.output_devices.iter() { output_device_rows.push(Row::new(vec![ device.description().unwrap().name().to_string(), - config.sample_format().to_string(), + config.channels().to_string(), ])); } let output_device_table = Table::new(output_device_rows, widths) @@ -413,6 +413,7 @@ impl DeviceSelectScreen { self.global_event_tx.subscribe(), self.audio_control.udp_socket.clone(), self.client_info.clone(), + 1, )); self.exit_screen = TuiScreen::Blank; -- 2.54.0 From cc8ed4bd4c07437b980a6ea583c6fc5c4efbd547 Mon Sep 17 00:00:00 2001 From: Filip Kojro Date: Fri, 25 Sep 2026 22:58:23 +0200 Subject: [PATCH 02/16] dockerfile for server --- Dockerfile | 8 ++++++++ README.md | 41 ++++++++++++++++++++--------------------- 2 files changed, 28 insertions(+), 21 deletions(-) create mode 100644 Dockerfile diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..859c9b2 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,8 @@ +FROM rust:alpine as builder +WORKDIR /usr/src/smalltalk +COPY . . +RUN cargo build -p server --release + +FROM alpine:latest +COPY --from=builder /usr/src/smalltalk/target/release/server /usr/local/bin/smalltalk +CMD ["smalltalk"] diff --git a/README.md b/README.md index b79dac3..1b526dc 100644 --- a/README.md +++ b/README.md @@ -1,24 +1,17 @@ # smalltalk -## build and run client +## running server -Usage: +### using docker ```bash -cargo run -p client --release -- --server-ip +docker build -t smalltalk . +docker run -d --rm -p 6767:6767/tcp -p 6767:6767/udp --name smalltalk smalltalk ``` -Options: +### using cargo -``` --s, --server-ip --h, --help Print help --V, --version Print version -``` - -## build and run server - -before, open port `8080` tcp/udp +before, open port `6767` tcp and udp Usage: @@ -29,17 +22,23 @@ cargo run -p server --release -- [OPTIONS] Options: ``` --s, --server-ip [default: 0.0.0.0:8080] +-s, --server-ip [default: 0.0.0.0:6767] -h, --help Print help -V, --version Print version ``` ---- +## build and run client -I think it will be based on: +Usage: -- `tokio`: async and network connection -- `cpal`: raw audio input and output -- `opus`: encoding and decoding audio stream -- `ffmpeg-next`: capturing, encoding and decoding video stream -- `egui` + `eframe`: ui +```bash +cargo run -p client --release -- --server-ip +``` + +Options: +``` +-c, --connect-with-defaults +-s, --server-ip +-h, --help Print help +-V, --version Print version +``` -- 2.54.0 From 8a621236493da5002645c6f475c645709c0dce63 Mon Sep 17 00:00:00 2001 From: Filip Kojro Date: Tue, 29 Sep 2026 20:40:46 +0200 Subject: [PATCH 03/16] adding actions --- .gitea/workflows/build.yaml | 46 +++++++++++++++++++++++++++++++++++++ 1 file changed, 46 insertions(+) create mode 100644 .gitea/workflows/build.yaml diff --git a/.gitea/workflows/build.yaml b/.gitea/workflows/build.yaml new file mode 100644 index 0000000..7c11061 --- /dev/null +++ b/.gitea/workflows/build.yaml @@ -0,0 +1,46 @@ +on: [push, pull_request] + +name: Continuous integration + +jobs: + check: + name: Check + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions-rust-lang/setup-rust-toolchain@v1 + - run: cargo check + + test: + name: Test Suite + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions-rust-lang/setup-rust-toolchain@v1 + - run: cargo test + + fmt: + name: Rustfmt + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions-rust-lang/setup-rust-toolchain@v1 + - run: rustup component add rustfmt + - run: cargo fmt --all -- --check + + clippy: + name: Clippy + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions-rust-lang/setup-rust-toolchain@v1 + - run: rustup component add clippy + - run: cargo clippy -- -D warnings + + build: + name: build + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions-rust-lang/setup-rust-toolchain@v1 + - run: cargo build -- 2.54.0 From a3d0f40d2999c6989dd2a030236a33b42b2c0ca8 Mon Sep 17 00:00:00 2001 From: Filip Kojro Date: Tue, 29 Sep 2026 20:55:54 +0200 Subject: [PATCH 04/16] adding libs to actions for alsa-sys --- .gitea/workflows/build.yaml | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/.gitea/workflows/build.yaml b/.gitea/workflows/build.yaml index 7c11061..0a9d233 100644 --- a/.gitea/workflows/build.yaml +++ b/.gitea/workflows/build.yaml @@ -9,6 +9,7 @@ jobs: steps: - uses: actions/checkout@v4 - uses: actions-rust-lang/setup-rust-toolchain@v1 + - run: apt-get update -y && apt-get upgrade -y && apt-get install g++ pkg-config libx11-dev libasound2-dev libudev-dev libxkbcommon-x11-0 - run: cargo check test: @@ -17,6 +18,7 @@ jobs: steps: - uses: actions/checkout@v4 - uses: actions-rust-lang/setup-rust-toolchain@v1 + - run: apt-get update -y && apt-get upgrade -y && apt-get install g++ pkg-config libx11-dev libasound2-dev libudev-dev libxkbcommon-x11-0 - run: cargo test fmt: @@ -25,6 +27,7 @@ jobs: steps: - uses: actions/checkout@v4 - uses: actions-rust-lang/setup-rust-toolchain@v1 + - run: apt-get update -y && apt-get upgrade -y && apt-get install g++ pkg-config libx11-dev libasound2-dev libudev-dev libxkbcommon-x11-0 - run: rustup component add rustfmt - run: cargo fmt --all -- --check @@ -34,6 +37,7 @@ jobs: steps: - uses: actions/checkout@v4 - uses: actions-rust-lang/setup-rust-toolchain@v1 + - run: apt-get update -y && apt-get upgrade -y && apt-get install g++ pkg-config libx11-dev libasound2-dev libudev-dev libxkbcommon-x11-0 - run: rustup component add clippy - run: cargo clippy -- -D warnings @@ -43,4 +47,5 @@ jobs: steps: - uses: actions/checkout@v4 - uses: actions-rust-lang/setup-rust-toolchain@v1 + - run: apt-get update -y && apt-get upgrade -y && apt-get install g++ pkg-config libx11-dev libasound2-dev libudev-dev libxkbcommon-x11-0 - run: cargo build -- 2.54.0 From 88f6384f90c062499ce07626b958cfa233a11ef9 Mon Sep 17 00:00:00 2001 From: Filip Kojro Date: Tue, 29 Sep 2026 20:58:25 +0200 Subject: [PATCH 05/16] yes install --- .gitea/workflows/build.yaml | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/.gitea/workflows/build.yaml b/.gitea/workflows/build.yaml index 0a9d233..45daea9 100644 --- a/.gitea/workflows/build.yaml +++ b/.gitea/workflows/build.yaml @@ -9,7 +9,7 @@ jobs: steps: - uses: actions/checkout@v4 - uses: actions-rust-lang/setup-rust-toolchain@v1 - - run: apt-get update -y && apt-get upgrade -y && apt-get install g++ pkg-config libx11-dev libasound2-dev libudev-dev libxkbcommon-x11-0 + - run: apt-get update -y && apt-get upgrade -y && apt-get install -y g++ pkg-config libx11-dev libasound2-dev libudev-dev libxkbcommon-x11-0 - run: cargo check test: @@ -18,7 +18,7 @@ jobs: steps: - uses: actions/checkout@v4 - uses: actions-rust-lang/setup-rust-toolchain@v1 - - run: apt-get update -y && apt-get upgrade -y && apt-get install g++ pkg-config libx11-dev libasound2-dev libudev-dev libxkbcommon-x11-0 + - run: apt-get update -y && apt-get upgrade -y && apt-get install -y g++ pkg-config libx11-dev libasound2-dev libudev-dev libxkbcommon-x11-0 - run: cargo test fmt: @@ -27,7 +27,7 @@ jobs: steps: - uses: actions/checkout@v4 - uses: actions-rust-lang/setup-rust-toolchain@v1 - - run: apt-get update -y && apt-get upgrade -y && apt-get install g++ pkg-config libx11-dev libasound2-dev libudev-dev libxkbcommon-x11-0 + - run: apt-get update -y && apt-get upgrade -y && apt-get install -y g++ pkg-config libx11-dev libasound2-dev libudev-dev libxkbcommon-x11-0 - run: rustup component add rustfmt - run: cargo fmt --all -- --check @@ -37,7 +37,7 @@ jobs: steps: - uses: actions/checkout@v4 - uses: actions-rust-lang/setup-rust-toolchain@v1 - - run: apt-get update -y && apt-get upgrade -y && apt-get install g++ pkg-config libx11-dev libasound2-dev libudev-dev libxkbcommon-x11-0 + - run: apt-get update -y && apt-get upgrade -y && apt-get install -y g++ pkg-config libx11-dev libasound2-dev libudev-dev libxkbcommon-x11-0 - run: rustup component add clippy - run: cargo clippy -- -D warnings @@ -47,5 +47,5 @@ jobs: steps: - uses: actions/checkout@v4 - uses: actions-rust-lang/setup-rust-toolchain@v1 - - run: apt-get update -y && apt-get upgrade -y && apt-get install g++ pkg-config libx11-dev libasound2-dev libudev-dev libxkbcommon-x11-0 + - run: apt-get update -y && apt-get upgrade -y && apt-get install -y g++ pkg-config libx11-dev libasound2-dev libudev-dev libxkbcommon-x11-0 - run: cargo build -- 2.54.0 From a1f4409ed56c7d26b47d5e9692c6c443bb7f355e Mon Sep 17 00:00:00 2001 From: Filip Kojro Date: Tue, 29 Sep 2026 21:05:51 +0200 Subject: [PATCH 06/16] allowing unused enum for events --- client/src/models.rs | 1 + server/src/main.rs | 1 + 2 files changed, 2 insertions(+) diff --git a/client/src/models.rs b/client/src/models.rs index f3f4187..0d03a59 100644 --- a/client/src/models.rs +++ b/client/src/models.rs @@ -25,6 +25,7 @@ pub enum TuiScreen { Exit, } +#[allow(dead_code)] #[derive(Clone)] pub enum ClientEvent { NewConnection, diff --git a/server/src/main.rs b/server/src/main.rs index 07ccd54..c1b1f2c 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -23,6 +23,7 @@ struct Client { audio_streams: Option>, } +#[allow(dead_code)] #[derive(Clone)] enum ServerEvent { NewAudioStream(AudioStreamInfo), -- 2.54.0 From a7873c564cd43d06db6d8ea4b2f1aac324b5b072 Mon Sep 17 00:00:00 2001 From: Filip Kojro Date: Wed, 30 Sep 2026 16:14:09 +0200 Subject: [PATCH 07/16] cache for apt packages for actions --- .gitea/actions/setup-env/actions.yaml | 14 ++++++++++++++ .gitea/workflows/build.yaml | 15 +++++---------- 2 files changed, 19 insertions(+), 10 deletions(-) create mode 100644 .gitea/actions/setup-env/actions.yaml diff --git a/.gitea/actions/setup-env/actions.yaml b/.gitea/actions/setup-env/actions.yaml new file mode 100644 index 0000000..76d084f --- /dev/null +++ b/.gitea/actions/setup-env/actions.yaml @@ -0,0 +1,14 @@ +name: 'Setup Environment' +description: 'Sets up Rust toolchain and caches APT dependencies' + +runs: + using: "composite" + steps: + - name: Setup Rust toolchain + uses: actions-rust-lang/setup-rust-toolchain@v1 + + - name: Cache and install APT packages + uses: awalsh128/cache-apt-pkgs-action@latest + with: + packages: g++ pkg-config libx11-dev libasound2-dev libudev-dev libxkbcommon-x11-0 + version: 1.0 diff --git a/.gitea/workflows/build.yaml b/.gitea/workflows/build.yaml index 45daea9..bb13c97 100644 --- a/.gitea/workflows/build.yaml +++ b/.gitea/workflows/build.yaml @@ -8,8 +8,7 @@ jobs: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - - uses: actions-rust-lang/setup-rust-toolchain@v1 - - run: apt-get update -y && apt-get upgrade -y && apt-get install -y g++ pkg-config libx11-dev libasound2-dev libudev-dev libxkbcommon-x11-0 + - uses: ./.gitea/actions/setup-env - run: cargo check test: @@ -17,8 +16,7 @@ jobs: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - - uses: actions-rust-lang/setup-rust-toolchain@v1 - - run: apt-get update -y && apt-get upgrade -y && apt-get install -y g++ pkg-config libx11-dev libasound2-dev libudev-dev libxkbcommon-x11-0 + - uses: ./.gitea/actions/setup-env - run: cargo test fmt: @@ -26,8 +24,7 @@ jobs: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - - uses: actions-rust-lang/setup-rust-toolchain@v1 - - run: apt-get update -y && apt-get upgrade -y && apt-get install -y g++ pkg-config libx11-dev libasound2-dev libudev-dev libxkbcommon-x11-0 + - uses: ./.gitea/actions/setup-env - run: rustup component add rustfmt - run: cargo fmt --all -- --check @@ -36,8 +33,7 @@ jobs: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - - uses: actions-rust-lang/setup-rust-toolchain@v1 - - run: apt-get update -y && apt-get upgrade -y && apt-get install -y g++ pkg-config libx11-dev libasound2-dev libudev-dev libxkbcommon-x11-0 + - uses: ./.gitea/actions/setup-env - run: rustup component add clippy - run: cargo clippy -- -D warnings @@ -46,6 +42,5 @@ jobs: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - - uses: actions-rust-lang/setup-rust-toolchain@v1 - - run: apt-get update -y && apt-get upgrade -y && apt-get install -y g++ pkg-config libx11-dev libasound2-dev libudev-dev libxkbcommon-x11-0 + - uses: ./.gitea/actions/setup-env - run: cargo build -- 2.54.0 From 0988ada8acf62634a4d77c4d6eb2cb1d591f84d0 Mon Sep 17 00:00:00 2001 From: Filip Kojro Date: Wed, 30 Sep 2026 16:15:48 +0200 Subject: [PATCH 08/16] fix cache action name --- .gitea/actions/setup-env/{actions.yaml => action.yaml} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename .gitea/actions/setup-env/{actions.yaml => action.yaml} (100%) diff --git a/.gitea/actions/setup-env/actions.yaml b/.gitea/actions/setup-env/action.yaml similarity index 100% rename from .gitea/actions/setup-env/actions.yaml rename to .gitea/actions/setup-env/action.yaml -- 2.54.0 From 687ef4ff2c9b39411b7a07bebb457303b872784d Mon Sep 17 00:00:00 2001 From: Filip Kojro Date: Mon, 19 Oct 2026 23:29:00 +0200 Subject: [PATCH 09/16] removing device selection filters; possibly converting to 48khz input --- Cargo.lock | 137 ++++++++++++++++++++++++++++++++++++ client/Cargo.toml | 2 + client/src/audio_streams.rs | 84 ++++++++++++++-------- client/src/tui.rs | 110 ++++++++--------------------- server/src/main.rs | 25 ++----- 5 files changed, 228 insertions(+), 130 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 94a57d8..cc679a0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -113,6 +113,39 @@ dependencies = [ "bytemuck", ] +[[package]] +name = "audio-codec-algorithms" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1254ebf6529f3763c491acfb5ab6e960809e3c75a38584cc664f8c5667fb7107" + +[[package]] +name = "audioadapter" +version = "5.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d1292ef9edf681b7426ed089004b021a42896492f58bd0c909ad44e5faec9ac4" + +[[package]] +name = "audioadapter-buffers" +version = "5.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "46289a81a3bfa26d0f8b2415f8ca9decde9a66308a150568af5c102381b3c8ce" +dependencies = [ + "audioadapter", + "audioadapter-sample", + "num-traits", +] + +[[package]] +name = "audioadapter-sample" +version = "5.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5258faecf4edbe35ec483bda9b1f7438e6b391891819269c6b5461aeaa0e8d7d" +dependencies = [ + "audio-codec-algorithms", + "num-traits", +] + [[package]] name = "autocfg" version = "1.5.1" @@ -280,6 +313,7 @@ checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9" name = "client" version = "0.1.0" dependencies = [ + "audioadapter-buffers", "clap", "cpal", "crossterm", @@ -289,6 +323,7 @@ dependencies = [ "ratatui", "ringbuf", "ropus", + "rubato", "serde", "serde_json", "shared", @@ -1128,6 +1163,15 @@ dependencies = [ "minimal-lexical", ] +[[package]] +name = "num-complex" +version = "0.4.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "73f88a1307638156682bada9d7604135552957b7818057dcef22705b4d509495" +dependencies = [ + "num-traits", +] + [[package]] name = "num-conv" version = "0.2.2" @@ -1145,6 +1189,15 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "num-integer" +version = "0.1.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7ce2d95d4b3734dc35aa2f45e1aa22cd416814592a4f9d9205e11affd5b8e10b" +dependencies = [ + "num-traits", +] + [[package]] name = "num-traits" version = "0.2.19" @@ -1479,6 +1532,15 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "439ee305def115ba05938db6eb1644ff94165c5ab5e9420d1c1bcedbba909391" +[[package]] +name = "primal-check" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc0d895b311e3af9902528fbb8f928688abbd95872819320517cc24ca6b2bd08" +dependencies = [ + "num-integer", +] + [[package]] name = "proc-macro-crate" version = "3.5.0" @@ -1651,6 +1713,15 @@ dependencies = [ "unicode-width", ] +[[package]] +name = "realfft" +version = "3.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f821338fddb99d089116342c46e9f1fbf3828dba077674613e734e01d6ea8677" +dependencies = [ + "rustfft", +] + [[package]] name = "redox_syscall" version = "0.5.18" @@ -1710,6 +1781,22 @@ dependencies = [ "wide", ] +[[package]] +name = "rubato" +version = "5.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7cb1ffaf8738df50aab642a7f6465df81c6ba9e2818268053487165298114be" +dependencies = [ + "audioadapter", + "audioadapter-buffers", + "num-complex", + "num-integer", + "num-traits", + "realfft", + "visibility", + "windowfunctions", +] + [[package]] name = "rustc_version" version = "0.4.1" @@ -1719,6 +1806,20 @@ dependencies = [ "semver", ] +[[package]] +name = "rustfft" +version = "6.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "21db5f9893e91f41798c88680037dba611ca6674703c1a18601b01a72c8adb89" +dependencies = [ + "num-complex", + "num-integer", + "num-traits", + "primal-check", + "strength_reduce", + "transpose", +] + [[package]] name = "rustix" version = "1.1.4" @@ -1936,6 +2037,12 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f" +[[package]] +name = "strength_reduce" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fe895eb47f22e2ddd4dabc02bce419d2e643c8e3b585c78158b349195bc24d82" + [[package]] name = "strsim" version = "0.11.1" @@ -2218,6 +2325,16 @@ dependencies = [ "winnow", ] +[[package]] +name = "transpose" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ad61aed86bc3faea4300c7aee358b4c6d0c8d6ccc36524c96e4c92ccf26e77e" +dependencies = [ + "num-integer", + "strength_reduce", +] + [[package]] name = "typenum" version = "1.20.1" @@ -2283,6 +2400,17 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" +[[package]] +name = "visibility" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d674d135b4a8c1d7e813e2f8d1c9a58308aee4a680323066025e53132218bd91" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "vtparse" version = "0.6.2" @@ -2485,6 +2613,15 @@ version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" +[[package]] +name = "windowfunctions" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "90628d739333b7c5d2ee0b70210b97b8cddc38440c682c96fd9e2c24c2db5f3a" +dependencies = [ + "num-traits", +] + [[package]] name = "windows" version = "0.62.2" diff --git a/client/Cargo.toml b/client/Cargo.toml index f053389..6766d4e 100644 --- a/client/Cargo.toml +++ b/client/Cargo.toml @@ -18,3 +18,5 @@ crossterm = "0.29.0" gag = "1.0.0" tokio-util = { version = "0.7.19", features = ["codec"]} futures = "0.3.34" +rubato = "5.0.0" +audioadapter-buffers = "5.2.0" diff --git a/client/src/audio_streams.rs b/client/src/audio_streams.rs index 1d85176..f3b7617 100644 --- a/client/src/audio_streams.rs +++ b/client/src/audio_streams.rs @@ -1,5 +1,7 @@ +use audioadapter_buffers::direct::InterleavedSlice; use cpal::{ - Device, FromSample, Sample, SampleFormat, SizedSample, StreamConfig, SupportedStreamConfig, + Device, FromSample, SAMPLE_RATE_48K, Sample, SampleFormat, SizedSample, StreamConfig, + SupportedStreamConfig, traits::{DeviceTrait, StreamTrait}, }; use ringbuf::{ @@ -9,6 +11,7 @@ use ringbuf::{ wrap::caching::Caching, }; use ropus::Encoder; +use rubato::{Async, FixedAsync, PolynomialDegree, Resampler, audioadapter::Adapter}; use shared::AudioStreamInfo; use std::sync::{Arc, RwLock}; use tokio::{ @@ -318,16 +321,26 @@ pub async fn audio_stream_spawner( input_stream.play().expect("failed to play input stream"); - let samples_needed = ((20.0 / 1000.0) - * input_device_config.sample_rate() as f32 - * input_device_config.channels() as f32) as usize; + let frames_needed = ((20.0 / 1000.0) * input_device_config.sample_rate() as f32) as usize; - let output_samples_needed = ((20.0 / 1000.0) - * input_device_config.sample_rate() as f32 - * force_channels as f32) as usize; + let samples_needed = frames_needed * input_device_config.channels() as usize; let mut frame_buffer = vec![0f32; samples_needed]; - let mut output_frame_buffer = vec![0f32; output_samples_needed]; + + let mut resampler_input_buffer = vec![0.0f32; frames_needed * force_channels as usize]; + + let mut resampler = Async::::new_poly( + 48000.0 / 44100.0, + 1.1, + PolynomialDegree::Cubic, + frames_needed, + force_channels as usize, + FixedAsync::Output, + ) + .unwrap(); + + let mut resampler_output_buffer = + vec![0.0f32; resampler.output_frames_max() * force_channels as usize]; let encoder_channels: ropus::Channels = match force_channels { 1 => ropus::Channels::Mono, @@ -338,13 +351,10 @@ pub async fn audio_stream_spawner( } }; - let mut opus_encoder = Encoder::builder( - input_device_config.sample_rate(), - encoder_channels, - ropus::Application::Voip, - ) - .build() - .unwrap(); + let mut opus_encoder = + Encoder::builder(SAMPLE_RATE_48K, encoder_channels, ropus::Application::Voip) + .build() + .unwrap(); let mut encoded_frame_buffer = [0u8; 2048]; @@ -352,26 +362,40 @@ pub async fn audio_stream_spawner( if consumer.occupied_len() >= samples_needed { consumer.pop_slice(&mut frame_buffer); - if force_channels == input_device_config.channels() { - output_frame_buffer.copy_from_slice(frame_buffer.as_slice()); - } else if force_channels < input_device_config.channels() { - let div = (input_device_config.channels() / force_channels) as usize; + let frame_buffer_adapter = InterleavedSlice::new( + &frame_buffer, + input_device_config.channels() as usize, + frames_needed, + ) + .unwrap(); - output_frame_buffer.fill(0.0); + frame_buffer_adapter.copy_from_channel_to_slice(0, 0, &mut resampler_input_buffer); // im only using first channel for now - for (i, input_sample) in frame_buffer.iter().enumerate() { - output_frame_buffer[i / div] += *input_sample / div as f32; - } - } else { - let div = (force_channels / input_device_config.channels()) as usize; + let resampler_input = InterleavedSlice::new( + &resampler_input_buffer, + force_channels as usize, + frames_needed, + ) + .unwrap(); - for (i, output_sample) in output_frame_buffer.iter_mut().enumerate() { - *output_sample = frame_buffer[i / div]; - } - } + let mut resampler_output = InterleavedSlice::new_mut( + &mut resampler_output_buffer, + force_channels as usize, + resampler.output_frames_max(), + ) + .unwrap(); + + let (_frames_read, frames_written) = resampler + .process_into_buffer(&resampler_input, &mut resampler_output, None) + .unwrap(); + + let valid_samples = frames_written * force_channels as usize; let encoded_len = opus_encoder - .encode_float(&output_frame_buffer.as_slice(), &mut encoded_frame_buffer) + .encode_float( + &resampler_output_buffer[..valid_samples], + &mut encoded_frame_buffer, + ) .expect("failed to encode frame"); let mut payload = Vec::::with_capacity(4 + 1 + encoded_len * 4); diff --git a/client/src/tui.rs b/client/src/tui.rs index f955928..d0bae39 100644 --- a/client/src/tui.rs +++ b/client/src/tui.rs @@ -123,46 +123,14 @@ impl DeviceSelectScreen { if let Some(default_input_device) = host.default_input_device() { if let Ok(default_config) = default_input_device.default_input_config() { - if default_config.sample_rate() == SAMPLE_RATE_48K - && (default_config.channels() == 1) - { - input_devices.push((default_input_device.clone(), default_config)); - } - } else if let Ok(supported_configs) = default_input_device.supported_input_configs() { - for supported_config in supported_configs { - if let Some(input_config) = - supported_config.try_with_sample_rate(SAMPLE_RATE_48K) - { - if input_config.channels() == 1 { - input_devices.push((default_input_device, input_config)); - break; - } - } - } + input_devices.push((default_input_device.clone(), default_config)); } } if let Ok(possible_input_devices) = host.input_devices() { for possible_input_device in possible_input_devices { if let Ok(default_config) = possible_input_device.default_input_config() { - if default_config.sample_rate() == SAMPLE_RATE_48K - && (default_config.channels() == 1) - { - input_devices.push((possible_input_device.clone(), default_config)); - continue; - } - } - if let Ok(supported_configs) = possible_input_device.supported_input_configs() { - for supported_config in supported_configs { - if let Some(input_config) = - supported_config.try_with_sample_rate(SAMPLE_RATE_48K) - { - if input_config.channels() == 1 { - input_devices.push((possible_input_device, input_config)); - break; - } - } - } + input_devices.push((possible_input_device.clone(), default_config)); } } } @@ -171,47 +139,14 @@ impl DeviceSelectScreen { if let Some(default_output_device) = host.default_output_device() { if let Ok(default_config) = default_output_device.default_output_config() { - if default_config.sample_rate() == SAMPLE_RATE_48K - && (default_config.channels() == 1 || default_config.channels() == 2) - { - output_devices.push((default_output_device.clone(), default_config.into())); - } - } else if let Ok(supported_configs) = default_output_device.supported_output_configs() { - for supported_config in supported_configs { - if let Some(output_config) = - supported_config.try_with_sample_rate(SAMPLE_RATE_48K) - { - if output_config.channels() == 1 || output_config.channels() == 2 { - output_devices.push((default_output_device, output_config.into())); - break; - } - } - } + output_devices.push((default_output_device.clone(), default_config.into())); } } if let Ok(possible_output_devices) = host.output_devices() { for possible_output_device in possible_output_devices { if let Ok(default_config) = possible_output_device.default_output_config() { - if default_config.sample_rate() == SAMPLE_RATE_48K - && (default_config.channels() == 1 || default_config.channels() == 2) - { - output_devices - .push((possible_output_device.clone(), default_config.into())); - continue; - } - } - if let Ok(supported_configs) = possible_output_device.supported_output_configs() { - for supported_config in supported_configs { - if let Some(output_config) = - supported_config.try_with_sample_rate(SAMPLE_RATE_48K) - { - if output_config.channels() == 1 || output_config.channels() == 2 { - output_devices.push((possible_output_device, output_config.into())); - break; - } - } - } + output_devices.push((possible_output_device.clone(), default_config.into())); } } } @@ -261,11 +196,18 @@ impl DeviceSelectScreen { let widths = [Constraint::Percentage(100), Constraint::Min(8)]; let mut input_device_rows: Vec = Vec::new(); - for (device, config) in self.input_devices.iter() { - input_device_rows.push(Row::new(vec![ - device.description().unwrap().name().to_string(), - config.channels().to_string(), - ])); + for (i, (device, config)) in self.input_devices.iter().enumerate() { + if i == 0 { + input_device_rows.push(Row::new(vec![ + format!("(default) {}", device.description().unwrap().name()), + config.channels().to_string(), + ])); + } else { + input_device_rows.push(Row::new(vec![ + device.description().unwrap().name().to_string(), + config.channels().to_string(), + ])); + } } let input_device_table = Table::new(input_device_rows, widths) .header( @@ -296,11 +238,21 @@ impl DeviceSelectScreen { frame.render_stateful_widget(input_device_table, left_block, &mut self.input_device_state); let mut output_device_rows: Vec = Vec::new(); - for (device, config) in self.output_devices.iter() { - output_device_rows.push(Row::new(vec![ - device.description().unwrap().name().to_string(), - config.channels().to_string(), - ])); + for (i, (device, config)) in self.output_devices.iter().enumerate() { + if i == 0 { + output_device_rows.push(Row::new(vec![ + format!( + "(default) {}", + device.description().unwrap().name().to_string() + ), + config.channels().to_string(), + ])); + } else { + output_device_rows.push(Row::new(vec![ + device.description().unwrap().name().to_string(), + config.channels().to_string(), + ])); + } } let output_device_table = Table::new(output_device_rows, widths) .header( diff --git a/server/src/main.rs b/server/src/main.rs index c1b1f2c..223e93a 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -26,8 +26,6 @@ struct Client { #[allow(dead_code)] #[derive(Clone)] enum ServerEvent { - NewAudioStream(AudioStreamInfo), - SendAllStreams, SendAllOtherStreams(u32), UserConnected(u32), UserDisconnected(u32), @@ -71,7 +69,7 @@ async fn main() { let mut buf = vec![0u8; 4096]; loop { - let (len, src_adddress) = udp_socket.recv_from(&mut buf).await.expect("receive error"); + let (len, src_address) = udp_socket.recv_from(&mut buf).await.expect("receive error"); if len < 4 { continue; @@ -82,13 +80,13 @@ async fn main() { let mut clients = udp_state.write().await; if let Some(client) = clients.get_mut(&client_id) { - client.udp_addr = Some(src_adddress); + client.udp_addr = Some(src_address); } for (&other_client_id, other_client) in clients.iter() { if other_client_id != client_id { - if let Some(destication_udp) = other_client.udp_addr { - let _ = udp_socket.send_to(&buf[..len], destication_udp).await; + if let Some(destination_udp) = other_client.udp_addr { + let _ = udp_socket.send_to(&buf[..len], destination_udp).await; } } } @@ -230,21 +228,6 @@ async fn main() { let _ = framed_tcp_stream.send(Bytes::from(response_json)).await; } } - ServerEvent::SendAllStreams => { - let clients = tcp_state.read().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!("unhandled event"); - } } } Err(e) => eprintln!("error: {}", e), -- 2.54.0 From bc786a02b4f7853548d03e08a7555cc2438b6cc6 Mon Sep 17 00:00:00 2001 From: Filip Kojro Date: Tue, 20 Oct 2026 00:08:17 +0200 Subject: [PATCH 10/16] always send 48khz --- client/src/audio_streams.rs | 2 +- client/src/tui.rs | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/client/src/audio_streams.rs b/client/src/audio_streams.rs index f3b7617..eaf0d82 100644 --- a/client/src/audio_streams.rs +++ b/client/src/audio_streams.rs @@ -257,7 +257,7 @@ pub async fn audio_stream_spawner( client_id: client_id, stream_id: current_stream_id, channels: force_channels, - sample_rate: input_device_config.sample_rate(), + sample_rate: SAMPLE_RATE_48K, })); break; } diff --git a/client/src/tui.rs b/client/src/tui.rs index d0bae39..671eb36 100644 --- a/client/src/tui.rs +++ b/client/src/tui.rs @@ -1,5 +1,5 @@ use cpal::{ - Device, SAMPLE_RATE_48K, SupportedStreamConfig, + Device, SupportedStreamConfig, traits::{DeviceTrait, HostTrait}, }; use crossterm::event::{self, Event, KeyCode, KeyEvent, KeyEventKind}; -- 2.54.0 From 4da5614a5423e97d28be005af57673eb3d917c31 Mon Sep 17 00:00:00 2001 From: Filip Kojro Date: Tue, 20 Oct 2026 12:23:38 +0200 Subject: [PATCH 11/16] separating audio from tcp; separate resampler for every input stream --- client/src/audio_streams.rs | 188 +++++++++++++++++++++++------------ client/src/main.rs | 21 +--- client/src/models.rs | 11 +- client/src/tcp_connection.rs | 43 +------- client/src/tui.rs | 9 -- 5 files changed, 133 insertions(+), 139 deletions(-) diff --git a/client/src/audio_streams.rs b/client/src/audio_streams.rs index eaf0d82..f223674 100644 --- a/client/src/audio_streams.rs +++ b/client/src/audio_streams.rs @@ -10,18 +10,21 @@ use ringbuf::{ traits::{Consumer, Observer, Producer, Split}, wrap::caching::Caching, }; -use ropus::Encoder; -use rubato::{Async, FixedAsync, PolynomialDegree, Resampler, audioadapter::Adapter}; +use ropus::{Decoder, Encoder}; +use rubato::{Fft, FixedSync, Resampler, audioadapter::Adapter}; use shared::AudioStreamInfo; -use std::sync::{Arc, RwLock}; +use std::{ + collections::HashMap, + sync::{Arc, Mutex, RwLock}, +}; use tokio::{ net::UdpSocket, sync::broadcast::{Receiver, Sender}, }; use crate::models::{ - ClientEvent, ClientInfo, InputAudioStreamsConsumers, InputAudioStreamsEncoders, - InputAudioStreamsInfo, InputAudioStreamsProducers, + ClientEvent, ClientInfo, InputAudioStreamsConsumers, InputAudioStreamsDecoders, + InputAudioStreamsInfo, InputAudioStreamsProducers, InputAudioStreamsResamplers, }; fn cpal_build_output_stream( @@ -58,62 +61,65 @@ where pub async fn audio_stream_receiver( output_device: Device, output_device_config: SupportedStreamConfig, - active_input_audio_streams_producers: InputAudioStreamsProducers, - active_input_audio_streams_consumers: InputAudioStreamsConsumers, - active_input_audio_streams_encoders: InputAudioStreamsEncoders, active_input_audio_streams_info: InputAudioStreamsInfo, udp_socket: Arc, ) { + let mut active_input_audio_streams_producers: InputAudioStreamsProducers = HashMap::new(); + let active_input_audio_streams_consumers: InputAudioStreamsConsumers = + Arc::new(Mutex::new(HashMap::new())); + let mut active_input_audio_streams_decoders: InputAudioStreamsDecoders = HashMap::new(); + let mut active_input_audio_streams_resamplers: InputAudioStreamsResamplers = HashMap::new(); + let output_stream = match output_device_config.sample_format() { SampleFormat::F64 => cpal_build_output_stream::( &output_device, output_device_config.into(), - active_input_audio_streams_consumers, + active_input_audio_streams_consumers.clone(), ), SampleFormat::I64 => cpal_build_output_stream::( &output_device, output_device_config.into(), - active_input_audio_streams_consumers, + active_input_audio_streams_consumers.clone(), ), SampleFormat::U64 => cpal_build_output_stream::( &output_device, output_device_config.into(), - active_input_audio_streams_consumers, + active_input_audio_streams_consumers.clone(), ), SampleFormat::F32 => cpal_build_output_stream::( &output_device, output_device_config.into(), - active_input_audio_streams_consumers, + active_input_audio_streams_consumers.clone(), ), SampleFormat::I32 => cpal_build_output_stream::( &output_device, output_device_config.into(), - active_input_audio_streams_consumers, + active_input_audio_streams_consumers.clone(), ), SampleFormat::U32 => cpal_build_output_stream::( &output_device, output_device_config.into(), - active_input_audio_streams_consumers, + active_input_audio_streams_consumers.clone(), ), SampleFormat::I16 => cpal_build_output_stream::( &output_device, output_device_config.into(), - active_input_audio_streams_consumers, + active_input_audio_streams_consumers.clone(), ), SampleFormat::U16 => cpal_build_output_stream::( &output_device, output_device_config.into(), - active_input_audio_streams_consumers, + active_input_audio_streams_consumers.clone(), ), SampleFormat::I8 => cpal_build_output_stream::( &output_device, output_device_config.into(), - active_input_audio_streams_consumers, + active_input_audio_streams_consumers.clone(), ), SampleFormat::U8 => cpal_build_output_stream::( &output_device, output_device_config.into(), - active_input_audio_streams_consumers, + active_input_audio_streams_consumers.clone(), ), _ => { eprintln!("unsupported output stream format :("); @@ -125,8 +131,15 @@ pub async fn audio_stream_receiver( let mut receive_buffer = vec![0u8; 4096]; - let mut decoded_samples = [0.0f32; 2048]; - let mut decoded_samples_converted = [0.0f32; 2048]; + let max_input_raw_samples = ((20.0 / 1000.0) * 2.0 * SAMPLE_RATE_48K as f32) as usize; // 20ms, 2channels, 48khz + let mut decoded_samples = vec![0.0f32; max_input_raw_samples]; + + let mut resampled_samples = vec![0.0f32; max_input_raw_samples]; + + let output_raw_samples = ((20.0 / 1000.0) + * output_device_config.channels() as f32 + * output_device_config.sample_rate() as f32) as usize; + let mut decoded_samples_converted = vec![0.0f32; output_raw_samples]; loop { match udp_socket.recv_from(&mut receive_buffer).await { @@ -155,15 +168,56 @@ pub async fn audio_stream_receiver( } }; - let decoders_map = active_input_audio_streams_encoders.read().unwrap(); + let input_frames = ((20.0 / 1000.0) * stream_info.sample_rate as f32) as usize; + let resampler_output_frames = + ((20.0 / 1000.0) * output_device_config.sample_rate() as f32) as usize; - let decoder_mutex = decoders_map - .get(&(client_id, stream_id)) - .expect("failed to get encoder mutex"); + if !active_input_audio_streams_decoders.contains_key(&(client_id, stream_id)) { + let channels: ropus::Channels = { + if stream_info.channels == 2 { + ropus::Channels::Stereo + } else { + ropus::Channels::Mono + } + }; - let mut decoder = decoder_mutex.lock().expect("failed to get encoder"); + let decoder = Decoder::new(stream_info.sample_rate, channels) + .expect("failed to create decoder"); + active_input_audio_streams_decoders + .insert((stream_info.client_id, stream_info.stream_id), decoder); + } - let decoded_len = decoder + if !active_input_audio_streams_resamplers.contains_key(&(client_id, stream_id)) { + let resampler = Fft::::new( + stream_info.sample_rate as usize, + output_device_config.sample_rate() as usize, + input_frames, + stream_info.channels as usize, + FixedSync::Both, + ) + .unwrap(); + active_input_audio_streams_resamplers.insert((client_id, stream_id), resampler); + } + + if !active_input_audio_streams_producers.contains_key(&(client_id, stream_id)) { + let ring_buffer_frames = (100.0 / 1000.0) * stream_info.sample_rate as f32; + let ring_buffer_samples = + ring_buffer_frames as usize * stream_info.channels as usize; + let ring_buffer = HeapRb::::new(ring_buffer_samples); + let (producer, consumer) = ring_buffer.split(); + + active_input_audio_streams_producers.insert((client_id, stream_id), producer); + active_input_audio_streams_consumers + .lock() + .unwrap() + .insert((client_id, stream_id), consumer); + } + + let decoder = active_input_audio_streams_decoders + .get_mut(&(client_id, stream_id)) + .expect("failed to get decoder"); + + let _ = decoder .decode_float( audio_packet, &mut decoded_samples, @@ -171,41 +225,50 @@ pub async fn audio_stream_receiver( ) .expect("failed to decode opus"); - let mut active_streams_map = active_input_audio_streams_producers.lock().unwrap(); + let decoded_samples_adapter = InterleavedSlice::new( + &decoded_samples, + stream_info.channels as usize, + input_frames, + ) + .unwrap(); - if let Some(producer) = active_streams_map.get_mut(&(client_id, stream_id)) { - if stream_info.channels == output_device_config.channels() { - producer.push_slice( - &decoded_samples[..decoded_len * stream_info.channels as usize], - ); - } else if stream_info.channels > output_device_config.channels() { - for (i, frame) in decoded_samples - [..decoded_len * stream_info.channels as usize] - .chunks_exact(2) - .enumerate() - { - decoded_samples_converted[i] = (frame[0] + frame[1]) / 2.0; + let mut resampled_samples_adapter = InterleavedSlice::new_mut( + &mut resampled_samples, + stream_info.channels as usize, + resampler_output_frames, + ) + .unwrap(); + + let (_, frames_out) = active_input_audio_streams_resamplers + .get_mut(&(client_id, stream_id)) + .expect("failed to get resampler") + .process_into_buffer( + &decoded_samples_adapter, + &mut resampled_samples_adapter, + None, + ) + .expect("error during resampling"); + + if let Some(producer) = + active_input_audio_streams_producers.get_mut(&(client_id, stream_id)) + { + for (&resampled_sample, output_frame) in resampled_samples + .iter() + .step_by(stream_info.channels as usize) + .zip( + decoded_samples_converted + .chunks_exact_mut(output_device_config.channels() as usize), + ) + { + for output_sample in output_frame.iter_mut() { + *output_sample = resampled_sample; } - - producer.push_slice( - &decoded_samples_converted - [..decoded_len * stream_info.channels as usize], - ); - } else { - for (i, frame) in decoded_samples_converted - [..decoded_len * output_device_config.channels() as usize] - .chunks_exact_mut(2) - .enumerate() - { - frame[0] = decoded_samples[i]; - frame[1] = decoded_samples[i]; - } - - producer.push_slice( - &decoded_samples_converted - [..decoded_len * output_device_config.channels() as usize], - ); } + + producer.push_slice( + &decoded_samples_converted + [..frames_out * output_device_config.channels() as usize], + ); } } Err(e) => eprintln!("failed to receive udp packet: {}", e), @@ -329,13 +392,12 @@ pub async fn audio_stream_spawner( let mut resampler_input_buffer = vec![0.0f32; frames_needed * force_channels as usize]; - let mut resampler = Async::::new_poly( - 48000.0 / 44100.0, - 1.1, - PolynomialDegree::Cubic, + let mut resampler = Fft::::new( + input_device_config.sample_rate() as usize, + SAMPLE_RATE_48K as usize, frames_needed, force_channels as usize, - FixedAsync::Output, + FixedSync::Both, ) .unwrap(); diff --git a/client/src/main.rs b/client/src/main.rs index 2ed1d5a..5c944e3 100644 --- a/client/src/main.rs +++ b/client/src/main.rs @@ -7,17 +7,14 @@ use clap::Parser; use cpal::traits::{DeviceTrait, HostTrait}; use std::{ collections::HashMap, - sync::{Arc, Mutex, RwLock}, + sync::{Arc, RwLock}, }; use tokio::net::UdpSocket; use tokio::sync::broadcast; use crate::{ audio_streams::{audio_stream_receiver, audio_stream_spawner}, - models::{ - AudioControl, CliArgs, ClientEvent, ClientInfo, InputAudioStreamsConsumers, - InputAudioStreamsEncoders, InputAudioStreamsInfo, InputAudioStreamsProducers, - }, + models::{AudioControl, CliArgs, ClientEvent, ClientInfo, InputAudioStreamsInfo}, tcp_connection::tcp_connection, tui::Tui, }; @@ -28,23 +25,12 @@ async fn main() { let (global_event_tx, _) = broadcast::channel::(16); - let active_input_audio_streams_producers: InputAudioStreamsProducers = - Arc::new(Mutex::new(HashMap::new())); - let active_input_audio_streams_consumers: InputAudioStreamsConsumers = - Arc::new(Mutex::new(HashMap::new())); - - let active_input_audio_streams_encoders: InputAudioStreamsEncoders = - Arc::new(RwLock::new(HashMap::new())); - let active_input_audio_streams_info: InputAudioStreamsInfo = Arc::new(RwLock::new(HashMap::new())); let udp_socket = Arc::new(UdpSocket::bind("0.0.0.0:0").await.unwrap()); let audio_control: Arc = Arc::new(AudioControl { - active_input_audio_streams_producers: active_input_audio_streams_producers.clone(), - active_input_audio_streams_consumers: active_input_audio_streams_consumers.clone(), - active_input_audio_streams_encoders: active_input_audio_streams_encoders.clone(), active_input_audio_streams_info: active_input_audio_streams_info.clone(), udp_socket: udp_socket.clone(), }); @@ -87,9 +73,6 @@ async fn main() { tokio::spawn(audio_stream_receiver( output_device, output_device_config.into(), - active_input_audio_streams_producers.clone(), - active_input_audio_streams_consumers.clone(), - active_input_audio_streams_encoders.clone(), active_input_audio_streams_info.clone(), udp_socket.clone(), )); diff --git a/client/src/models.rs b/client/src/models.rs index 0d03a59..adfaedd 100644 --- a/client/src/models.rs +++ b/client/src/models.rs @@ -1,6 +1,7 @@ use clap::Parser; use ringbuf::{HeapRb, wrap::caching::Caching}; use ropus::Decoder; +use rubato::Fft; use shared::AudioStreamInfo; use std::{ collections::HashMap, @@ -39,21 +40,19 @@ pub struct ClientInfo { pub connected: bool, } -type AudioStreamKey = (u32, u8); +pub type AudioStreamKey = (u32, u8); pub type InputAudioStreamsProducers = - Arc>, true, false>>>>; + HashMap>, true, false>>; pub type InputAudioStreamsConsumers = Arc>, false, true>>>>; -pub type InputAudioStreamsEncoders = Arc>>>; +pub type InputAudioStreamsDecoders = HashMap; +pub type InputAudioStreamsResamplers = HashMap>; pub type InputAudioStreamsInfo = Arc>>; pub struct AudioControl { - pub active_input_audio_streams_producers: InputAudioStreamsProducers, - pub active_input_audio_streams_consumers: InputAudioStreamsConsumers, - pub active_input_audio_streams_encoders: InputAudioStreamsEncoders, pub active_input_audio_streams_info: InputAudioStreamsInfo, pub udp_socket: Arc, } diff --git a/client/src/tcp_connection.rs b/client/src/tcp_connection.rs index 159dd72..d58a03d 100644 --- a/client/src/tcp_connection.rs +++ b/client/src/tcp_connection.rs @@ -1,8 +1,6 @@ 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 std::sync::{Arc, RwLock}; use tokio::{ net::TcpStream, sync::broadcast::{Receiver, Sender}, @@ -57,49 +55,10 @@ pub async fn tcp_connection( 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 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(); - - 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 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, diff --git a/client/src/tui.rs b/client/src/tui.rs index 671eb36..db41705 100644 --- a/client/src/tui.rs +++ b/client/src/tui.rs @@ -345,15 +345,6 @@ impl DeviceSelectScreen { tokio::spawn(audio_stream_receiver( output_device.clone(), output_config.clone(), - self.audio_control - .active_input_audio_streams_producers - .clone(), - self.audio_control - .active_input_audio_streams_consumers - .clone(), - self.audio_control - .active_input_audio_streams_encoders - .clone(), self.audio_control.active_input_audio_streams_info.clone(), self.audio_control.udp_socket.clone(), )); -- 2.54.0 From 9ee645c1444023d9c15b60690433d1f95478483b Mon Sep 17 00:00:00 2001 From: Filip Kojro Date: Tue, 20 Oct 2026 14:22:36 +0200 Subject: [PATCH 12/16] clippy fix --- client/src/audio_streams.rs | 46 ++++++++++++++++++------------------ client/src/main.rs | 4 ++-- client/src/tcp_connection.rs | 4 ++-- client/src/tui.rs | 37 +++++++++++++---------------- server/src/main.rs | 25 +++++++++----------- 5 files changed, 54 insertions(+), 62 deletions(-) diff --git a/client/src/audio_streams.rs b/client/src/audio_streams.rs index f223674..c3fc12e 100644 --- a/client/src/audio_streams.rs +++ b/client/src/audio_streams.rs @@ -41,11 +41,8 @@ where for sample in data { let mut mixed_value: f32 = 0.0; - for (_key, consumer) in streams.iter_mut() { - mixed_value += match consumer.try_pop() { - Some(s) => s, - None => 0.0, - } + for consumer in streams.values_mut() { + mixed_value += consumer.try_pop().unwrap_or(0.0) } *sample = mixed_value.to_sample::(); @@ -53,9 +50,9 @@ where }; 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) - .expect("building output stream failed"); + .expect("building output stream failed") } 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); } - if !active_input_audio_streams_resamplers.contains_key(&(client_id, stream_id)) { - let resampler = Fft::::new( - stream_info.sample_rate as usize, - output_device_config.sample_rate() as usize, - input_frames, - stream_info.channels as usize, - FixedSync::Both, - ) - .unwrap(); - active_input_audio_streams_resamplers.insert((client_id, stream_id), resampler); - } + active_input_audio_streams_resamplers + .entry((client_id, stream_id)) + .or_insert_with(|| { + Fft::::new( + stream_info.sample_rate as usize, + output_device_config.sample_rate() as usize, + input_frames, + stream_info.channels as usize, + FixedSync::Both, + ) + .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_samples = ring_buffer_frames as usize * stream_info.channels as usize; let ring_buffer = HeapRb::::new(ring_buffer_samples); 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 .lock() .unwrap() @@ -294,9 +294,9 @@ where }; 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) - .expect("building input stream failed"); + .expect("building input stream failed") } pub async fn audio_stream_spawner( @@ -317,7 +317,7 @@ pub async fn audio_stream_spawner( }; if connected { let _ = global_event_tx.send(ClientEvent::NewAudioStreamRequest(AudioStreamInfo { - client_id: client_id, + client_id, stream_id: current_stream_id, channels: force_channels, sample_rate: SAMPLE_RATE_48K, diff --git a/client/src/main.rs b/client/src/main.rs index 5c944e3..3f5396e 100644 --- a/client/src/main.rs +++ b/client/src/main.rs @@ -72,14 +72,14 @@ async fn main() { tokio::spawn(audio_stream_receiver( output_device, - output_device_config.into(), + output_device_config, active_input_audio_streams_info.clone(), udp_socket.clone(), )); tokio::spawn(audio_stream_spawner( input_device, - input_device_config.into(), + input_device_config, global_event_tx.clone(), global_event_tx.subscribe(), udp_socket.clone(), diff --git a/client/src/tcp_connection.rs b/client/src/tcp_connection.rs index d58a03d..33cd159 100644 --- a/client/src/tcp_connection.rs +++ b/client/src/tcp_connection.rs @@ -94,7 +94,7 @@ pub async fn tcp_connection( }) .expect("failed to parse initial request"); - let _ = framed_tcp_stream.send(Bytes::from(request_json)) + framed_tcp_stream.send(Bytes::from(request_json)) .await .expect("failed to send initial message"); @@ -110,7 +110,7 @@ pub async fn tcp_connection( } }).expect("failed to parse initial request"); - let _ = framed_tcp_stream.send(Bytes::from(request_json)) + framed_tcp_stream.send(Bytes::from(request_json)) .await .expect("failed to send initial message"); } diff --git a/client/src/tui.rs b/client/src/tui.rs index db41705..54cf81e 100644 --- a/client/src/tui.rs +++ b/client/src/tui.rs @@ -75,7 +75,7 @@ impl BlankScreen { self.handle_events(); } - return TuiScreen::Exit; + TuiScreen::Exit } fn draw(&mut self, frame: &mut Frame) { @@ -85,10 +85,7 @@ impl BlankScreen { fn handle_events(&mut self) { match event::read().expect("failed to read event") { Event::Key(key_event) if key_event.kind == KeyEventKind::Press => { - match key_event.code { - KeyCode::Char('q') => self.exit = true, - _ => {} - } + if let KeyCode::Char('q') = key_event.code { self.exit = true } } _ => {} } @@ -121,11 +118,10 @@ impl DeviceSelectScreen { let mut input_devices: Vec<(Device, SupportedStreamConfig)> = Vec::new(); - if let Some(default_input_device) = host.default_input_device() { - if let Ok(default_config) = default_input_device.default_input_config() { + if let Some(default_input_device) = host.default_input_device() + && let Ok(default_config) = default_input_device.default_input_config() { input_devices.push((default_input_device.clone(), default_config)); } - } if let Ok(possible_input_devices) = host.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(); - if let Some(default_output_device) = host.default_output_device() { - if let Ok(default_config) = default_output_device.default_output_config() { - output_devices.push((default_output_device.clone(), default_config.into())); + if let Some(default_output_device) = host.default_output_device() + && let Ok(default_config) = default_output_device.default_output_config() { + output_devices.push((default_output_device.clone(), default_config)); } - } if let Ok(possible_output_devices) = host.output_devices() { for possible_output_device in possible_output_devices { 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, input_device_state: input_state, output_device_state: output_state, - input_devices: input_devices, - output_devices: output_devices, + input_devices, + output_devices, selected_selection: 0, - audio_control: audio_control, - global_event_tx: global_event_tx, - client_info: client_info, + audio_control, + global_event_tx, + client_info, } } @@ -180,7 +175,7 @@ impl DeviceSelectScreen { self.handle_events(); } - return self.exit_screen; + self.exit_screen } fn draw(&mut self, frame: &mut Frame) { @@ -344,14 +339,14 @@ impl DeviceSelectScreen { tokio::spawn(audio_stream_receiver( output_device.clone(), - output_config.clone(), + *output_config, self.audio_control.active_input_audio_streams_info.clone(), self.audio_control.udp_socket.clone(), )); tokio::spawn(audio_stream_spawner( input_device.clone(), - input_config.clone(), + *input_config, self.global_event_tx.clone(), self.global_event_tx.subscribe(), self.audio_control.udp_socket.clone(), diff --git a/server/src/main.rs b/server/src/main.rs index 223e93a..43540af 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -35,13 +35,13 @@ type State = Arc>>; fn scrape_all_client_streams_info(clients: &HashMap) -> Vec { let mut all_audio_streams_info: Vec = Vec::new(); - for (_, data) in clients { + for data in clients.values() { if let Some(audio_streams_info) = &data.audio_streams { all_audio_streams_info.extend(audio_streams_info.iter().cloned()); } } - return all_audio_streams_info; + all_audio_streams_info } #[tokio::main] @@ -84,11 +84,10 @@ async fn main() { } for (&other_client_id, other_client) in clients.iter() { - if other_client_id != client_id { - if let Some(destination_udp) = other_client.udp_addr { + if other_client_id != client_id + && let Some(destination_udp) = other_client.udp_addr { let _ = udp_socket.send_to(&buf[..len], destination_udp).await; } - } } } }); @@ -136,7 +135,7 @@ async fn main() { }, }).expect("failed to serialize TcpRequestData::NewAudioStreamAccepted"); - let _ = framed_tcp_stream.send(Bytes::from(response_json)) + framed_tcp_stream.send(Bytes::from(response_json)) .await .expect("failed to send message"); @@ -145,9 +144,9 @@ async fn main() { 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()); + audio_streams.push(audio_stream_info); } 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)); } 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) { + if let Some(client) = tcp_state.write().await.get_mut(¤t_client_id) + && let Some(audio_streams) = &mut client.audio_streams + && let Some(idx) = audio_streams.iter().position(|s| s.stream_id == audio_stream_id) { audio_streams.remove(idx); } - } - } 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 { status: ResponseStatus::Ok, data: shared::TcpResponseData::UserDisconnected { - user_id: user_id + user_id }, }).expect("failed to serialize ServerEvent::UserDisconnected"); -- 2.54.0 From f11cd939ccb690443fd402331af0a1017abbd759 Mon Sep 17 00:00:00 2001 From: Filip Kojro Date: Tue, 20 Oct 2026 14:26:21 +0200 Subject: [PATCH 13/16] fmt fix --- client/src/tui.rs | 18 +++++++++++------- server/src/main.rs | 7 ++++--- 2 files changed, 15 insertions(+), 10 deletions(-) diff --git a/client/src/tui.rs b/client/src/tui.rs index 54cf81e..03a8210 100644 --- a/client/src/tui.rs +++ b/client/src/tui.rs @@ -85,7 +85,9 @@ impl BlankScreen { fn handle_events(&mut self) { match event::read().expect("failed to read event") { Event::Key(key_event) if key_event.kind == KeyEventKind::Press => { - if let KeyCode::Char('q') = key_event.code { self.exit = true } + if let KeyCode::Char('q') = key_event.code { + self.exit = true + } } _ => {} } @@ -119,9 +121,10 @@ impl DeviceSelectScreen { let mut input_devices: Vec<(Device, SupportedStreamConfig)> = Vec::new(); if let Some(default_input_device) = host.default_input_device() - && let Ok(default_config) = default_input_device.default_input_config() { - input_devices.push((default_input_device.clone(), default_config)); - } + && let Ok(default_config) = default_input_device.default_input_config() + { + input_devices.push((default_input_device.clone(), default_config)); + } if let Ok(possible_input_devices) = host.input_devices() { for possible_input_device in possible_input_devices { @@ -134,9 +137,10 @@ impl DeviceSelectScreen { let mut output_devices: Vec<(Device, SupportedStreamConfig)> = Vec::new(); if let Some(default_output_device) = host.default_output_device() - && let Ok(default_config) = default_output_device.default_output_config() { - output_devices.push((default_output_device.clone(), default_config)); - } + && let Ok(default_config) = default_output_device.default_output_config() + { + output_devices.push((default_output_device.clone(), default_config)); + } if let Ok(possible_output_devices) = host.output_devices() { for possible_output_device in possible_output_devices { diff --git a/server/src/main.rs b/server/src/main.rs index 43540af..0db195c 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -85,9 +85,10 @@ async fn main() { for (&other_client_id, other_client) in clients.iter() { if other_client_id != client_id - && let Some(destination_udp) = other_client.udp_addr { - let _ = udp_socket.send_to(&buf[..len], destination_udp).await; - } + && let Some(destination_udp) = other_client.udp_addr + { + let _ = udp_socket.send_to(&buf[..len], destination_udp).await; + } } } }); -- 2.54.0 From 6eb4f47aad4cf364bd8762a9c8019b27ab4d2bad Mon Sep 17 00:00:00 2001 From: Filip Kojro Date: Sat, 24 Oct 2026 23:28:24 +0200 Subject: [PATCH 14/16] better handling lost packets --- client/src/audio_streams.rs | 292 ++++++++++++++++++++++------------- client/src/tcp_connection.rs | 14 +- server/src/main.rs | 2 +- 3 files changed, 193 insertions(+), 115 deletions(-) diff --git a/client/src/audio_streams.rs b/client/src/audio_streams.rs index c3fc12e..68069f6 100644 --- a/client/src/audio_streams.rs +++ b/client/src/audio_streams.rs @@ -15,6 +15,7 @@ use rubato::{Fft, FixedSync, Resampler, audioadapter::Adapter}; use shared::AudioStreamInfo; use std::{ collections::HashMap, + num::Wrapping, sync::{Arc, Mutex, RwLock}, }; use tokio::{ @@ -23,7 +24,7 @@ use tokio::{ }; use crate::models::{ - ClientEvent, ClientInfo, InputAudioStreamsConsumers, InputAudioStreamsDecoders, + AudioStreamKey, ClientEvent, ClientInfo, InputAudioStreamsConsumers, InputAudioStreamsDecoders, InputAudioStreamsInfo, InputAudioStreamsProducers, InputAudioStreamsResamplers, }; @@ -55,6 +56,120 @@ where .expect("building output stream failed") } +fn decode_sample( + client_id: u32, + stream_id: u8, + stream_info: AudioStreamInfo, + output_device_config: SupportedStreamConfig, + active_input_audio_streams_decoders: &mut InputAudioStreamsDecoders, + active_input_audio_streams_resamplers: &mut InputAudioStreamsResamplers, + active_input_audio_streams_producers: &mut InputAudioStreamsProducers, + active_input_audio_streams_consumers: &mut InputAudioStreamsConsumers, + audio_packet: &[u8], + decoded_samples: &mut Vec, + resampled_samples: &mut Vec, + decoded_samples_converted: &mut Vec, + decode_mode: ropus::DecodeMode, +) { + let input_frames = ((20.0 / 1000.0) * stream_info.sample_rate as f32) as usize; + let resampler_output_frames = + ((20.0 / 1000.0) * output_device_config.sample_rate() as f32) as usize; + + if !active_input_audio_streams_decoders.contains_key(&(client_id, stream_id)) { + let channels: ropus::Channels = { + if stream_info.channels == 2 { + ropus::Channels::Stereo + } else { + ropus::Channels::Mono + } + }; + + let decoder = + Decoder::new(stream_info.sample_rate, channels).expect("failed to create decoder"); + active_input_audio_streams_decoders + .insert((stream_info.client_id, stream_info.stream_id), decoder); + } + + active_input_audio_streams_resamplers + .entry((client_id, stream_id)) + .or_insert_with(|| { + Fft::::new( + stream_info.sample_rate as usize, + output_device_config.sample_rate() as usize, + input_frames, + stream_info.channels as usize, + FixedSync::Both, + ) + .unwrap() + }); + + 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_samples = ring_buffer_frames as usize * stream_info.channels as usize; + let ring_buffer = HeapRb::::new(ring_buffer_samples); + let (producer, consumer) = ring_buffer.split(); + + e.insert(producer); + active_input_audio_streams_consumers + .lock() + .unwrap() + .insert((client_id, stream_id), consumer); + } + + let decoder = active_input_audio_streams_decoders + .get_mut(&(client_id, stream_id)) + .expect("failed to get decoder"); + + let _ = decoder + .decode_float(audio_packet, decoded_samples, decode_mode) + .expect("failed to decode opus"); + + let decoded_samples_adapter = InterleavedSlice::new( + &decoded_samples, + stream_info.channels as usize, + input_frames, + ) + .unwrap(); + + let mut resampled_samples_adapter = InterleavedSlice::new_mut( + resampled_samples, + stream_info.channels as usize, + resampler_output_frames, + ) + .unwrap(); + + let (_, frames_out) = active_input_audio_streams_resamplers + .get_mut(&(client_id, stream_id)) + .expect("failed to get resampler") + .process_into_buffer( + &decoded_samples_adapter, + &mut resampled_samples_adapter, + None, + ) + .expect("error during resampling"); + + if let Some(producer) = active_input_audio_streams_producers.get_mut(&(client_id, stream_id)) { + for (&resampled_sample, output_frame) in resampled_samples + .iter() + .step_by(stream_info.channels as usize) + .zip( + decoded_samples_converted + .chunks_exact_mut(output_device_config.channels() as usize), + ) + { + for output_sample in output_frame.iter_mut() { + *output_sample = resampled_sample; + } + } + + producer.push_slice( + &decoded_samples_converted[..frames_out * output_device_config.channels() as usize], + ); + } +} + pub async fn audio_stream_receiver( output_device: Device, output_device_config: SupportedStreamConfig, @@ -67,6 +182,8 @@ pub async fn audio_stream_receiver( let mut active_input_audio_streams_decoders: InputAudioStreamsDecoders = HashMap::new(); let mut active_input_audio_streams_resamplers: InputAudioStreamsResamplers = HashMap::new(); + let mut last_frame_number: HashMap> = HashMap::new(); + let output_stream = match output_device_config.sample_format() { SampleFormat::F64 => cpal_build_output_stream::( &output_device, @@ -141,17 +258,20 @@ pub async fn audio_stream_receiver( loop { match udp_socket.recv_from(&mut receive_buffer).await { Ok((len, _)) => { - if len < 4 { + if len < 5 { continue; } let client_id_bytes: [u8; 4] = receive_buffer[0..4].try_into().unwrap(); - let client_id = u32::from_be_bytes(client_id_bytes); + let client_id = u32::from_le_bytes(client_id_bytes); let stream_id_bytes = receive_buffer[4]; - let stream_id = u8::from_be(stream_id_bytes); + let stream_id = u8::from_le(stream_id_bytes); - let audio_packet = &receive_buffer[5..len]; + let frame_number_bytes = receive_buffer[5]; + let frame_number = Wrapping(u8::from_le(frame_number_bytes)); + + let audio_packet = &receive_buffer[6..len]; let stream_info = { if let Some(&info) = active_input_audio_streams_info @@ -165,110 +285,63 @@ pub async fn audio_stream_receiver( } }; - let input_frames = ((20.0 / 1000.0) * stream_info.sample_rate as f32) as usize; - let resampler_output_frames = - ((20.0 / 1000.0) * output_device_config.sample_rate() as f32) as usize; - - if !active_input_audio_streams_decoders.contains_key(&(client_id, stream_id)) { - let channels: ropus::Channels = { - if stream_info.channels == 2 { - ropus::Channels::Stereo - } else { - ropus::Channels::Mono - } - }; - - let decoder = Decoder::new(stream_info.sample_rate, channels) - .expect("failed to create decoder"); - active_input_audio_streams_decoders - .insert((stream_info.client_id, stream_info.stream_id), decoder); + if !last_frame_number.contains_key(&(client_id, stream_id)) { + let last_frame = frame_number - Wrapping(1); + last_frame_number.insert((client_id, stream_id), last_frame); } - active_input_audio_streams_resamplers - .entry((client_id, stream_id)) - .or_insert_with(|| { - Fft::::new( - stream_info.sample_rate as usize, - output_device_config.sample_rate() as usize, - input_frames, - stream_info.channels as usize, - FixedSync::Both, - ) - .unwrap() - }); + let last_frame = last_frame_number.get(&(client_id, stream_id)).unwrap(); - 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_samples = - ring_buffer_frames as usize * stream_info.channels as usize; - let ring_buffer = HeapRb::::new(ring_buffer_samples); - let (producer, consumer) = ring_buffer.split(); - - e.insert(producer); - active_input_audio_streams_consumers - .lock() - .unwrap() - .insert((client_id, stream_id), consumer); - } - - let decoder = active_input_audio_streams_decoders - .get_mut(&(client_id, stream_id)) - .expect("failed to get decoder"); - - let _ = decoder - .decode_float( + if last_frame + Wrapping(1) == frame_number { + decode_sample( + client_id, + stream_id, + stream_info, + output_device_config, + &mut active_input_audio_streams_decoders, + &mut active_input_audio_streams_resamplers, + &mut active_input_audio_streams_producers, + &mut active_input_audio_streams_consumers.clone(), audio_packet, &mut decoded_samples, + &mut resampled_samples, + &mut decoded_samples_converted, ropus::DecodeMode::Normal, - ) - .expect("failed to decode opus"); - - let decoded_samples_adapter = InterleavedSlice::new( - &decoded_samples, - stream_info.channels as usize, - input_frames, - ) - .unwrap(); - - let mut resampled_samples_adapter = InterleavedSlice::new_mut( - &mut resampled_samples, - stream_info.channels as usize, - resampler_output_frames, - ) - .unwrap(); - - let (_, frames_out) = active_input_audio_streams_resamplers - .get_mut(&(client_id, stream_id)) - .expect("failed to get resampler") - .process_into_buffer( - &decoded_samples_adapter, - &mut resampled_samples_adapter, - None, - ) - .expect("error during resampling"); - - if let Some(producer) = - active_input_audio_streams_producers.get_mut(&(client_id, stream_id)) - { - for (&resampled_sample, output_frame) in resampled_samples - .iter() - .step_by(stream_info.channels as usize) - .zip( - decoded_samples_converted - .chunks_exact_mut(output_device_config.channels() as usize), - ) - { - for output_sample in output_frame.iter_mut() { - *output_sample = resampled_sample; - } - } - - producer.push_slice( - &decoded_samples_converted - [..frames_out * output_device_config.channels() as usize], ); + + last_frame_number.insert((client_id, stream_id), frame_number); + } else if last_frame + Wrapping(2) <= frame_number { + decode_sample( + client_id, + stream_id, + stream_info, + output_device_config, + &mut active_input_audio_streams_decoders, + &mut active_input_audio_streams_resamplers, + &mut active_input_audio_streams_producers, + &mut active_input_audio_streams_consumers.clone(), + audio_packet, + &mut decoded_samples, + &mut resampled_samples, + &mut decoded_samples_converted, + ropus::DecodeMode::Fec, + ); + decode_sample( + client_id, + stream_id, + stream_info, + output_device_config, + &mut active_input_audio_streams_decoders, + &mut active_input_audio_streams_resamplers, + &mut active_input_audio_streams_producers, + &mut active_input_audio_streams_consumers.clone(), + audio_packet, + &mut decoded_samples, + &mut resampled_samples, + &mut decoded_samples_converted, + ropus::DecodeMode::Normal, + ); + last_frame_number.insert((client_id, stream_id), frame_number); } } Err(e) => eprintln!("failed to receive udp packet: {}", e), @@ -331,7 +404,7 @@ pub async fn audio_stream_spawner( Ok(ClientEvent::NewAudioStreamAccepted(stream_id)) if stream_id == current_stream_id => { - println!("new stream accepted: {}", stream_id); + // println!("new stream accepted: {}", stream_id); break; } Ok(_) => {} @@ -418,7 +491,9 @@ pub async fn audio_stream_spawner( .build() .unwrap(); - let mut encoded_frame_buffer = [0u8; 2048]; + let mut encoded_frame_buffer = vec![0u8; 2048]; + + let mut frame_num = Wrapping(0u8); loop { if consumer.occupied_len() >= samples_needed { @@ -460,15 +535,18 @@ pub async fn audio_stream_spawner( ) .expect("failed to encode frame"); - let mut payload = Vec::::with_capacity(4 + 1 + encoded_len * 4); + let mut payload = Vec::::with_capacity(1 + 4 + 1 + encoded_len * 4); let (server_address, client_id) = { let info = client_info.read().unwrap(); (info.server_address.clone(), info.client_id) }; - payload.extend(client_id.to_be_bytes()); - payload.extend(current_stream_id.to_be_bytes()); + payload.extend(client_id.to_le_bytes()); + payload.extend(current_stream_id.to_le_bytes()); + + payload.extend(frame_num.0.to_le_bytes()); + frame_num += Wrapping(1); payload.extend_from_slice(&encoded_frame_buffer[..encoded_len]); diff --git a/client/src/tcp_connection.rs b/client/src/tcp_connection.rs index 33cd159..35d05e1 100644 --- a/client/src/tcp_connection.rs +++ b/client/src/tcp_connection.rs @@ -34,25 +34,25 @@ pub async fn tcp_connection( frame_result = framed_tcp_stream.next() => { match frame_result { None => { - println!("server shut down"); + // println!("server shut down"); break; } Some(Ok(bytes)) => { match serde_json::from_slice::(&bytes) { Ok(response) => { - println!("connection Ok: {}", response.status == ResponseStatus::Ok); + // 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); + // println!("fetched client id: {}", client_id); } TcpResponseData::UserConnected { user_id } => { - println!("new user connected: {}", user_id); + // println!("new user connected: {}", user_id); } TcpResponseData::UserDisconnected { user_id } => { - println!("user disconnected: {}", user_id); + // println!("user disconnected: {}", user_id); } TcpResponseData::AudioStreamsFullUpdate { audio_stream_info } => { let mut info_map = audio_control.active_input_audio_streams_info.write().unwrap(); @@ -64,7 +64,7 @@ pub async fn tcp_connection( audio_stream, ); } - println!("audio streams update"); + // println!("audio streams update"); } TcpResponseData::NewAudioStreamAccepted { audio_stream_id } => { let _ = global_event_tx.send(ClientEvent::NewAudioStreamAccepted(audio_stream_id)); @@ -100,7 +100,7 @@ pub async fn tcp_connection( } ClientEvent::NewAudioStreamRequest(audio_stream_info) => { - println!("new audio stream request"); + // println!("new audio stream request"); let session_client_id = client_info.read().unwrap().client_id; let request_json = serde_json::to_string(&TcpRequest { diff --git a/server/src/main.rs b/server/src/main.rs index 0db195c..69b2ac0 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -76,7 +76,7 @@ async fn main() { } let id_bytes: [u8; 4] = buf[0..4].try_into().unwrap(); - let client_id = u32::from_be_bytes(id_bytes); + let client_id = u32::from_le_bytes(id_bytes); let mut clients = udp_state.write().await; if let Some(client) = clients.get_mut(&client_id) { -- 2.54.0 From 8f0a80a8b9c37aa15bb1ff85043c7778cbea9c84 Mon Sep 17 00:00:00 2001 From: Filip Kojro Date: Mon, 26 Oct 2026 18:49:46 +0100 Subject: [PATCH 15/16] better delta --- client/src/audio_streams.rs | 46 +++++++++++++++++++------------------ 1 file changed, 24 insertions(+), 22 deletions(-) diff --git a/client/src/audio_streams.rs b/client/src/audio_streams.rs index 68069f6..67cc28f 100644 --- a/client/src/audio_streams.rs +++ b/client/src/audio_streams.rs @@ -258,7 +258,7 @@ pub async fn audio_stream_receiver( loop { match udp_socket.recv_from(&mut receive_buffer).await { Ok((len, _)) => { - if len < 5 { + if len < 6 { continue; } @@ -290,27 +290,10 @@ pub async fn audio_stream_receiver( last_frame_number.insert((client_id, stream_id), last_frame); } - let last_frame = last_frame_number.get(&(client_id, stream_id)).unwrap(); + let frame_delta = + (frame_number - last_frame_number.get(&(client_id, stream_id)).unwrap()).0; - if last_frame + Wrapping(1) == frame_number { - decode_sample( - client_id, - stream_id, - stream_info, - output_device_config, - &mut active_input_audio_streams_decoders, - &mut active_input_audio_streams_resamplers, - &mut active_input_audio_streams_producers, - &mut active_input_audio_streams_consumers.clone(), - audio_packet, - &mut decoded_samples, - &mut resampled_samples, - &mut decoded_samples_converted, - ropus::DecodeMode::Normal, - ); - - last_frame_number.insert((client_id, stream_id), frame_number); - } else if last_frame + Wrapping(2) <= frame_number { + if frame_delta == 2 { decode_sample( client_id, stream_id, @@ -341,8 +324,25 @@ pub async fn audio_stream_receiver( &mut decoded_samples_converted, ropus::DecodeMode::Normal, ); - last_frame_number.insert((client_id, stream_id), frame_number); + } else { + decode_sample( + client_id, + stream_id, + stream_info, + output_device_config, + &mut active_input_audio_streams_decoders, + &mut active_input_audio_streams_resamplers, + &mut active_input_audio_streams_producers, + &mut active_input_audio_streams_consumers.clone(), + audio_packet, + &mut decoded_samples, + &mut resampled_samples, + &mut decoded_samples_converted, + ropus::DecodeMode::Normal, + ); } + + last_frame_number.insert((client_id, stream_id), frame_number); } Err(e) => eprintln!("failed to receive udp packet: {}", e), } @@ -488,6 +488,8 @@ pub async fn audio_stream_spawner( let mut opus_encoder = Encoder::builder(SAMPLE_RATE_48K, encoder_channels, ropus::Application::Voip) + .inband_fec(ropus::InbandFec::Enabled) + .packet_loss_perc(10) .build() .unwrap(); -- 2.54.0 From f5390f77658141d033b99610b9ad931ee3b4dba1 Mon Sep 17 00:00:00 2001 From: Filip Kojro Date: Fri, 30 Oct 2026 20:40:23 +0100 Subject: [PATCH 16/16] removing clippy (i dont like it) --- .gitea/workflows/build.yaml | 9 --------- client/src/tcp_connection.rs | 6 +++--- 2 files changed, 3 insertions(+), 12 deletions(-) diff --git a/.gitea/workflows/build.yaml b/.gitea/workflows/build.yaml index bb13c97..0512587 100644 --- a/.gitea/workflows/build.yaml +++ b/.gitea/workflows/build.yaml @@ -28,15 +28,6 @@ jobs: - run: rustup component add rustfmt - run: cargo fmt --all -- --check - clippy: - name: Clippy - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v4 - - uses: ./.gitea/actions/setup-env - - run: rustup component add clippy - - run: cargo clippy -- -D warnings - build: name: build runs-on: ubuntu-latest diff --git a/client/src/tcp_connection.rs b/client/src/tcp_connection.rs index 35d05e1..92daf4c 100644 --- a/client/src/tcp_connection.rs +++ b/client/src/tcp_connection.rs @@ -1,5 +1,5 @@ use futures::{SinkExt, StreamExt}; -use shared::{ResponseStatus, TcpRequest, TcpResponse, TcpResponseData}; +use shared::{TcpRequest, TcpResponse, TcpResponseData}; use std::sync::{Arc, RwLock}; use tokio::{ net::TcpStream, @@ -48,10 +48,10 @@ pub async fn tcp_connection( client_info_var.connected = true; // println!("fetched client id: {}", client_id); } - TcpResponseData::UserConnected { user_id } => { + TcpResponseData::UserConnected { .. } => { // println!("new user connected: {}", user_id); } - TcpResponseData::UserDisconnected { user_id } => { + TcpResponseData::UserDisconnected { .. } => { // println!("user disconnected: {}", user_id); } TcpResponseData::AudioStreamsFullUpdate { audio_stream_info } => { -- 2.54.0