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

.gitignore

@@ -1 +1,2 @@
 /target
+.DS_Store

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:
-
 ```
+-c, --connect-with-defaults <CONNECT_WITH_DEFAULTS>
 -s, --server-ip <SERVER_IP>
 -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

@@ -13,3 +13,10 @@ tokio = { version = "1.53.1", features = ["full"] }
 rand = "0.10.2"
 clap = { version = "4.6.6", features = ["derive"] }
 ropus = "0.12.18"
+ratatui = "0.30.2"
+crossterm = "0.29.0"
+gag = "1.0.0"
+tokio-util = { version = "0.7.19", features = ["codec"]}
+futures = "0.3.34"
+rubato = "5.0.0"
+audioadapter-buffers = "5.2.0"

client/src/audio_streams.rs

@@ -0,0 +1,560 @@
+use audioadapter_buffers::direct::InterleavedSlice;
+use cpal::{
+    Device, FromSample, SAMPLE_RATE_48K, Sample, SampleFormat, SizedSample, StreamConfig,
+    SupportedStreamConfig,
+    traits::{DeviceTrait, StreamTrait},
+};
+use ringbuf::{
+    HeapRb, SharedRb,
+    storage::Heap,
+    traits::{Consumer, Observer, Producer, Split},
+    wrap::caching::Caching,
+};
+use ropus::{Decoder, Encoder};
+use rubato::{Fft, FixedSync, Resampler, audioadapter::Adapter};
+use shared::AudioStreamInfo;
+use std::{
+    collections::HashMap,
+    num::Wrapping,
+    sync::{Arc, Mutex, RwLock},
+};
+use tokio::{
+    net::UdpSocket,
+    sync::broadcast::{Receiver, Sender},
+};
+
+use crate::models::{
+    AudioStreamKey, ClientEvent, ClientInfo, InputAudioStreamsConsumers, InputAudioStreamsDecoders,
+    InputAudioStreamsInfo, InputAudioStreamsProducers, InputAudioStreamsResamplers,
+};
+
+fn cpal_build_output_stream<T>(
+    output_device: &Device,
+    output_device_config: StreamConfig,
+    active_input_audio_streams_consumers: InputAudioStreamsConsumers,
+) -> cpal::Stream
+where
+    T: SizedSample,
+    T: FromSample<f32>,
+{
+    let output_data_fn = move |data: &mut [T], _: &cpal::OutputCallbackInfo| {
+        let mut streams = active_input_audio_streams_consumers.lock().unwrap();
+        for sample in data {
+            let mut mixed_value: f32 = 0.0;
+
+            for consumer in streams.values_mut() {
+                mixed_value += consumer.try_pop().unwrap_or(0.0)
+            }
+
+            *sample = mixed_value.to_sample::<T>();
+        }
+    };
+    let err_fn = |err| eprintln!("stream error: {}", err);
+
+    output_device
+        .build_output_stream(output_device_config, output_data_fn, err_fn, None)
+        .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_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.clone(),
+        ),
+        SampleFormat::I64 => cpal_build_output_stream::<i64>(
+            &output_device,
+            output_device_config.into(),
+            active_input_audio_streams_consumers.clone(),
+        ),
+        SampleFormat::U64 => cpal_build_output_stream::<u64>(
+            &output_device,
+            output_device_config.into(),
+            active_input_audio_streams_consumers.clone(),
+        ),
+        SampleFormat::F32 => cpal_build_output_stream::<f32>(
+            &output_device,
+            output_device_config.into(),
+            active_input_audio_streams_consumers.clone(),
+        ),
+        SampleFormat::I32 => cpal_build_output_stream::<i32>(
+            &output_device,
+            output_device_config.into(),
+            active_input_audio_streams_consumers.clone(),
+        ),
+        SampleFormat::U32 => cpal_build_output_stream::<u32>(
+            &output_device,
+            output_device_config.into(),
+            active_input_audio_streams_consumers.clone(),
+        ),
+        SampleFormat::I16 => cpal_build_output_stream::<i16>(
+            &output_device,
+            output_device_config.into(),
+            active_input_audio_streams_consumers.clone(),
+        ),
+        SampleFormat::U16 => cpal_build_output_stream::<u16>(
+            &output_device,
+            output_device_config.into(),
+            active_input_audio_streams_consumers.clone(),
+        ),
+        SampleFormat::I8 => cpal_build_output_stream::<i8>(
+            &output_device,
+            output_device_config.into(),
+            active_input_audio_streams_consumers.clone(),
+        ),
+        SampleFormat::U8 => cpal_build_output_stream::<u8>(
+            &output_device,
+            output_device_config.into(),
+            active_input_audio_streams_consumers.clone(),
+        ),
+        _ => {
+            eprintln!("unsupported output stream format :(");
+            return;
+        }
+    };
+
+    output_stream.play().expect("failed to play output stream");
+
+    let mut receive_buffer = vec![0u8; 4096];
+
+    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 {
+            Ok((len, _)) => {
+                if len < 6 {
+                    continue;
+                }
+
+                let client_id_bytes: [u8; 4] = receive_buffer[0..4].try_into().unwrap();
+                let client_id = u32::from_le_bytes(client_id_bytes);
+
+                let stream_id_bytes = receive_buffer[4];
+                let stream_id = u8::from_le(stream_id_bytes);
+
+                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
+                        .read()
+                        .unwrap()
+                        .get(&(client_id, stream_id))
+                    {
+                        info
+                    } else {
+                        continue;
+                    }
+                };
+
+                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 frame_delta =
+                    (frame_number - last_frame_number.get(&(client_id, stream_id)).unwrap()).0;
+
+                if frame_delta == 2 {
+                    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,
+                    );
+                } 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),
+        }
+    }
+}
+
+fn cpal_build_input_stream<T>(
+    input_device: &Device,
+    input_device_config: StreamConfig,
+    mut producer: Caching<Arc<SharedRb<Heap<f32>>>, true, false>,
+) -> cpal::Stream
+where
+    T: SizedSample,
+    f32: FromSample<T>,
+{
+    let input_data_fn = move |data: &[T], _: &cpal::InputCallbackInfo| {
+        for &sample in data {
+            let f32_sample: f32 = sample.to_sample::<f32>();
+
+            if producer.try_push(f32_sample).is_err() {}
+        }
+    };
+    let err_fn = |err| eprintln!("stream error: {}", err);
+
+    input_device
+        .build_input_stream(input_device_config, input_data_fn, err_fn, None)
+        .expect("building input stream failed")
+}
+
+pub async fn audio_stream_spawner(
+    input_device: Device,
+    input_device_config: SupportedStreamConfig,
+    global_event_tx: Sender<ClientEvent>,
+    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();
+
+    loop {
+        let (connected, client_id) = {
+            let info = client_info.read().unwrap();
+            (info.connected, info.client_id)
+        };
+        if connected {
+            let _ = global_event_tx.send(ClientEvent::NewAudioStreamRequest(AudioStreamInfo {
+                client_id,
+                stream_id: current_stream_id,
+                channels: force_channels,
+                sample_rate: SAMPLE_RATE_48K,
+            }));
+            break;
+        }
+    }
+
+    loop {
+        match global_event_rx.recv().await {
+            Ok(ClientEvent::NewAudioStreamAccepted(stream_id))
+                if stream_id == current_stream_id =>
+            {
+                // println!("new stream accepted: {}", stream_id);
+                break;
+            }
+            Ok(_) => {}
+            Err(e) => eprintln!("error: {}", e),
+        }
+    }
+
+    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_size = ring_buffer_frames as usize * input_device_config.channels() as usize;
+    let ring_buffer = HeapRb::<f32>::new(ring_buffer_size);
+    let (producer, mut consumer) = ring_buffer.split();
+
+    let input_stream = match input_device_config.sample_format() {
+        SampleFormat::F64 => {
+            cpal_build_input_stream::<f64>(&input_device, input_device_config.into(), producer)
+        }
+        SampleFormat::I64 => {
+            cpal_build_input_stream::<i64>(&input_device, input_device_config.into(), producer)
+        }
+        SampleFormat::U64 => {
+            cpal_build_input_stream::<u64>(&input_device, input_device_config.into(), producer)
+        }
+        SampleFormat::F32 => {
+            cpal_build_input_stream::<f32>(&input_device, input_device_config.into(), producer)
+        }
+        SampleFormat::I32 => {
+            cpal_build_input_stream::<i32>(&input_device, input_device_config.into(), producer)
+        }
+        SampleFormat::U32 => {
+            cpal_build_input_stream::<u32>(&input_device, input_device_config.into(), producer)
+        }
+        SampleFormat::I16 => {
+            cpal_build_input_stream::<i16>(&input_device, input_device_config.into(), producer)
+        }
+        SampleFormat::U16 => {
+            cpal_build_input_stream::<u16>(&input_device, input_device_config.into(), producer)
+        }
+        SampleFormat::I8 => {
+            cpal_build_input_stream::<i8>(&input_device, input_device_config.into(), producer)
+        }
+        SampleFormat::U8 => {
+            cpal_build_input_stream::<u8>(&input_device, input_device_config.into(), producer)
+        }
+        _ => {
+            eprintln!("unsupported input stream format :(");
+            return;
+        }
+    };
+
+    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 samples_needed = frames_needed * input_device_config.channels() as usize;
+
+    let mut frame_buffer = vec![0f32; samples_needed];
+
+    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(SAMPLE_RATE_48K, encoder_channels, ropus::Application::Voip)
+            .inband_fec(ropus::InbandFec::Enabled)
+            .packet_loss_perc(10)
+            .build()
+            .unwrap();
+
+    let mut encoded_frame_buffer = vec![0u8; 2048];
+
+    let mut frame_num = Wrapping(0u8);
+
+    loop {
+        if consumer.occupied_len() >= samples_needed {
+            consumer.pop_slice(&mut frame_buffer);
+
+            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_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]);
+
+            let _ = udp_socket.send_to(&payload, server_address).await;
+        } else {
+            tokio::time::sleep(std::time::Duration::from_millis(2)).await;
+        }
+    }
+}

client/src/main.rs

@@ -1,510 +1,119 @@
+mod audio_streams;
+mod models;
+mod tcp_connection;
+mod tui;
+
 use clap::Parser;
-use cpal::{
+use cpal::traits::{DeviceTrait, HostTrait};
-    Device,
-    traits::{DeviceTrait, HostTrait, StreamTrait},
-};
-use ringbuf::{
-    HeapRb,
-    traits::{Consumer, Observer, Producer, Split},
-    wrap::caching::Caching,
-};
-use ropus::{Decoder, Encoder};
-use shared::{AudioStreamInfo, ResponseStatus, TcpRequest, TcpResponse, TcpResponseData};
 use std::{
     collections::HashMap,
-    sync::{Arc, Mutex, RwLock},
+    sync::{Arc, RwLock},
 };
-use tokio::sync::broadcast::{self, Receiver, Sender};
+use tokio::net::UdpSocket;
-use tokio::{
+use tokio::sync::broadcast;
-    io::{AsyncReadExt, AsyncWriteExt},
+
-    net::{TcpStream, UdpSocket},
+use crate::{
+    audio_streams::{audio_stream_receiver, audio_stream_spawner},
+    models::{AudioControl, CliArgs, ClientEvent, ClientInfo, InputAudioStreamsInfo},
+    tcp_connection::tcp_connection,
+    tui::Tui,
 };
 
-#[derive(Parser)]
-#[command(version, about)]
-struct Cli {
-    #[arg(short, long, required(true))]
-    server_ip: Option<String>,
-}
-
-#[derive(Clone)]
-enum ClientEvent {
-    NewConnection,
-    NewAudioStreamRequest(AudioStreamInfo),
-    NewAudioStreamAccepted(u8),
-}
-
-type AudioStreamKey = (u32, u8);
-
-type InputAudioStreamsProducers =
-    Arc<Mutex<HashMap<AudioStreamKey, Caching<Arc<HeapRb<f32>>, true, false>>>>;
-type InputAudioStreamsConsumers =
-    Arc<Mutex<HashMap<AudioStreamKey, Caching<Arc<HeapRb<f32>>, false, true>>>>;
-
-type InputAudioStreamsEncoders = Arc<RwLock<HashMap<AudioStreamKey, Mutex<Decoder>>>>;
-
-type InputAudioStreamsInfo = Arc<RwLock<HashMap<AudioStreamKey, AudioStreamInfo>>>;
-
-struct ClientInfo {
-    client_id: u32,
-    server_address: String,
-    connected: bool,
-}
-
 #[tokio::main]
 async fn main() {
-    let cli = Cli::parse();
+    let cli = CliArgs::parse();
-
-    let client_info: Arc<RwLock<ClientInfo>> = Arc::new(RwLock::new(ClientInfo {
-        client_id: 0,
-        server_address: cli.server_ip.unwrap(),
-        connected: false,
-    }));
 
     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 active_input_audio_streams_producers_tcp = active_input_audio_streams_producers.clone();
+    let udp_socket = Arc::new(UdpSocket::bind("0.0.0.0:0").await.unwrap());
-    let active_input_audio_streams_consumers_tcp = active_input_audio_streams_consumers.clone();
-    let active_input_audio_streams_encoders_tcp = active_input_audio_streams_encoders.clone();
-    let active_input_audio_streams_info_tcp = active_input_audio_streams_info.clone();
 
-    let server_address = {
+    let audio_control: Arc<AudioControl> = Arc::new(AudioControl {
-        let info = client_info.read().unwrap();
+        active_input_audio_streams_info: active_input_audio_streams_info.clone(),
-        info.server_address.clone()
+        udp_socket: udp_socket.clone(),
-    };
-
-    let mut tcp_stream = TcpStream::connect(server_address)
-        .await
-        .expect("failed to connect using tcp");
-    // tcp_stream.readable().await.expect("idk");
-    //
-    let client_info_tcp = client_info.clone();
-
-    let mut global_event_rx = global_event_tx.subscribe();
-
-    let global_event_tx_clone = global_event_tx.clone();
-
-    // tcp connection
-    tokio::spawn(async move {
-        let mut tcp_buf = [0u8; 4096];
-        loop {
-            tokio::select! {
-                socket_result = tcp_stream.read(&mut tcp_buf) => {
-
-                    match socket_result {
-                        Ok(0) => {
-                            println!("server shut down");
-                            break;
-                        }
-                        Ok(n) => {
-                            let ssss = String::from_utf8(tcp_buf[0..n].to_vec()).unwrap();
-                            println!("json data: {}", ssss);
-
-                            match serde_json::from_slice::<TcpResponse>(&tcp_buf[0..n]) {
-                            Ok(response) => {
-                                println!("connection Ok: {}", response.status == ResponseStatus::Ok);
-                                match response.data {
-                                    TcpResponseData::NewConnection { client_id } => {
-                                        let mut client_info_var = client_info_tcp.write().unwrap();
-                                        client_info_var.client_id = client_id;
-                                        client_info_var.connected = true;
-                                        println!("fetched client id: {}", client_id);
-                                    }
-                                    TcpResponseData::UserConnected { user_id } => {
-                                        println!("new user connected: {}", user_id);
-                                    }
-                                    TcpResponseData::UserDisconnected { user_id } => {
-                                        println!("user disconnected: {}", user_id);
-                                    }
-                                    TcpResponseData::AudioStreamsFullUpdate { audio_stream_info } => {
-                                        let mut producer_map = active_input_audio_streams_producers_tcp.lock().unwrap();
-                                        producer_map.clear();
-                                        let mut consumer_map = active_input_audio_streams_consumers_tcp.lock().unwrap();
-                                        consumer_map.clear();
-
-                                        let mut encoder_map = active_input_audio_streams_encoders_tcp.write().unwrap();
-                                        encoder_map.clear();
-
-                                        let mut info_map = active_input_audio_streams_info_tcp.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");
-                                    }
-                                    TcpResponseData::NewAudioStreamAccepted { audio_stream_id } => {
-                                        let _ = global_event_tx_clone.send(ClientEvent::NewAudioStreamAccepted(audio_stream_id));
-                                    }
-                                    _ => {
-                                        println!("wrong response")
-                                    }
-                                }
-                            }
-                            Err(e) => eprintln!("failed to parse tcp packet: {}", e),}
-                        },
-                        Err(e) => eprintln!("failed to receive tcp packet: {}", e),
-                    }
-                }
-                event_result = global_event_rx.recv() => {
-                    match event_result {
-                        Ok(event) => {
-                            match event {
-                                ClientEvent::NewConnection => {
-                                    let _ = tcp_stream
-                                        .write_all(
-                                            serde_json::to_string(&TcpRequest {
-                                                client_id: 0,
-                                                data: shared::TcpRequestData::NewConnection,
-                                            })
-                                            .expect("failed to parse initial request")
-                                            .as_bytes(),
-                                        )
-                                        .await
-                                        .expect("failed to send initial message");
-                                }
-                                ClientEvent::NewAudioStreamRequest(audio_stream_info) => {
-                                    println!("new audio stream request");
-                                    let session_client_id = client_info_tcp.read().unwrap().client_id;
-                                    let _ = tcp_stream
-                                        .write_all(
-                                            serde_json::to_string(&TcpRequest {
-                                                client_id: session_client_id,
-                                                data: shared::TcpRequestData::NewAudioStream {
-                                                    audio_stream_info
-                                                },
-                                            })
-                                            .expect("failed to parse initial request")
-                                            .as_bytes(),
-                                        )
-                                        .await
-                                        .expect("failed to send initial message");
-                                }
-                                _ => {}
-                            }
-                        }
-                        Err(e) => eprintln!("error: {}", e),
-                    }
-                }
-            }
-        }
     });
 
+    if let Some(server_ip) = cli.connect_with_defaults {
+        let client_info: Arc<RwLock<ClientInfo>> = Arc::new(RwLock::new(ClientInfo {
+            client_id: 0,
+            server_address: server_ip,
+            connected: false,
+        }));
+
+        tokio::spawn(tcp_connection(
+            client_info.clone(),
+            global_event_tx.clone(),
+            global_event_tx.subscribe(),
+            audio_control.clone(),
+        ));
+
         let _ = global_event_tx.send(ClientEvent::NewConnection);
 
-    let socket = Arc::new(UdpSocket::bind("0.0.0.0:0").await.unwrap());
-
-    let output_socket = socket.clone();
-    let input_socket = socket.clone();
-
         let host = cpal::default_host();
 
         let input_device = host
             .default_input_device()
             .expect("no input device available");
+        let input_device_config = input_device
+            .default_input_config()
+            .expect("failed to get default input config");
+
         let output_device = host
             .default_output_device()
             .expect("no output device available");
+        let output_device_config = output_device
+            .default_output_config()
+            .expect("failed to get default output config");
 
         println!("input device: {}", input_device);
         println!("output device: {}", output_device);
 
-    let speaker_config: cpal::StreamConfig = output_device
+        tokio::spawn(audio_stream_receiver(
-        .default_output_config()
+            output_device,
-        .expect("stream config init failed")
+            output_device_config,
-        .into();
+            active_input_audio_streams_info.clone(),
-
+            udp_socket.clone(),
-    println!("speaker config: {:?}", speaker_config);
+        ));
-
-    let active_input_audio_streams_consumer_cpal = active_input_audio_streams_consumers.clone();
-
-    let output_data_fn = move |data: &mut [f32], _: &cpal::OutputCallbackInfo| {
-        let mut streams = active_input_audio_streams_consumer_cpal.lock().unwrap();
-        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,
-                }
-            }
-
-            *sample = mixed_value;
-        }
-    };
-
-    let err_fn = |err| eprintln!("stream error: {}", err);
-    let output_stream = output_device
-        .build_output_stream(speaker_config, output_data_fn, err_fn, None)
-        .expect("building output stream failed");
-
-    output_stream.play().expect("failed to play output stream");
-
-    let active_input_audio_streams_producers_udp = active_input_audio_streams_producers.clone();
-    let active_input_audio_streams_info_udp = active_input_audio_streams_info.clone();
-    let speaker_config_clone = speaker_config.clone();
-    let active_input_audio_streams_encoders_clone = active_input_audio_streams_encoders.clone();
-
-    tokio::spawn(async move {
-        let mut receive_buffer = vec![0u8; 4096];
-
-        let mut decoded_samples = [0.0f32; 1024];
-        let mut decoded_samples_converted = [0.0f32; 2048];
-
-        loop {
-            match input_socket.recv_from(&mut receive_buffer).await {
-                Ok((len, _)) => {
-                    if len < 4 {
-                        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 stream_id_bytes = receive_buffer[4];
-                    let stream_id = u8::from_be(stream_id_bytes);
-
-                    let audio_packet = &receive_buffer[5..len];
-
-                    let stream_info = {
-                        if let Some(&info) = active_input_audio_streams_info_udp
-                            .read()
-                            .unwrap()
-                            .get(&(client_id, stream_id))
-                        {
-                            info
-                        } else {
-                            continue;
-                        }
-                    };
-
-                    let decoders_map = active_input_audio_streams_encoders_clone.read().unwrap();
-
-                    let decoder_mutex = decoders_map
-                        .get(&(client_id, stream_id))
-                        .expect("failed to get encoder mutex");
-
-                    let mut decoder = decoder_mutex.lock().expect("failed to get encoder");
-
-                    let decoded_len = decoder
-                        .decode_float(
-                            audio_packet,
-                            &mut decoded_samples,
-                            ropus::DecodeMode::Normal,
-                        )
-                        .expect("failed to decode opus");
-
-                    let mut active_streams_map =
-                        active_input_audio_streams_producers_udp.lock().unwrap();
-
-                    if let Some(producer) = active_streams_map.get_mut(&(client_id, stream_id)) {
-                        if stream_info.channels == speaker_config_clone.channels {
-                            producer.push_slice(
-                                &decoded_samples[..decoded_len * stream_info.channels as usize],
-                            );
-                        } else if stream_info.channels > speaker_config_clone.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;
-                            }
-
-                            producer.push_slice(
-                                &decoded_samples_converted
-                                    [..decoded_len * stream_info.channels as usize],
-                            );
-                        } else {
-                            for (i, frame) in decoded_samples_converted
-                                [..decoded_len * speaker_config_clone.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 * speaker_config_clone.channels as usize],
-                            );
-                        }
-                    }
-                }
-                Err(e) => eprintln!("failed to receive udp packet: {}", e),
-            }
-        }
-    });
-
-    let broadcast_transmitter = global_event_tx.clone();
-    let broadcast_receiver = global_event_tx.subscribe();
-
-    let client_info_audio_spawner = client_info.clone();
 
         tokio::spawn(audio_stream_spawner(
             input_device,
-        broadcast_transmitter,
+            input_device_config,
-        broadcast_receiver,
+            global_event_tx.clone(),
-        output_socket,
+            global_event_tx.subscribe(),
-        client_info_audio_spawner,
+            udp_socket.clone(),
+            client_info.clone(),
+            1,
         ));
 
         tokio::signal::ctrl_c().await.expect("failed to close");
         println!("shutting down");
     }
 
-async fn audio_stream_spawner(
+    if let Some(server_ip) = cli.server_ip {
-    input_device: Device,
+        let client_info: Arc<RwLock<ClientInfo>> = Arc::new(RwLock::new(ClientInfo {
-    broadcast_transmitter: Sender<ClientEvent>,
+            client_id: 0,
-    mut broadcast_receiver: Receiver<ClientEvent>,
+            server_address: server_ip,
-    udp_socket: Arc<UdpSocket>,
+            connected: false,
-    clinet_info: Arc<RwLock<ClientInfo>>,
-) {
-    let config: cpal::StreamConfig = input_device
-        .default_input_config()
-        .expect("input stream config init failed")
-        .into();
-
-    let current_stream_id: u8 = rand::random();
-
-    loop {
-        let (connected, client_id) = {
-            let info = clinet_info.read().unwrap();
-            (info.connected, info.client_id)
-        };
-        if connected {
-            let _ =
-                broadcast_transmitter.send(ClientEvent::NewAudioStreamRequest(AudioStreamInfo {
-                    client_id: client_id,
-                    stream_id: current_stream_id,
-                    channels: config.channels,
-                    sample_rate: config.sample_rate,
         }));
-            break;
-        }
-        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
-    }
 
-    loop {
+        tokio::spawn(tcp_connection(
-        match broadcast_receiver.recv().await {
+            client_info.clone(),
-            Ok(ClientEvent::NewAudioStreamAccepted(stream_id))
+            global_event_tx.clone(),
-                if stream_id == current_stream_id =>
+            global_event_tx.subscribe(),
-            {
+            audio_control.clone(),
-                println!("new stream accepted: {}", stream_id);
+        ));
-                break;
+
-            }
+        let _ = global_event_tx.send(ClientEvent::NewConnection);
-            Ok(_) => {}
+
-            Err(e) => eprintln!("error: {}", e),
+        ratatui::run(|terminal| {
-        }
+            Tui::new().run(
-    }
+                terminal,
-
+                audio_control.clone(),
-    let ring_buffer_ms = 150.0;
+                global_event_tx.clone(),
-    let ring_buffer_frames = (ring_buffer_ms / 1000.0) * config.sample_rate as f32;
+                client_info.clone(),
-    let ring_buffer_samples = ring_buffer_frames as usize * config.channels as usize;
+            )
-    let ring_buffer_size = ring_buffer_samples;
+        });
-    let ring_buffer = HeapRb::<f32>::new(ring_buffer_size);
-    let (mut producer, mut consumer) = ring_buffer.split();
-
-    let input_data_fn = move |data: &[f32], _: &cpal::InputCallbackInfo| {
-        for &sample in data {
-            if producer.try_push(sample).is_err() {}
-        }
-    };
-    let err_fn = |err| eprintln!("stream error: {}", err);
-
-    let input_stream = input_device
-        .build_input_stream(config, input_data_fn, err_fn, None)
-        .expect("building input stream failed");
-
-    input_stream.play().expect("failed to play input stream");
-
-    let frames_needed = ((20.0 / 1000.0) * config.sample_rate as f32) as usize;
-    let mut frame_buffer = vec![0.0; frames_needed];
-
-    let channels: ropus::Channels = {
-        if config.channels == 1 {
-            ropus::Channels::Mono
-        } else {
-            ropus::Channels::Stereo
-        }
-    };
-
-    let mut opus_encoder = Encoder::builder(config.sample_rate, channels, ropus::Application::Voip)
-        .build()
-        .unwrap();
-
-    let mut encoded_frame_buffer = [0u8; 1024];
-
-    loop {
-        if consumer.occupied_len() >= frames_needed {
-            consumer.pop_slice(&mut frame_buffer);
-
-            let mut payload = Vec::with_capacity(4 + 1 + frames_needed * 4);
-
-            let (server_address, client_id) = {
-                let info = clinet_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());
-
-            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;
-        } else {
-            tokio::time::sleep(std::time::Duration::from_millis(2)).await;
-        }
     }
 }

client/src/models.rs

@@ -0,0 +1,58 @@
+use clap::Parser;
+use ringbuf::{HeapRb, wrap::caching::Caching};
+use ropus::Decoder;
+use rubato::Fft;
+use shared::AudioStreamInfo;
+use std::{
+    collections::HashMap,
+    sync::{Arc, Mutex, RwLock},
+};
+use tokio::net::UdpSocket;
+
+#[derive(Parser)]
+#[command(version, about)]
+pub struct CliArgs {
+    #[arg(short, long, required(false))]
+    pub connect_with_defaults: Option<String>,
+
+    #[arg(short, long, required(false))]
+    pub server_ip: Option<String>,
+}
+
+#[derive(Debug, Clone, Copy)]
+pub enum TuiScreen {
+    DeviceSelect,
+    Blank,
+    Exit,
+}
+
+#[allow(dead_code)]
+#[derive(Clone)]
+pub enum ClientEvent {
+    NewConnection,
+    NewAudioStreamRequest(AudioStreamInfo),
+    NewAudioStreamAccepted(u8),
+}
+
+pub struct ClientInfo {
+    pub client_id: u32,
+    pub server_address: String,
+    pub connected: bool,
+}
+
+pub type AudioStreamKey = (u32, u8);
+
+pub type InputAudioStreamsProducers =
+    HashMap<AudioStreamKey, Caching<Arc<HeapRb<f32>>, true, false>>;
+pub type InputAudioStreamsConsumers =
+    Arc<Mutex<HashMap<AudioStreamKey, Caching<Arc<HeapRb<f32>>, false, true>>>>;
+
+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_info: InputAudioStreamsInfo,
+    pub udp_socket: Arc<UdpSocket>,
+}

client/src/tcp_connection.rs

@@ -0,0 +1,125 @@
+use futures::{SinkExt, StreamExt};
+use shared::{TcpRequest, TcpResponse, TcpResponseData};
+use std::sync::{Arc, RwLock};
+use tokio::{
+    net::TcpStream,
+    sync::broadcast::{Receiver, Sender},
+};
+use tokio_util::{
+    bytes::Bytes,
+    codec::{Framed, LengthDelimitedCodec},
+};
+
+use crate::models::{AudioControl, ClientEvent, ClientInfo};
+
+pub async fn tcp_connection(
+    client_info: Arc<RwLock<ClientInfo>>,
+    global_event_tx: Sender<ClientEvent>,
+    mut global_event_rx: Receiver<ClientEvent>,
+    audio_control: Arc<AudioControl>,
+) {
+    let server_address = {
+        let info = client_info.read().unwrap();
+        info.server_address.clone()
+    };
+
+    let tcp_stream = TcpStream::connect(server_address)
+        .await
+        .expect("failed to connect using tcp");
+
+    let mut framed_tcp_stream = Framed::new(tcp_stream, LengthDelimitedCodec::new());
+
+    loop {
+        tokio::select! {
+            frame_result = framed_tcp_stream.next() => {
+                match frame_result {
+                    None => {
+                        // println!("server shut down");
+                        break;
+                    }
+                    Some(Ok(bytes)) => {
+                        match serde_json::from_slice::<TcpResponse>(&bytes) {
+                            Ok(response) => {
+                                // println!("connection Ok: {}", response.status == ResponseStatus::Ok);
+                                match response.data {
+                                    TcpResponseData::NewConnection { client_id } => {
+                                        let mut client_info_var = client_info.write().unwrap();
+                                        client_info_var.client_id = client_id;
+                                        client_info_var.connected = true;
+                                        // println!("fetched client id: {}", client_id);
+                                    }
+                                    TcpResponseData::UserConnected { .. } => {
+                                        // println!("new user connected: {}", user_id);
+                                    }
+                                    TcpResponseData::UserDisconnected { .. } => {
+                                        // println!("user disconnected: {}", user_id);
+                                    }
+                                    TcpResponseData::AudioStreamsFullUpdate { audio_stream_info } => {
+                                        let mut info_map = audio_control.active_input_audio_streams_info.write().unwrap();
+                                        info_map.clear();
+
+                                        for audio_stream in audio_stream_info {
+                                            info_map.insert(
+                                                (audio_stream.client_id, audio_stream.stream_id),
+                                                audio_stream,
+                                            );
+                                        }
+                                        // println!("audio streams update");
+                                    }
+                                    TcpResponseData::NewAudioStreamAccepted { audio_stream_id } => {
+                                        let _ = global_event_tx.send(ClientEvent::NewAudioStreamAccepted(audio_stream_id));
+                                    }
+                                }
+                            }
+                            Err(e) => {
+                                eprintln!("parsing error: {}", e);
+                                break;
+                            }
+                        }
+                    }
+                    Some(Err(e)) => {
+                        eprintln!("socket read error: {}", e);
+                            break;
+                    }
+                }
+            }
+            event_result = global_event_rx.recv() => {
+                match event_result {
+                    Ok(event) => {
+                        match event {
+                            ClientEvent::NewConnection => {
+                                let request_json = serde_json::to_string(&TcpRequest {
+                                    client_id: 0,
+                                    data: shared::TcpRequestData::NewConnection,
+                                })
+                                .expect("failed to parse initial request");
+
+                                framed_tcp_stream.send(Bytes::from(request_json))
+                                .await
+                                .expect("failed to send initial message");
+
+                            }
+                            ClientEvent::NewAudioStreamRequest(audio_stream_info) => {
+                                // println!("new audio stream request");
+                                let session_client_id = client_info.read().unwrap().client_id;
+
+                                let request_json = serde_json::to_string(&TcpRequest {
+                                    client_id: session_client_id,
+                                    data: shared::TcpRequestData::NewAudioStream {
+                                        audio_stream_info
+                                    }
+                                }).expect("failed to parse initial request");
+
+                                framed_tcp_stream.send(Bytes::from(request_json))
+                                .await
+                                .expect("failed to send initial message");
+                            }
+                            _ => {}
+                        }
+                    }
+                    Err(e) => eprintln!("error: {}", e),
+                }
+            }
+        }
+    }
+}

client/src/tui.rs

@@ -0,0 +1,365 @@
+use cpal::{
+    Device, SupportedStreamConfig,
+    traits::{DeviceTrait, HostTrait},
+};
+use crossterm::event::{self, Event, KeyCode, KeyEvent, KeyEventKind};
+use gag::Gag;
+use ratatui::{
+    DefaultTerminal, Frame,
+    layout::{Constraint, Direction, Layout},
+    style::Style,
+    widgets::{Block, Paragraph, Row, Table, TableState},
+};
+use std::{
+    sync::{Arc, RwLock},
+    vec,
+};
+use tokio::sync::broadcast::Sender;
+
+use crate::{
+    audio_streams::{audio_stream_receiver, audio_stream_spawner},
+    models::{AudioControl, ClientEvent, ClientInfo, TuiScreen},
+};
+
+pub struct Tui {
+    current_screen: TuiScreen,
+}
+
+impl Tui {
+    pub fn new() -> Self {
+        Self {
+            current_screen: TuiScreen::DeviceSelect,
+        }
+    }
+
+    pub fn run(
+        &mut self,
+        terminal: &mut DefaultTerminal,
+        audio_control: Arc<AudioControl>,
+        global_event_tx: Sender<ClientEvent>,
+        client_info: Arc<RwLock<ClientInfo>>,
+    ) {
+        loop {
+            match self.current_screen {
+                TuiScreen::Exit => {
+                    break;
+                }
+                TuiScreen::Blank => self.current_screen = BlankScreen::new().run(terminal),
+                TuiScreen::DeviceSelect => {
+                    self.current_screen = DeviceSelectScreen::new(
+                        audio_control.clone(),
+                        global_event_tx.clone(),
+                        client_info.clone(),
+                    )
+                    .run(terminal);
+                }
+            }
+        }
+    }
+}
+
+struct BlankScreen {
+    exit: bool,
+}
+
+impl BlankScreen {
+    pub fn new() -> Self {
+        Self { exit: false }
+    }
+
+    pub fn run(&mut self, terminal: &mut DefaultTerminal) -> TuiScreen {
+        while !self.exit {
+            terminal
+                .draw(|frame| self.draw(frame))
+                .expect("failed to draw");
+            self.handle_events();
+        }
+
+        TuiScreen::Exit
+    }
+
+    fn draw(&mut self, frame: &mut Frame) {
+        frame.render_widget(Paragraph::new("q - quit"), frame.area());
+    }
+
+    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
+                }
+            }
+            _ => {}
+        }
+    }
+}
+
+struct DeviceSelectScreen {
+    exit: bool,
+    exit_screen: TuiScreen,
+    input_device_state: TableState,
+    output_device_state: TableState,
+    input_devices: Vec<(Device, SupportedStreamConfig)>,
+    output_devices: Vec<(Device, SupportedStreamConfig)>,
+    selected_selection: u8,
+
+    audio_control: Arc<AudioControl>,
+    global_event_tx: Sender<ClientEvent>,
+    client_info: Arc<RwLock<ClientInfo>>,
+}
+
+impl DeviceSelectScreen {
+    pub fn new(
+        audio_control: Arc<AudioControl>,
+        global_event_tx: Sender<ClientEvent>,
+        client_info: Arc<RwLock<ClientInfo>>,
+    ) -> Self {
+        let _print_gag = Gag::stderr().ok();
+
+        let host = cpal::default_host();
+
+        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));
+        }
+
+        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() {
+                    input_devices.push((possible_input_device.clone(), default_config));
+                }
+            }
+        }
+
+        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));
+        }
+
+        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));
+                }
+            }
+        }
+
+        let mut input_state = TableState::default();
+        input_state.select(Some(0));
+
+        let mut output_state = TableState::default();
+        output_state.select(Some(0));
+
+        Self {
+            exit: false,
+            exit_screen: TuiScreen::Exit,
+            input_device_state: input_state,
+            output_device_state: output_state,
+            input_devices,
+            output_devices,
+            selected_selection: 0,
+
+            audio_control,
+            global_event_tx,
+            client_info,
+        }
+    }
+
+    pub fn run(&mut self, terminal: &mut DefaultTerminal) -> TuiScreen {
+        while !self.exit {
+            terminal
+                .draw(|frame| self.draw(frame))
+                .expect("failed to draw");
+            self.handle_events();
+        }
+
+        self.exit_screen
+    }
+
+    fn draw(&mut self, frame: &mut Frame) {
+        let outer_layout = Layout::default()
+            .direction(Direction::Horizontal)
+            .constraints(vec![Constraint::Percentage(50), Constraint::Percentage(50)])
+            .split(frame.area());
+
+        let block = Block::new();
+        let left_block = block.inner(outer_layout[0]);
+        let right_block = block.inner(outer_layout[1]);
+
+        let widths = [Constraint::Percentage(100), Constraint::Min(8)];
+
+        let mut input_device_rows: Vec<Row> = Vec::new();
+        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(
+                Row::new(vec!["device name", "channels"])
+                    .style(Style::new().bold())
+                    .bottom_margin(1),
+            )
+            .block(
+                Block::bordered()
+                    .border_style({
+                        if self.selected_selection == 0 {
+                            Style::new().bold()
+                        } else {
+                            Style::default()
+                        }
+                    })
+                    .border_type({
+                        if self.selected_selection == 0 {
+                            ratatui::widgets::BorderType::Thick
+                        } else {
+                            ratatui::widgets::BorderType::Plain
+                        }
+                    })
+                    .title("select input device"),
+            )
+            .row_highlight_style(Style::new().reversed());
+
+        frame.render_stateful_widget(input_device_table, left_block, &mut self.input_device_state);
+
+        let mut output_device_rows: Vec<Row> = Vec::new();
+        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(
+                Row::new(vec!["device name", "channels"])
+                    .style(Style::new().bold())
+                    .bottom_margin(1),
+            )
+            .block(
+                Block::bordered()
+                    .title("select output device")
+                    .border_style({
+                        if self.selected_selection == 1 {
+                            Style::new().bold()
+                        } else {
+                            Style::default()
+                        }
+                    })
+                    .border_type({
+                        if self.selected_selection == 1 {
+                            ratatui::widgets::BorderType::Thick
+                        } else {
+                            ratatui::widgets::BorderType::Plain
+                        }
+                    }),
+            )
+            .row_highlight_style(Style::new().reversed());
+
+        frame.render_stateful_widget(
+            output_device_table,
+            right_block,
+            &mut self.output_device_state,
+        );
+    }
+
+    fn handle_events(&mut self) {
+        match event::read().expect("failed to read event") {
+            Event::Key(key_event) if key_event.kind == KeyEventKind::Press => {
+                self.handle_key_events(key_event);
+            }
+            _ => {}
+        }
+    }
+
+    fn handle_key_events(&mut self, key_event: KeyEvent) {
+        match key_event.code {
+            KeyCode::Char('q') => self.exit(),
+            KeyCode::Down => self.increment_table(),
+            KeyCode::Up => self.decrement_table(),
+            KeyCode::Left => self.selected_selection = 0,
+            KeyCode::Right => self.selected_selection = 1,
+            KeyCode::Enter => self.connect_devices(),
+            _ => {}
+        }
+    }
+
+    fn exit(&mut self) {
+        self.exit = true;
+    }
+
+    fn increment_table(&mut self) {
+        if self.selected_selection == 0 {
+            self.input_device_state.scroll_down_by(1);
+        } else {
+            self.output_device_state.scroll_down_by(1);
+        }
+    }
+
+    fn decrement_table(&mut self) {
+        if self.selected_selection == 0 {
+            self.input_device_state.scroll_up_by(1);
+        } else {
+            self.output_device_state.scroll_up_by(1);
+        }
+    }
+
+    fn connect_devices(&mut self) {
+        if let (Some(input_device_row), Some(output_device_row)) = (
+            self.input_device_state.selected(),
+            self.output_device_state.selected(),
+        ) {
+            let (input_device, input_config) = self
+                .input_devices
+                .get(input_device_row)
+                .expect("failed to get input config");
+            let (output_device, output_config) = self
+                .output_devices
+                .get(output_device_row)
+                .expect("failed to get output config");
+
+            tokio::spawn(audio_stream_receiver(
+                output_device.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,
+                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;
+            self.exit = true;
+        }
+    }
+}

server/Cargo.toml

@@ -10,3 +10,5 @@ serde_json = "1.0.151"
 tokio = { version = "1.53.1", features = ["full"] }
 rand = "0.10.2"
 clap = { version = "4.6.6", features = ["derive"] }
+tokio-util = { version = "0.7.19", features = ["codec"]}
+futures = "0.3.34"

server/src/main.rs

@@ -1,17 +1,19 @@
 use clap::Parser;
+use futures::{SinkExt, StreamExt};
 use shared::{AudioStreamInfo, ResponseStatus, TcpRequest, TcpRequestData, TcpResponse};
 use std::sync::Arc;
 use std::{collections::HashMap, net::SocketAddr};
-use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::{
     net::{TcpListener, UdpSocket},
     sync::{RwLock, broadcast},
 };
+use tokio_util::bytes::Bytes;
+use tokio_util::codec::{Framed, LengthDelimitedCodec};
 
 #[derive(Parser)]
 #[command(version, about)]
 struct Cli {
-    #[arg(short, long, default_value("0.0.0.0:8080"))]
+    #[arg(short, long, default_value("0.0.0.0:6767"))]
     server_ip: Option<String>,
 }
 
@@ -21,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),
@@ -34,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]
@@ -68,33 +69,34 @@ 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;
                 }
             }
         }
     });
 
     loop {
-        let (mut tcp_stream, tcp_address) =
+        let (tcp_stream, tcp_address) = tcp_listener.accept().await.expect("failed to accept tcp");
-            tcp_listener.accept().await.expect("failed to accept tcp");
+
+        let mut framed_tcp_stream = Framed::new(tcp_stream, LengthDelimitedCodec::new());
 
         let tcp_state = state.clone();
 
@@ -103,18 +105,18 @@ async fn main() {
 
         tokio::spawn(async move {
             let current_client_id: u32 = rand::random();
-            let mut tcp_buf = [0; 4096];
 
             loop {
                 tokio::select! {
-                    socket_result = tcp_stream.read(&mut tcp_buf) => {
+                    frame_result = framed_tcp_stream.next() => {
-                        match socket_result {
+                        match frame_result {
-                            Ok(0) => {
+                            None => {
                                 let _ = global_event_tx_clone.send(ServerEvent::UserDisconnected(current_client_id));
                                 println!("client disconnected: {}", tcp_address);
                                 break;
                             }
-                            Ok(n) => match serde_json::from_slice::<TcpRequest>(&tcp_buf[0..n]) {
+                            Some(Ok(bytes)) => {
+                                match serde_json::from_slice::<TcpRequest>(&bytes) {
                                     Ok(request) => match request.data {
                                         TcpRequestData::NewConnection => {
                                             tcp_state.write().await.insert(
@@ -126,18 +128,17 @@ async fn main() {
                                             );
 
                                             println!("new client: {}, from: {}", current_client_id, tcp_address);
-                                        let _ = tcp_stream
+
-                                            .write_all(
+                                            let response_json = serde_json::to_string(&TcpResponse {
-                                                serde_json::to_string(&TcpResponse {
                                                 status: ResponseStatus::Ok,
                                                 data: shared::TcpResponseData::NewConnection {
                                                     client_id: current_client_id,
                                                 },
-                                                })
+                                            }).expect("failed to serialize TcpRequestData::NewAudioStreamAccepted");
-                                                .expect("failed to parse response json TcpRequestData::NewConnection")
+
-                                                .as_bytes(),
+                                            framed_tcp_stream.send(Bytes::from(response_json))
-                                            )
+                                            .await
-                                            .await;
+                                            .expect("failed to send message");
 
                                             let _ = global_event_tx_clone.send(ServerEvent::UserConnected(current_client_id));
                                         }
@@ -149,42 +150,34 @@ async fn main() {
                                                     client.audio_streams = Some(vec![audio_stream_info]);
                                                 }
                                             }
-                                        let _ = tcp_stream
+
-                                            .write_all(
+                                            let response_json = serde_json::to_string(&TcpResponse {
-                                                serde_json::to_string(&TcpResponse {
                                                 status: ResponseStatus::Ok,
                                                 data: shared::TcpResponseData::NewAudioStreamAccepted {
                                                     audio_stream_id: audio_stream_info.stream_id,
                                                 },
-                                                })
+                                            }).expect("failed to serialize TcpRequestData::NewAudioStreamAccepted");
-                                                .expect("failed to parse response json TcpRequestData::NewConnection")
+
-                                                .as_bytes(),
+                                            let _ = framed_tcp_stream.send(Bytes::from(response_json)).await;
-                                            )
-                                            .await;
 
                                             let _ = global_event_tx_clone.send(ServerEvent::SendAllOtherStreams(current_client_id));
                                         }
                                         TcpRequestData::DropAudioStream { audio_stream_id } => {
-                                        if let Some(client) = tcp_state.write().await.get_mut(&current_client_id) {
+                                            if let Some(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
-                                                let mut maybe_idx: Option<usize> = None;
+                                                    && let Some(idx) = audio_streams.iter().position(|s| s.stream_id == audio_stream_id) {
-                                                for (i, audio_stream) in audio_streams.iter_mut().enumerate() {
-                                                    if audio_stream.stream_id == audio_stream_id {
-                                                        maybe_idx = Some(i);
-                                                    }
-                                                }
-                                                if let Some(idx) = maybe_idx {
                                                         audio_streams.remove(idx);
                                                     }
-                                            }
-                                        }
                                             let _ = global_event_tx_clone.send(ServerEvent::SendAllOtherStreams(current_client_id));
                                         }
-                                    _ => eprintln!("i dont support other requests yet"),
                                     },
                                     Err(e) => eprintln!("failed to parse tcp packet: {}", e),
-                            },
+                                }
-                            Err(e) => eprintln!("failed to receive tcp packet: {}", e),
+                            }
+                            Some(Err(e)) => {
+                                eprintln!("socket read error: {}", e);
+                                break;
+                            }
                         }
                     }
                     event_result = global_event_rx.recv() => {
@@ -193,36 +186,28 @@ async fn main() {
                                 match event {
                                     ServerEvent::UserConnected(new_user_id) => {
                                         if new_user_id != current_client_id {
-                                            let _ = tcp_stream
+                                            let response_json = serde_json::to_string(&TcpResponse {
-                                                .write_all(
-                                                    serde_json::to_string(&TcpResponse {
                                                 status: ResponseStatus::Ok,
                                                 data: shared::TcpResponseData::UserConnected {
                                                     user_id: new_user_id
                                                 },
-                                                    })
+                                            }).expect("failed to serialize ServerEvent::UserConnected");
-                                                    .expect("failed to parse response json ServerEvent::UserConnected")
+
-                                                    .as_bytes(),
+                                            let _ = framed_tcp_stream.send(Bytes::from(response_json)).await;
-                                                )
-                                                .await;
 
                                             let _ = global_event_tx_clone.send(ServerEvent::SendAllOtherStreams(current_client_id));
                                         }
                                     }
                                     ServerEvent::UserDisconnected(user_id) => {
                                         if user_id != current_client_id {
-                                            let _ = tcp_stream
+                                            let response_json = serde_json::to_string(&TcpResponse {
-                                                .write_all(
-                                                    serde_json::to_string(&TcpResponse {
                                                 status: ResponseStatus::Ok,
                                                 data: shared::TcpResponseData::UserDisconnected {
-                                                            user_id: user_id
+                                                    user_id
                                                 },
-                                                    })
+                                            }).expect("failed to serialize ServerEvent::UserDisconnected");
-                                                    .expect("failed to parse response json ServerEvent::UserConnected")
+
-                                                    .as_bytes(),
+                                            let _ = framed_tcp_stream.send(Bytes::from(response_json)).await;
-                                                )
-                                                .await;
 
                                             let _ = global_event_tx_clone.send(ServerEvent::SendAllOtherStreams(current_client_id));
                                         }
@@ -231,38 +216,15 @@ async fn main() {
                                         if current_client_id != id {
                                             let clients = tcp_state.read().await;
 
-                                            let _ = tcp_stream
+                                            let response_json = serde_json::to_string(&TcpResponse {
-                                                .write_all(
-                                                    serde_json::to_string(&TcpResponse {
                                                 status: ResponseStatus::Ok,
                                                 data: shared::TcpResponseData::AudioStreamsFullUpdate {
                                                     audio_stream_info: scrape_all_client_streams_info(&clients),
                                                 },
-                                                    })
+                                            }).expect("failed to serialize ServerEvent::SendAllOtherStreams");
-                                                    .expect("failed to parse response json TcpRequestData::NewConnection")
-                                                    .as_bytes(),
-                                                )
-                                                .await;
-                                        }
-                                    }
-                                    ServerEvent::SendAllStreams => {
-                                        let clients = tcp_state.read().await;
 
-                                        let _ = tcp_stream
+                                            let _ = framed_tcp_stream.send(Bytes::from(response_json)).await;
-                                            .write_all(
-                                                serde_json::to_string(&TcpResponse {
-                                                    status: ResponseStatus::Ok,
-                                                    data: shared::TcpResponseData::AudioStreamsFullUpdate {
-                                                        audio_stream_info: scrape_all_client_streams_info(&clients),
-                                                    },
-                                                })
-                                                .expect("failed to parse response json TcpRequestData::NewConnection")
-                                                .as_bytes(),
-                                            )
-                                            .await;
                                         }
-                                    _ => {
-                                        println!("lkdahjsf");
                                     }
                                 }
                             }