readme update

Reviewed-on: #1

.gitea/actions/setup-env/action.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

.gitea/workflows/build.yaml

@@ -0,0 +1,37 @@
+on: [push, pull_request]
+
+name: Continuous integration
+
+jobs:
+  check:
+    name: Check
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: ./.gitea/actions/setup-env
+      - run: cargo check
+
+  test:
+    name: Test Suite
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: ./.gitea/actions/setup-env
+      - run: cargo test
+
+  fmt:
+    name: Rustfmt
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: ./.gitea/actions/setup-env
+      - run: rustup component add rustfmt
+      - run: cargo fmt --all -- --check
+
+  build:
+    name: build
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: ./.gitea/actions/setup-env
+      - run: cargo build

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"

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"]

README.md

@@ -1,26 +1,37 @@
 # smalltalk
 
-## build and run client
+very lightweight voice communicator written in Rust
 
-Usage: 
+***Don't use this project for secret conversations as it don't use encryption yet!***
+
+## Client
+
+### run using cargo
 
 ```bash
 cargo run -p client --release -- --server-ip <SERVER_IP>
 ```
 
 Options:
-
 ```
--s, --server-ip <SERVER_IP>  
+-c, --connect-with-defaults <CONNECT_WITH_DEFAULTS>
--h, --help                   Print help
+-s, --server-ip <SERVER_IP>
--V, --version                Print version
+-h, --help                                           Print help
+-V, --version                                        Print version
 ```
 
-## build and run server
+## Server
 
-before, open port `8080` tcp/udp
+### run using docker
 
-Usage:
+```bash
+docker build -t smalltalk .
+docker run -d --rm -p 6767:6767/tcp -p 6767:6767/udp --name smalltalk smalltalk
+```
+
+### run using cargo
+
+before, open port `6767` tcp and udp
 
 ```bash
 cargo run -p server --release -- [OPTIONS]
@@ -29,17 +40,7 @@ cargo run -p server --release -- [OPTIONS]
 Options:
 
 ```
--s, --server-ip <SERVER_IP>  [default: 0.0.0.0:8080]
+-s, --server-ip <SERVER_IP>  [default: 0.0.0.0:6767]
 -h, --help                   Print help
 -V, --version                Print version
 ```
-
----
-
-I think it will be based on:
-
-- `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

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"

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::{
@@ -8,17 +10,22 @@ use ringbuf::{
     traits::{Consumer, Observer, Producer, Split},
     wrap::caching::Caching,
 };
-use ropus::Encoder;
+use ropus::{Decoder, Encoder};
+use rubato::{Fft, FixedSync, Resampler, audioadapter::Adapter};
 use shared::AudioStreamInfo;
-use std::sync::{Arc, RwLock};
+use std::{
+    collections::HashMap,
+    num::Wrapping,
+    sync::{Arc, Mutex, RwLock},
+};
 use tokio::{
     net::UdpSocket,
     sync::broadcast::{Receiver, Sender},
 };
 
 use crate::models::{
-    ClientEvent, ClientInfo, InputAudioStreamsConsumers, InputAudioStreamsEncoders,
+    AudioStreamKey, ClientEvent, ClientInfo, InputAudioStreamsConsumers, InputAudioStreamsDecoders,
-    InputAudioStreamsInfo, InputAudioStreamsProducers,
+    InputAudioStreamsInfo, InputAudioStreamsProducers, InputAudioStreamsResamplers,
 };
 
 fn cpal_build_output_stream<T>(
@@ -35,11 +42,8 @@ where
         for sample in data {
             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>();
@@ -47,70 +51,189 @@ 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")
+}
+
+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<f32>,
+    resampled_samples: &mut Vec<f32>,
+    decoded_samples_converted: &mut Vec<f32>,
+    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::<f32>::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::<f32>::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,
-    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<UdpSocket>,
 ) {
+    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 mut last_frame_number: HashMap<AudioStreamKey, Wrapping<u8>> = HashMap::new();
+
     let output_stream = match output_device_config.sample_format() {
         SampleFormat::F64 => cpal_build_output_stream::<f64>(
             &output_device,
             output_device_config.into(),
-            active_input_audio_streams_consumers,
+            active_input_audio_streams_consumers.clone(),
         ),
         SampleFormat::I64 => cpal_build_output_stream::<i64>(
             &output_device,
             output_device_config.into(),
-            active_input_audio_streams_consumers,
+            active_input_audio_streams_consumers.clone(),
         ),
         SampleFormat::U64 => cpal_build_output_stream::<u64>(
             &output_device,
             output_device_config.into(),
-            active_input_audio_streams_consumers,
+            active_input_audio_streams_consumers.clone(),
         ),
         SampleFormat::F32 => cpal_build_output_stream::<f32>(
             &output_device,
             output_device_config.into(),
-            active_input_audio_streams_consumers,
+            active_input_audio_streams_consumers.clone(),
         ),
         SampleFormat::I32 => cpal_build_output_stream::<i32>(
             &output_device,
             output_device_config.into(),
-            active_input_audio_streams_consumers,
+            active_input_audio_streams_consumers.clone(),
         ),
         SampleFormat::U32 => cpal_build_output_stream::<u32>(
             &output_device,
             output_device_config.into(),
-            active_input_audio_streams_consumers,
+            active_input_audio_streams_consumers.clone(),
         ),
         SampleFormat::I16 => cpal_build_output_stream::<i16>(
             &output_device,
             output_device_config.into(),
-            active_input_audio_streams_consumers,
+            active_input_audio_streams_consumers.clone(),
         ),
         SampleFormat::U16 => cpal_build_output_stream::<u16>(
             &output_device,
             output_device_config.into(),
-            active_input_audio_streams_consumers,
+            active_input_audio_streams_consumers.clone(),
         ),
         SampleFormat::I8 => cpal_build_output_stream::<i8>(
             &output_device,
             output_device_config.into(),
-            active_input_audio_streams_consumers,
+            active_input_audio_streams_consumers.clone(),
         ),
         SampleFormat::U8 => cpal_build_output_stream::<u8>(
             &output_device,
             output_device_config.into(),
-            active_input_audio_streams_consumers,
+            active_input_audio_streams_consumers.clone(),
         ),
         _ => {
             eprintln!("unsupported output stream format :(");
@@ -122,23 +245,33 @@ pub async fn audio_stream_receiver(
 
     let mut receive_buffer = vec![0u8; 4096];
 
-    let mut decoded_samples = [0.0f32; 1024];
+    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_converted = [0.0f32; 2048];
+    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 {
             Ok((len, _)) => {
-                if len < 4 {
+                if len < 6 {
                     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
@@ -152,58 +285,64 @@ pub async fn audio_stream_receiver(
                     }
                 };
 
-                let decoders_map = active_input_audio_streams_encoders.read().unwrap();
+                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);
+                }
 
-                let decoder_mutex = decoders_map
+                let frame_delta =
-                    .get(&(client_id, stream_id))
+                    (frame_number - last_frame_number.get(&(client_id, stream_id)).unwrap()).0;
-                    .expect("failed to get encoder mutex");
 
-                let mut decoder = decoder_mutex.lock().expect("failed to get encoder");
+                if frame_delta == 2 {
-
+                    decode_sample(
-                let decoded_len = decoder
+                        client_id,
-                    .decode_float(
+                        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,
-                    )
+                    );
-                    .expect("failed to decode opus");
+                } else {
-
+                    decode_sample(
-                let mut active_streams_map = active_input_audio_streams_producers.lock().unwrap();
+                        client_id,
-
+                        stream_id,
-                if let Some(producer) = active_streams_map.get_mut(&(client_id, stream_id)) {
+                        stream_info,
-                    if stream_info.channels == output_device_config.channels() {
+                        output_device_config,
-                        producer.push_slice(
+                        &mut active_input_audio_streams_decoders,
-                            &decoded_samples[..decoded_len * stream_info.channels as usize],
+                        &mut active_input_audio_streams_resamplers,
-                        );
+                        &mut active_input_audio_streams_producers,
-                    } else if stream_info.channels > output_device_config.channels() {
+                        &mut active_input_audio_streams_consumers.clone(),
-                        for (i, frame) in decoded_samples
+                        audio_packet,
-                            [..decoded_len * stream_info.channels as usize]
+                        &mut decoded_samples,
-                            .chunks_exact(2)
+                        &mut resampled_samples,
-                            .enumerate()
+                        &mut decoded_samples_converted,
-                        {
+                        ropus::DecodeMode::Normal,
-                            decoded_samples_converted[i] = (frame[0] + frame[1]) / 2.0;
+                    );
-                        }
-
-                        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],
-                        );
-                    }
                 }
+
+                last_frame_number.insert((client_id, stream_id), frame_number);
             }
             Err(e) => eprintln!("failed to receive udp packet: {}", e),
         }
@@ -228,9 +367,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(
@@ -240,6 +379,7 @@ pub async fn audio_stream_spawner(
     mut global_event_rx: Receiver<ClientEvent>,
     udp_socket: Arc<UdpSocket>,
     client_info: Arc<RwLock<ClientInfo>>,
+    force_channels: u16,
 ) {
     let current_stream_id: u8 = rand::random();
 
@@ -250,10 +390,10 @@ 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: input_device_config.channels(),
+                channels: force_channels,
-                sample_rate: input_device_config.sample_rate(),
+                sample_rate: SAMPLE_RATE_48K,
             }));
             break;
         }
@@ -264,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(_) => {}
@@ -274,8 +414,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_frames as usize * input_device_config.channels() as usize;
-    let ring_buffer_size = ring_buffer_samples;
     let ring_buffer = HeapRb::<f32>::new(ring_buffer_size);
     let (producer, mut consumer) = ring_buffer.split();
 
@@ -319,43 +458,97 @@ 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 channels: ropus::Channels = {
+    let samples_needed = frames_needed * input_device_config.channels() as usize;
-        if input_device_config.channels() == 2 {
+
-            ropus::Channels::Stereo
+    let mut frame_buffer = vec![0f32; samples_needed];
-        } else {
+
-            ropus::Channels::Mono
+    let mut resampler_input_buffer = vec![0.0f32; frames_needed * force_channels as usize];
+
+    let mut resampler = Fft::<f32>::new(
+        input_device_config.sample_rate() as usize,
+        SAMPLE_RATE_48K as usize,
+        frames_needed,
+        force_channels as usize,
+        FixedSync::Both,
+    )
+    .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,
+        2 => ropus::Channels::Stereo,
+        _ => {
+            eprintln!("unsupported by encoder channel count");
+            return;
         }
     };
 
-    let mut opus_encoder = Encoder::builder(
+    let mut opus_encoder =
-        input_device_config.sample_rate(),
+        Encoder::builder(SAMPLE_RATE_48K, encoder_channels, ropus::Application::Voip)
-        channels,
+            .inband_fec(ropus::InbandFec::Enabled)
-        ropus::Application::Voip,
+            .packet_loss_perc(10)
-    )
+            .build()
-    .build()
+            .unwrap();
-    .unwrap();
 
-    let mut encoded_frame_buffer = [0u8; 1024];
+    let mut encoded_frame_buffer = vec![0u8; 2048];
+
+    let mut frame_num = Wrapping(0u8);
 
     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);
+            let frame_buffer_adapter = InterleavedSlice::new(
+                &frame_buffer,
+                input_device_config.channels() as usize,
+                frames_needed,
+            )
+            .unwrap();
+
+            frame_buffer_adapter.copy_from_channel_to_slice(0, 0, &mut resampler_input_buffer); // im only using first channel for now
+
+            let resampler_input = InterleavedSlice::new(
+                &resampler_input_buffer,
+                force_channels as usize,
+                frames_needed,
+            )
+            .unwrap();
+
+            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(
+                    &resampler_output_buffer[..valid_samples],
+                    &mut encoded_frame_buffer,
+                )
+                .expect("failed to encode frame");
+
+            let mut payload = Vec::<u8>::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(client_id.to_le_bytes());
-            payload.extend(current_stream_id.to_be_bytes());
+            payload.extend(current_stream_id.to_le_bytes());
 
-            let encoded_len = opus_encoder
+            payload.extend(frame_num.0.to_le_bytes());
-                .encode_float(&frame_buffer.as_slice(), &mut encoded_frame_buffer)
+            frame_num += Wrapping(1);
-                .expect("failed to encode frame");
 
             payload.extend_from_slice(&encoded_frame_buffer[..encoded_len]);
 

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::{
+    models::{AudioControl, CliArgs, ClientEvent, ClientInfo, InputAudioStreamsInfo},
-        AudioControl, CliArgs, ClientEvent, ClientInfo, InputAudioStreamsConsumers,
-        InputAudioStreamsEncoders, InputAudioStreamsInfo, InputAudioStreamsProducers,
-    },
     tcp_connection::tcp_connection,
     tui::Tui,
 };
@@ -28,23 +25,12 @@ async fn main() {
 
     let (global_event_tx, _) = broadcast::channel::<ClientEvent>(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<AudioControl> = 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(),
     });
@@ -86,21 +72,19 @@ async fn main() {
 
         tokio::spawn(audio_stream_receiver(
             output_device,
-            output_device_config.into(),
+            output_device_config,
-            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(),
         ));
 
         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(),
             client_info.clone(),
+            1,
         ));
 
         tokio::signal::ctrl_c().await.expect("failed to close");

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,
@@ -25,6 +26,7 @@ pub enum TuiScreen {
     Exit,
 }
 
+#[allow(dead_code)]
 #[derive(Clone)]
 pub enum ClientEvent {
     NewConnection,
@@ -38,21 +40,19 @@ pub struct ClientInfo {
     pub connected: bool,
 }
 
-type AudioStreamKey = (u32, u8);
+pub type AudioStreamKey = (u32, u8);
 
 pub type InputAudioStreamsProducers =
-    Arc<Mutex<HashMap<AudioStreamKey, Caching<Arc<HeapRb<f32>>, true, false>>>>;
+    HashMap<AudioStreamKey, Caching<Arc<HeapRb<f32>>, true, false>>;
 pub type InputAudioStreamsConsumers =
     Arc<Mutex<HashMap<AudioStreamKey, Caching<Arc<HeapRb<f32>>, false, true>>>>;
 
-pub type InputAudioStreamsEncoders = Arc<RwLock<HashMap<AudioStreamKey, Mutex<Decoder>>>>;
+pub type InputAudioStreamsDecoders = HashMap<AudioStreamKey, Decoder>;
+pub type InputAudioStreamsResamplers = HashMap<AudioStreamKey, Fft<f32>>;
 
 pub type InputAudioStreamsInfo = Arc<RwLock<HashMap<AudioStreamKey, AudioStreamInfo>>>;
 
 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<UdpSocket>,
 }

client/src/tcp_connection.rs

@@ -1,8 +1,6 @@
 use futures::{SinkExt, StreamExt};
-use ringbuf::{HeapRb, traits::Split};
+use shared::{TcpRequest, TcpResponse, TcpResponseData};
-use ropus::Decoder;
+use std::sync::{Arc, RwLock};
-use shared::{ResponseStatus, TcpRequest, TcpResponse, TcpResponseData};
-use std::sync::{Arc, Mutex, RwLock};
 use tokio::{
     net::TcpStream,
     sync::broadcast::{Receiver, Sender},
@@ -36,76 +34,37 @@ 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::<TcpResponse>(&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 } => {
+                                    TcpResponseData::UserConnected { .. } => {
-                                        println!("new user connected: {}", user_id);
+                                        // println!("new user connected: {}", user_id);
                                     }
-                                    TcpResponseData::UserDisconnected { user_id } => {
+                                    TcpResponseData::UserDisconnected { .. } => {
-                                        println!("user disconnected: {}", user_id);
+                                        // println!("user disconnected: {}", user_id);
                                     }
                                     TcpResponseData::AudioStreamsFullUpdate { audio_stream_info } => {
-                                        let mut producer_map = audio_control.active_input_audio_streams_producers.lock().unwrap();
-                                        producer_map.clear();
-                                        let mut consumer_map = audio_control.active_input_audio_streams_consumers.lock().unwrap();
-                                        consumer_map.clear();
-
-                                        let mut encoder_map = audio_control.active_input_audio_streams_encoders.write().unwrap();
-                                        encoder_map.clear();
-
                                         let mut info_map = audio_control.active_input_audio_streams_info.write().unwrap();
                                         info_map.clear();
 
-
                                         for audio_stream in audio_stream_info {
-                                            let ring_buffer_ms = 150.0;
-                                            let ring_buffer_frames = (ring_buffer_ms / 1000.0) * audio_stream.sample_rate as f32;
-                                            let ring_buffer_samples = ring_buffer_frames as usize * audio_stream.channels as usize;
-                                            let ring_buffer_size = ring_buffer_samples;
-                                            let ring_buffer = HeapRb::<f32>::new(ring_buffer_size);
-                                            let (producer, consumer) = ring_buffer.split();
-
-                                            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,
                                             );
                                         }
-                                        println!("audio streams update");
+                                        // println!("audio streams update");
                                     }
                                     TcpResponseData::NewAudioStreamAccepted { audio_stream_id } => {
                                         let _ = global_event_tx.send(ClientEvent::NewAudioStreamAccepted(audio_stream_id));
@@ -135,13 +94,13 @@ 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");
 
                             }
                             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 {
@@ -151,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");
                             }

client/src/tui.rs

@@ -1,5 +1,5 @@
 use cpal::{
-    Device, SAMPLE_RATE_48K, SampleFormat, SupportedStreamConfig,
+    Device, SupportedStreamConfig,
     traits::{DeviceTrait, HostTrait},
 };
 use crossterm::event::{self, Event, KeyCode, KeyEvent, KeyEventKind};
@@ -75,7 +75,7 @@ impl BlankScreen {
             self.handle_events();
         }
 
-        return TuiScreen::Exit;
+        TuiScreen::Exit
     }
 
     fn draw(&mut self, frame: &mut Frame) {
@@ -85,9 +85,8 @@ 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 {
+                if let KeyCode::Char('q') = key_event.code {
-                    KeyCode::Char('q') => self.exit = true,
+                    self.exit = true
-                    _ => {}
                 }
             }
             _ => {}
@@ -121,97 +120,32 @@ impl DeviceSelectScreen {
 
         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()
-                if default_config.sample_rate() == SAMPLE_RATE_48K
+        {
-                    && (default_config.channels() == 1)
+            input_devices.push((default_input_device.clone(), default_config));
-                {
-                    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;
-                        }
-                    }
-                }
-            }
         }
 
         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
+                    input_devices.push((possible_input_device.clone(), default_config));
-                        && (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;
-                            }
-                        }
-                    }
                 }
             }
         }
 
         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()
-                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));
-                {
-                    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;
-                        }
-                    }
-                }
-            }
         }
 
         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
+                    output_devices.push((possible_output_device.clone(), default_config));
-                        && (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;
-                            }
-                        }
-                    }
                 }
             }
         }
@@ -227,13 +161,13 @@ impl DeviceSelectScreen {
             exit_screen: TuiScreen::Exit,
             input_device_state: input_state,
             output_device_state: output_state,
-            input_devices: input_devices,
+            input_devices,
-            output_devices: output_devices,
+            output_devices,
             selected_selection: 0,
 
-            audio_control: audio_control,
+            audio_control,
-            global_event_tx: global_event_tx,
+            global_event_tx,
-            client_info: client_info,
+            client_info,
         }
     }
 
@@ -245,7 +179,7 @@ impl DeviceSelectScreen {
             self.handle_events();
         }
 
-        return self.exit_screen;
+        self.exit_screen
     }
 
     fn draw(&mut self, frame: &mut Frame) {
@@ -261,11 +195,18 @@ impl DeviceSelectScreen {
         let widths = [Constraint::Percentage(100), Constraint::Min(8)];
 
         let mut input_device_rows: Vec<Row> = Vec::new();
-        for (device, config) in self.input_devices.iter() {
+        for (i, (device, config)) in self.input_devices.iter().enumerate() {
-            input_device_rows.push(Row::new(vec![
+            if i == 0 {
-                device.description().unwrap().name().to_string(),
+                input_device_rows.push(Row::new(vec![
-                config.sample_format().to_string(),
+                    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 +237,21 @@ impl DeviceSelectScreen {
         frame.render_stateful_widget(input_device_table, left_block, &mut self.input_device_state);
 
         let mut output_device_rows: Vec<Row> = Vec::new();
-        for (device, config) in self.output_devices.iter() {
+        for (i, (device, config)) in self.output_devices.iter().enumerate() {
-            output_device_rows.push(Row::new(vec![
+            if i == 0 {
-                device.description().unwrap().name().to_string(),
+                output_device_rows.push(Row::new(vec![
-                config.sample_format().to_string(),
+                    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(
@@ -392,27 +343,19 @@ impl DeviceSelectScreen {
 
             tokio::spawn(audio_stream_receiver(
                 output_device.clone(),
-                output_config.clone(),
+                *output_config,
-                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(),
             ));
 
             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(),
                 self.client_info.clone(),
+                1,
             ));
 
             self.exit_screen = TuiScreen::Blank;

server/src/main.rs

@@ -23,10 +23,9 @@ struct Client {
     audio_streams: Option<Vec<AudioStreamInfo>>,
 }
 
+#[allow(dead_code)]
 #[derive(Clone)]
 enum ServerEvent {
-    NewAudioStream(AudioStreamInfo),
-    SendAllStreams,
     SendAllOtherStreams(u32),
     UserConnected(u32),
     UserDisconnected(u32),
@@ -36,13 +35,13 @@ type State = Arc<RwLock<HashMap<u32, Client>>>;
 
 fn scrape_all_client_streams_info(clients: &HashMap<u32, Client>) -> Vec<AudioStreamInfo> {
     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 {
             all_audio_streams_info.extend(audio_streams_info.iter().cloned());
         }
     }
 
-    return all_audio_streams_info;
+    all_audio_streams_info
 }
 
 #[tokio::main]
@@ -70,25 +69,25 @@ 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;
             }
 
             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) {
-                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 other_client_id != client_id
-                    if let Some(destication_udp) = other_client.udp_addr {
+                    && let Some(destination_udp) = other_client.udp_addr
-                        let _ = udp_socket.send_to(&buf[..len], destication_udp).await;
+                {
-                    }
+                    let _ = udp_socket.send_to(&buf[..len], destination_udp).await;
                 }
             }
         }
@@ -137,7 +136,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");
 
@@ -146,9 +145,9 @@ async fn main() {
                                         TcpRequestData::NewAudioStream { audio_stream_info } => {
                                             if let Some(client) = tcp_state.write().await.get_mut(&current_client_id) {
                                                 if let Some(audio_streams) = &mut client.audio_streams {
-                                                    audio_streams.push(audio_stream_info.clone());
+                                                    audio_streams.push(audio_stream_info);
                                                 } else {
-                                                    client.audio_streams = Some(vec![audio_stream_info.clone()]);
+                                                    client.audio_streams = Some(vec![audio_stream_info]);
                                                 }
                                             }
 
@@ -164,13 +163,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(&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);
                                                     }
-                                                }
-                                            }
                                             let _ = global_event_tx_clone.send(ServerEvent::SendAllOtherStreams(current_client_id));
                                         }
                                     },
@@ -206,7 +203,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");
 
@@ -229,21 +226,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),