diff --git a/.gitea/actions/setup-env/action.yaml b/.gitea/actions/setup-env/action.yaml new file mode 100644 index 0000000..76d084f --- /dev/null +++ b/.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 diff --git a/.gitea/workflows/build.yaml b/.gitea/workflows/build.yaml new file mode 100644 index 0000000..0512587 --- /dev/null +++ b/.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 diff --git a/Cargo.lock b/Cargo.lock index 94a57d8..cc679a0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -113,6 +113,39 @@ dependencies = [ "bytemuck", ] +[[package]] +name = "audio-codec-algorithms" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1254ebf6529f3763c491acfb5ab6e960809e3c75a38584cc664f8c5667fb7107" + +[[package]] +name = "audioadapter" +version = "5.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d1292ef9edf681b7426ed089004b021a42896492f58bd0c909ad44e5faec9ac4" + +[[package]] +name = "audioadapter-buffers" +version = "5.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "46289a81a3bfa26d0f8b2415f8ca9decde9a66308a150568af5c102381b3c8ce" +dependencies = [ + "audioadapter", + "audioadapter-sample", + "num-traits", +] + +[[package]] +name = "audioadapter-sample" +version = "5.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5258faecf4edbe35ec483bda9b1f7438e6b391891819269c6b5461aeaa0e8d7d" +dependencies = [ + "audio-codec-algorithms", + "num-traits", +] + [[package]] name = "autocfg" version = "1.5.1" @@ -280,6 +313,7 @@ checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9" name = "client" version = "0.1.0" dependencies = [ + "audioadapter-buffers", "clap", "cpal", "crossterm", @@ -289,6 +323,7 @@ dependencies = [ "ratatui", "ringbuf", "ropus", + "rubato", "serde", "serde_json", "shared", @@ -1128,6 +1163,15 @@ dependencies = [ "minimal-lexical", ] +[[package]] +name = "num-complex" +version = "0.4.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "73f88a1307638156682bada9d7604135552957b7818057dcef22705b4d509495" +dependencies = [ + "num-traits", +] + [[package]] name = "num-conv" version = "0.2.2" @@ -1145,6 +1189,15 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "num-integer" +version = "0.1.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7ce2d95d4b3734dc35aa2f45e1aa22cd416814592a4f9d9205e11affd5b8e10b" +dependencies = [ + "num-traits", +] + [[package]] name = "num-traits" version = "0.2.19" @@ -1479,6 +1532,15 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "439ee305def115ba05938db6eb1644ff94165c5ab5e9420d1c1bcedbba909391" +[[package]] +name = "primal-check" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc0d895b311e3af9902528fbb8f928688abbd95872819320517cc24ca6b2bd08" +dependencies = [ + "num-integer", +] + [[package]] name = "proc-macro-crate" version = "3.5.0" @@ -1651,6 +1713,15 @@ dependencies = [ "unicode-width", ] +[[package]] +name = "realfft" +version = "3.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f821338fddb99d089116342c46e9f1fbf3828dba077674613e734e01d6ea8677" +dependencies = [ + "rustfft", +] + [[package]] name = "redox_syscall" version = "0.5.18" @@ -1710,6 +1781,22 @@ dependencies = [ "wide", ] +[[package]] +name = "rubato" +version = "5.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7cb1ffaf8738df50aab642a7f6465df81c6ba9e2818268053487165298114be" +dependencies = [ + "audioadapter", + "audioadapter-buffers", + "num-complex", + "num-integer", + "num-traits", + "realfft", + "visibility", + "windowfunctions", +] + [[package]] name = "rustc_version" version = "0.4.1" @@ -1719,6 +1806,20 @@ dependencies = [ "semver", ] +[[package]] +name = "rustfft" +version = "6.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "21db5f9893e91f41798c88680037dba611ca6674703c1a18601b01a72c8adb89" +dependencies = [ + "num-complex", + "num-integer", + "num-traits", + "primal-check", + "strength_reduce", + "transpose", +] + [[package]] name = "rustix" version = "1.1.4" @@ -1936,6 +2037,12 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f" +[[package]] +name = "strength_reduce" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fe895eb47f22e2ddd4dabc02bce419d2e643c8e3b585c78158b349195bc24d82" + [[package]] name = "strsim" version = "0.11.1" @@ -2218,6 +2325,16 @@ dependencies = [ "winnow", ] +[[package]] +name = "transpose" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ad61aed86bc3faea4300c7aee358b4c6d0c8d6ccc36524c96e4c92ccf26e77e" +dependencies = [ + "num-integer", + "strength_reduce", +] + [[package]] name = "typenum" version = "1.20.1" @@ -2283,6 +2400,17 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" +[[package]] +name = "visibility" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d674d135b4a8c1d7e813e2f8d1c9a58308aee4a680323066025e53132218bd91" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "vtparse" version = "0.6.2" @@ -2485,6 +2613,15 @@ version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" +[[package]] +name = "windowfunctions" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "90628d739333b7c5d2ee0b70210b97b8cddc38440c682c96fd9e2c24c2db5f3a" +dependencies = [ + "num-traits", +] + [[package]] name = "windows" version = "0.62.2" diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..859c9b2 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,8 @@ +FROM rust:alpine as builder +WORKDIR /usr/src/smalltalk +COPY . . +RUN cargo build -p server --release + +FROM alpine:latest +COPY --from=builder /usr/src/smalltalk/target/release/server /usr/local/bin/smalltalk +CMD ["smalltalk"] diff --git a/README.md b/README.md index b79dac3..1b526dc 100644 --- a/README.md +++ b/README.md @@ -1,24 +1,17 @@ # smalltalk -## build and run client +## running server -Usage: +### using docker ```bash -cargo run -p client --release -- --server-ip +docker build -t smalltalk . +docker run -d --rm -p 6767:6767/tcp -p 6767:6767/udp --name smalltalk smalltalk ``` -Options: +### using cargo -``` --s, --server-ip --h, --help Print help --V, --version Print version -``` - -## build and run server - -before, open port `8080` tcp/udp +before, open port `6767` tcp and udp Usage: @@ -29,17 +22,23 @@ cargo run -p server --release -- [OPTIONS] Options: ``` --s, --server-ip [default: 0.0.0.0:8080] +-s, --server-ip [default: 0.0.0.0:6767] -h, --help Print help -V, --version Print version ``` ---- +## build and run client -I think it will be based on: +Usage: -- `tokio`: async and network connection -- `cpal`: raw audio input and output -- `opus`: encoding and decoding audio stream -- `ffmpeg-next`: capturing, encoding and decoding video stream -- `egui` + `eframe`: ui +```bash +cargo run -p client --release -- --server-ip +``` + +Options: +``` +-c, --connect-with-defaults +-s, --server-ip +-h, --help Print help +-V, --version Print version +``` diff --git a/client/Cargo.toml b/client/Cargo.toml index f053389..6766d4e 100644 --- a/client/Cargo.toml +++ b/client/Cargo.toml @@ -18,3 +18,5 @@ crossterm = "0.29.0" gag = "1.0.0" tokio-util = { version = "0.7.19", features = ["codec"]} futures = "0.3.34" +rubato = "5.0.0" +audioadapter-buffers = "5.2.0" diff --git a/client/src/audio_streams.rs b/client/src/audio_streams.rs index c917509..67cc28f 100644 --- a/client/src/audio_streams.rs +++ b/client/src/audio_streams.rs @@ -1,5 +1,7 @@ +use audioadapter_buffers::direct::InterleavedSlice; use cpal::{ - Device, FromSample, Sample, SampleFormat, SizedSample, StreamConfig, SupportedStreamConfig, + Device, FromSample, SAMPLE_RATE_48K, Sample, SampleFormat, SizedSample, StreamConfig, + SupportedStreamConfig, traits::{DeviceTrait, StreamTrait}, }; use ringbuf::{ @@ -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, - InputAudioStreamsInfo, InputAudioStreamsProducers, + AudioStreamKey, ClientEvent, ClientInfo, InputAudioStreamsConsumers, InputAudioStreamsDecoders, + InputAudioStreamsInfo, InputAudioStreamsProducers, InputAudioStreamsResamplers, }; fn cpal_build_output_stream( @@ -35,11 +42,8 @@ where for sample in data { let mut mixed_value: f32 = 0.0; - for (_key, consumer) in streams.iter_mut() { - mixed_value += match consumer.try_pop() { - Some(s) => s, - None => 0.0, - } + for consumer in streams.values_mut() { + mixed_value += consumer.try_pop().unwrap_or(0.0) } *sample = mixed_value.to_sample::(); @@ -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, + resampled_samples: &mut Vec, + decoded_samples_converted: &mut Vec, + decode_mode: ropus::DecodeMode, +) { + let input_frames = ((20.0 / 1000.0) * stream_info.sample_rate as f32) as usize; + let resampler_output_frames = + ((20.0 / 1000.0) * output_device_config.sample_rate() as f32) as usize; + + if !active_input_audio_streams_decoders.contains_key(&(client_id, stream_id)) { + let channels: ropus::Channels = { + if stream_info.channels == 2 { + ropus::Channels::Stereo + } else { + ropus::Channels::Mono + } + }; + + let decoder = + Decoder::new(stream_info.sample_rate, channels).expect("failed to create decoder"); + active_input_audio_streams_decoders + .insert((stream_info.client_id, stream_info.stream_id), decoder); + } + + active_input_audio_streams_resamplers + .entry((client_id, stream_id)) + .or_insert_with(|| { + Fft::::new( + stream_info.sample_rate as usize, + output_device_config.sample_rate() as usize, + input_frames, + stream_info.channels as usize, + FixedSync::Both, + ) + .unwrap() + }); + + if let std::collections::hash_map::Entry::Vacant(e) = + active_input_audio_streams_producers.entry((client_id, stream_id)) + { + let ring_buffer_frames = (100.0 / 1000.0) * stream_info.sample_rate as f32; + let ring_buffer_samples = ring_buffer_frames as usize * stream_info.channels as usize; + let ring_buffer = HeapRb::::new(ring_buffer_samples); + let (producer, consumer) = ring_buffer.split(); + + e.insert(producer); + active_input_audio_streams_consumers + .lock() + .unwrap() + .insert((client_id, stream_id), consumer); + } + + let decoder = active_input_audio_streams_decoders + .get_mut(&(client_id, stream_id)) + .expect("failed to get decoder"); + + let _ = decoder + .decode_float(audio_packet, decoded_samples, decode_mode) + .expect("failed to decode opus"); + + let decoded_samples_adapter = InterleavedSlice::new( + &decoded_samples, + stream_info.channels as usize, + input_frames, + ) + .unwrap(); + + let mut resampled_samples_adapter = InterleavedSlice::new_mut( + resampled_samples, + stream_info.channels as usize, + resampler_output_frames, + ) + .unwrap(); + + let (_, frames_out) = active_input_audio_streams_resamplers + .get_mut(&(client_id, stream_id)) + .expect("failed to get resampler") + .process_into_buffer( + &decoded_samples_adapter, + &mut resampled_samples_adapter, + None, + ) + .expect("error during resampling"); + + if let Some(producer) = active_input_audio_streams_producers.get_mut(&(client_id, stream_id)) { + for (&resampled_sample, output_frame) in resampled_samples + .iter() + .step_by(stream_info.channels as usize) + .zip( + decoded_samples_converted + .chunks_exact_mut(output_device_config.channels() as usize), + ) + { + for output_sample in output_frame.iter_mut() { + *output_sample = resampled_sample; + } + } + + producer.push_slice( + &decoded_samples_converted[..frames_out * output_device_config.channels() as usize], + ); + } } pub async fn audio_stream_receiver( output_device: Device, output_device_config: SupportedStreamConfig, - active_input_audio_streams_producers: InputAudioStreamsProducers, - active_input_audio_streams_consumers: InputAudioStreamsConsumers, - active_input_audio_streams_encoders: InputAudioStreamsEncoders, active_input_audio_streams_info: InputAudioStreamsInfo, udp_socket: Arc, ) { + let mut active_input_audio_streams_producers: InputAudioStreamsProducers = HashMap::new(); + let active_input_audio_streams_consumers: InputAudioStreamsConsumers = + Arc::new(Mutex::new(HashMap::new())); + let mut active_input_audio_streams_decoders: InputAudioStreamsDecoders = HashMap::new(); + let mut active_input_audio_streams_resamplers: InputAudioStreamsResamplers = HashMap::new(); + + let mut last_frame_number: HashMap> = HashMap::new(); + let output_stream = match output_device_config.sample_format() { SampleFormat::F64 => cpal_build_output_stream::( &output_device, output_device_config.into(), - active_input_audio_streams_consumers, + active_input_audio_streams_consumers.clone(), ), SampleFormat::I64 => cpal_build_output_stream::( &output_device, output_device_config.into(), - active_input_audio_streams_consumers, + active_input_audio_streams_consumers.clone(), ), SampleFormat::U64 => cpal_build_output_stream::( &output_device, output_device_config.into(), - active_input_audio_streams_consumers, + active_input_audio_streams_consumers.clone(), ), SampleFormat::F32 => cpal_build_output_stream::( &output_device, output_device_config.into(), - active_input_audio_streams_consumers, + active_input_audio_streams_consumers.clone(), ), SampleFormat::I32 => cpal_build_output_stream::( &output_device, output_device_config.into(), - active_input_audio_streams_consumers, + active_input_audio_streams_consumers.clone(), ), SampleFormat::U32 => cpal_build_output_stream::( &output_device, output_device_config.into(), - active_input_audio_streams_consumers, + active_input_audio_streams_consumers.clone(), ), SampleFormat::I16 => cpal_build_output_stream::( &output_device, output_device_config.into(), - active_input_audio_streams_consumers, + active_input_audio_streams_consumers.clone(), ), SampleFormat::U16 => cpal_build_output_stream::( &output_device, output_device_config.into(), - active_input_audio_streams_consumers, + active_input_audio_streams_consumers.clone(), ), SampleFormat::I8 => cpal_build_output_stream::( &output_device, output_device_config.into(), - active_input_audio_streams_consumers, + active_input_audio_streams_consumers.clone(), ), SampleFormat::U8 => cpal_build_output_stream::( &output_device, output_device_config.into(), - active_input_audio_streams_consumers, + active_input_audio_streams_consumers.clone(), ), _ => { eprintln!("unsupported output stream format :("); @@ -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 mut decoded_samples_converted = [0.0f32; 2048]; + let max_input_raw_samples = ((20.0 / 1000.0) * 2.0 * SAMPLE_RATE_48K as f32) as usize; // 20ms, 2channels, 48khz + let mut decoded_samples = vec![0.0f32; max_input_raw_samples]; + + let mut resampled_samples = vec![0.0f32; max_input_raw_samples]; + + let output_raw_samples = ((20.0 / 1000.0) + * output_device_config.channels() as f32 + * output_device_config.sample_rate() as f32) as usize; + let mut decoded_samples_converted = vec![0.0f32; output_raw_samples]; loop { match udp_socket.recv_from(&mut receive_buffer).await { 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 - .get(&(client_id, stream_id)) - .expect("failed to get encoder mutex"); + let frame_delta = + (frame_number - last_frame_number.get(&(client_id, stream_id)).unwrap()).0; - let mut decoder = decoder_mutex.lock().expect("failed to get encoder"); - - let decoded_len = decoder - .decode_float( + 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, - ) - .expect("failed to decode opus"); - - let mut active_streams_map = active_input_audio_streams_producers.lock().unwrap(); - - if let Some(producer) = active_streams_map.get_mut(&(client_id, stream_id)) { - if stream_info.channels == output_device_config.channels() { - producer.push_slice( - &decoded_samples[..decoded_len * stream_info.channels as usize], - ); - } else if stream_info.channels > output_device_config.channels() { - for (i, frame) in decoded_samples - [..decoded_len * stream_info.channels as usize] - .chunks_exact(2) - .enumerate() - { - decoded_samples_converted[i] = (frame[0] + frame[1]) / 2.0; - } - - 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], - ); - } + ); + } 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), } @@ -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, udp_socket: Arc, client_info: Arc>, + 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(), - sample_rate: input_device_config.sample_rate(), + channels: force_channels, + 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_samples; + let ring_buffer_size = ring_buffer_frames as usize * input_device_config.channels() as usize; let ring_buffer = HeapRb::::new(ring_buffer_size); let (producer, mut consumer) = ring_buffer.split(); @@ -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 = { - if input_device_config.channels() == 2 { - ropus::Channels::Stereo - } else { - ropus::Channels::Mono + 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::::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( - input_device_config.sample_rate(), - channels, - ropus::Application::Voip, - ) - .build() - .unwrap(); + 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 = [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::::with_capacity(1 + 4 + 1 + encoded_len * 4); let (server_address, client_id) = { let info = client_info.read().unwrap(); (info.server_address.clone(), info.client_id) }; - payload.extend(client_id.to_be_bytes()); - payload.extend(current_stream_id.to_be_bytes()); + payload.extend(client_id.to_le_bytes()); + payload.extend(current_stream_id.to_le_bytes()); - let encoded_len = opus_encoder - .encode_float(&frame_buffer.as_slice(), &mut encoded_frame_buffer) - .expect("failed to encode frame"); + payload.extend(frame_num.0.to_le_bytes()); + frame_num += Wrapping(1); payload.extend_from_slice(&encoded_frame_buffer[..encoded_len]); diff --git a/client/src/main.rs b/client/src/main.rs index a2c591a..3f5396e 100644 --- a/client/src/main.rs +++ b/client/src/main.rs @@ -7,17 +7,14 @@ use clap::Parser; use cpal::traits::{DeviceTrait, HostTrait}; use std::{ collections::HashMap, - sync::{Arc, Mutex, RwLock}, + sync::{Arc, RwLock}, }; use tokio::net::UdpSocket; use tokio::sync::broadcast; use crate::{ audio_streams::{audio_stream_receiver, audio_stream_spawner}, - models::{ - AudioControl, CliArgs, ClientEvent, ClientInfo, InputAudioStreamsConsumers, - InputAudioStreamsEncoders, InputAudioStreamsInfo, InputAudioStreamsProducers, - }, + models::{AudioControl, CliArgs, ClientEvent, ClientInfo, InputAudioStreamsInfo}, tcp_connection::tcp_connection, tui::Tui, }; @@ -28,23 +25,12 @@ async fn main() { let (global_event_tx, _) = broadcast::channel::(16); - let active_input_audio_streams_producers: InputAudioStreamsProducers = - Arc::new(Mutex::new(HashMap::new())); - let active_input_audio_streams_consumers: InputAudioStreamsConsumers = - Arc::new(Mutex::new(HashMap::new())); - - let active_input_audio_streams_encoders: InputAudioStreamsEncoders = - Arc::new(RwLock::new(HashMap::new())); - let active_input_audio_streams_info: InputAudioStreamsInfo = Arc::new(RwLock::new(HashMap::new())); let udp_socket = Arc::new(UdpSocket::bind("0.0.0.0:0").await.unwrap()); let audio_control: Arc = Arc::new(AudioControl { - active_input_audio_streams_producers: active_input_audio_streams_producers.clone(), - active_input_audio_streams_consumers: active_input_audio_streams_consumers.clone(), - active_input_audio_streams_encoders: active_input_audio_streams_encoders.clone(), active_input_audio_streams_info: active_input_audio_streams_info.clone(), udp_socket: udp_socket.clone(), }); @@ -86,21 +72,19 @@ async fn main() { tokio::spawn(audio_stream_receiver( output_device, - output_device_config.into(), - active_input_audio_streams_producers.clone(), - active_input_audio_streams_consumers.clone(), - active_input_audio_streams_encoders.clone(), + output_device_config, active_input_audio_streams_info.clone(), udp_socket.clone(), )); tokio::spawn(audio_stream_spawner( input_device, - input_device_config.into(), + input_device_config, global_event_tx.clone(), global_event_tx.subscribe(), udp_socket.clone(), client_info.clone(), + 1, )); tokio::signal::ctrl_c().await.expect("failed to close"); diff --git a/client/src/models.rs b/client/src/models.rs index f3f4187..adfaedd 100644 --- a/client/src/models.rs +++ b/client/src/models.rs @@ -1,6 +1,7 @@ use clap::Parser; use ringbuf::{HeapRb, wrap::caching::Caching}; use ropus::Decoder; +use rubato::Fft; use shared::AudioStreamInfo; use std::{ collections::HashMap, @@ -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>, true, false>>>>; + HashMap>, true, false>>; pub type InputAudioStreamsConsumers = Arc>, false, true>>>>; -pub type InputAudioStreamsEncoders = Arc>>>; +pub type InputAudioStreamsDecoders = HashMap; +pub type InputAudioStreamsResamplers = HashMap>; pub type InputAudioStreamsInfo = Arc>>; pub struct AudioControl { - pub active_input_audio_streams_producers: InputAudioStreamsProducers, - pub active_input_audio_streams_consumers: InputAudioStreamsConsumers, - pub active_input_audio_streams_encoders: InputAudioStreamsEncoders, pub active_input_audio_streams_info: InputAudioStreamsInfo, pub udp_socket: Arc, } diff --git a/client/src/tcp_connection.rs b/client/src/tcp_connection.rs index 159dd72..92daf4c 100644 --- a/client/src/tcp_connection.rs +++ b/client/src/tcp_connection.rs @@ -1,8 +1,6 @@ use futures::{SinkExt, StreamExt}; -use ringbuf::{HeapRb, traits::Split}; -use ropus::Decoder; -use shared::{ResponseStatus, TcpRequest, TcpResponse, TcpResponseData}; -use std::sync::{Arc, Mutex, RwLock}; +use shared::{TcpRequest, TcpResponse, TcpResponseData}; +use std::sync::{Arc, 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::(&bytes) { Ok(response) => { - println!("connection Ok: {}", response.status == ResponseStatus::Ok); + // println!("connection Ok: {}", response.status == ResponseStatus::Ok); match response.data { TcpResponseData::NewConnection { client_id } => { let mut client_info_var = client_info.write().unwrap(); client_info_var.client_id = client_id; client_info_var.connected = true; - println!("fetched client id: {}", client_id); + // println!("fetched client id: {}", client_id); } - TcpResponseData::UserConnected { user_id } => { - println!("new user connected: {}", user_id); + TcpResponseData::UserConnected { .. } => { + // println!("new user connected: {}", user_id); } - TcpResponseData::UserDisconnected { user_id } => { - println!("user disconnected: {}", user_id); + TcpResponseData::UserDisconnected { .. } => { + // println!("user disconnected: {}", user_id); } TcpResponseData::AudioStreamsFullUpdate { audio_stream_info } => { - let mut producer_map = audio_control.active_input_audio_streams_producers.lock().unwrap(); - producer_map.clear(); - let mut consumer_map = audio_control.active_input_audio_streams_consumers.lock().unwrap(); - consumer_map.clear(); - - let mut encoder_map = audio_control.active_input_audio_streams_encoders.write().unwrap(); - encoder_map.clear(); - let mut info_map = audio_control.active_input_audio_streams_info.write().unwrap(); info_map.clear(); - for audio_stream in audio_stream_info { - let ring_buffer_ms = 150.0; - let ring_buffer_frames = (ring_buffer_ms / 1000.0) * audio_stream.sample_rate as f32; - let ring_buffer_samples = ring_buffer_frames as usize * audio_stream.channels as usize; - let ring_buffer_size = ring_buffer_samples; - let ring_buffer = HeapRb::::new(ring_buffer_size); - let (producer, consumer) = ring_buffer.split(); - - producer_map.insert( - (audio_stream.client_id, audio_stream.stream_id), - producer, - ); - consumer_map.insert( - (audio_stream.client_id, audio_stream.stream_id), - consumer, - ); - - let channels: ropus::Channels = { - if audio_stream.channels == 1 { - ropus::Channels::Mono - } else { - ropus::Channels::Stereo - } - }; - - let decoder = Decoder::new(audio_stream.sample_rate, channels).expect("failed to creat decoder"); - encoder_map.insert( - (audio_stream.client_id, audio_stream.stream_id), - Mutex::new(decoder), - ); - info_map.insert( (audio_stream.client_id, audio_stream.stream_id), audio_stream, ); } - 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"); } diff --git a/client/src/tui.rs b/client/src/tui.rs index fad8a91..03a8210 100644 --- a/client/src/tui.rs +++ b/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 { - KeyCode::Char('q') => self.exit = true, - _ => {} + if let KeyCode::Char('q') = key_event.code { + 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 Ok(default_config) = default_input_device.default_input_config() { - if default_config.sample_rate() == SAMPLE_RATE_48K - && (default_config.channels() == 1) - { - input_devices.push((default_input_device.clone(), default_config)); - } - } else if let Ok(supported_configs) = default_input_device.supported_input_configs() { - for supported_config in supported_configs { - if let Some(input_config) = - supported_config.try_with_sample_rate(SAMPLE_RATE_48K) - { - if input_config.channels() == 1 { - input_devices.push((default_input_device, input_config)); - break; - } - } - } - } + 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() { - if default_config.sample_rate() == SAMPLE_RATE_48K - && (default_config.channels() == 1) - { - input_devices.push((possible_input_device.clone(), default_config)); - continue; - } - } - if let Ok(supported_configs) = possible_input_device.supported_input_configs() { - for supported_config in supported_configs { - if let Some(input_config) = - supported_config.try_with_sample_rate(SAMPLE_RATE_48K) - { - if input_config.channels() == 1 { - input_devices.push((possible_input_device, input_config)); - break; - } - } - } + input_devices.push((possible_input_device.clone(), default_config)); } } } let mut output_devices: Vec<(Device, SupportedStreamConfig)> = Vec::new(); - if let Some(default_output_device) = host.default_output_device() { - if let Ok(default_config) = default_output_device.default_output_config() { - if default_config.sample_rate() == SAMPLE_RATE_48K - && (default_config.channels() == 1 || default_config.channels() == 2) - { - output_devices.push((default_output_device.clone(), default_config.into())); - } - } else if let Ok(supported_configs) = default_output_device.supported_output_configs() { - for supported_config in supported_configs { - if let Some(output_config) = - supported_config.try_with_sample_rate(SAMPLE_RATE_48K) - { - if output_config.channels() == 1 || output_config.channels() == 2 { - output_devices.push((default_output_device, output_config.into())); - break; - } - } - } - } + 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() { - if default_config.sample_rate() == SAMPLE_RATE_48K - && (default_config.channels() == 1 || default_config.channels() == 2) - { - output_devices - .push((possible_output_device.clone(), default_config.into())); - continue; - } - } - if let Ok(supported_configs) = possible_output_device.supported_output_configs() { - for supported_config in supported_configs { - if let Some(output_config) = - supported_config.try_with_sample_rate(SAMPLE_RATE_48K) - { - if output_config.channels() == 1 || output_config.channels() == 2 { - output_devices.push((possible_output_device, output_config.into())); - break; - } - } - } + output_devices.push((possible_output_device.clone(), default_config)); } } } @@ -227,13 +161,13 @@ impl DeviceSelectScreen { exit_screen: TuiScreen::Exit, input_device_state: input_state, output_device_state: output_state, - input_devices: input_devices, - output_devices: output_devices, + input_devices, + output_devices, selected_selection: 0, - audio_control: audio_control, - global_event_tx: global_event_tx, - client_info: client_info, + audio_control, + global_event_tx, + client_info, } } @@ -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 = Vec::new(); - for (device, config) in self.input_devices.iter() { - input_device_rows.push(Row::new(vec![ - device.description().unwrap().name().to_string(), - config.sample_format().to_string(), - ])); + for (i, (device, config)) in self.input_devices.iter().enumerate() { + if i == 0 { + input_device_rows.push(Row::new(vec![ + format!("(default) {}", device.description().unwrap().name()), + config.channels().to_string(), + ])); + } else { + input_device_rows.push(Row::new(vec![ + device.description().unwrap().name().to_string(), + config.channels().to_string(), + ])); + } } let input_device_table = Table::new(input_device_rows, widths) .header( @@ -296,11 +237,21 @@ impl DeviceSelectScreen { frame.render_stateful_widget(input_device_table, left_block, &mut self.input_device_state); let mut output_device_rows: Vec = Vec::new(); - for (device, config) in self.output_devices.iter() { - output_device_rows.push(Row::new(vec![ - device.description().unwrap().name().to_string(), - config.sample_format().to_string(), - ])); + for (i, (device, config)) in self.output_devices.iter().enumerate() { + if i == 0 { + output_device_rows.push(Row::new(vec![ + format!( + "(default) {}", + device.description().unwrap().name().to_string() + ), + config.channels().to_string(), + ])); + } else { + output_device_rows.push(Row::new(vec![ + device.description().unwrap().name().to_string(), + config.channels().to_string(), + ])); + } } let output_device_table = Table::new(output_device_rows, widths) .header( @@ -392,27 +343,19 @@ impl DeviceSelectScreen { tokio::spawn(audio_stream_receiver( output_device.clone(), - output_config.clone(), - self.audio_control - .active_input_audio_streams_producers - .clone(), - self.audio_control - .active_input_audio_streams_consumers - .clone(), - self.audio_control - .active_input_audio_streams_encoders - .clone(), + *output_config, self.audio_control.active_input_audio_streams_info.clone(), self.audio_control.udp_socket.clone(), )); tokio::spawn(audio_stream_spawner( input_device.clone(), - input_config.clone(), + *input_config, self.global_event_tx.clone(), self.global_event_tx.subscribe(), self.audio_control.udp_socket.clone(), self.client_info.clone(), + 1, )); self.exit_screen = TuiScreen::Blank; diff --git a/server/src/main.rs b/server/src/main.rs index 07ccd54..69b2ac0 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -23,10 +23,9 @@ struct Client { audio_streams: Option>, } +#[allow(dead_code)] #[derive(Clone)] enum ServerEvent { - NewAudioStream(AudioStreamInfo), - SendAllStreams, SendAllOtherStreams(u32), UserConnected(u32), UserDisconnected(u32), @@ -36,13 +35,13 @@ type State = Arc>>; fn scrape_all_client_streams_info(clients: &HashMap) -> Vec { let mut all_audio_streams_info: Vec = Vec::new(); - for (_, data) in clients { + for data in clients.values() { if let Some(audio_streams_info) = &data.audio_streams { all_audio_streams_info.extend(audio_streams_info.iter().cloned()); } } - return all_audio_streams_info; + all_audio_streams_info } #[tokio::main] @@ -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 let Some(destication_udp) = other_client.udp_addr { - let _ = udp_socket.send_to(&buf[..len], destication_udp).await; - } + if other_client_id != client_id + && let Some(destination_udp) = other_client.udp_addr + { + 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(¤t_client_id) { if let Some(audio_streams) = &mut client.audio_streams { - audio_streams.push(audio_stream_info.clone()); + audio_streams.push(audio_stream_info); } else { - client.audio_streams = Some(vec![audio_stream_info.clone()]); + client.audio_streams = Some(vec![audio_stream_info]); } } @@ -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(¤t_client_id) { - if let Some(audio_streams) = &mut client.audio_streams { - if let Some(idx) = audio_streams.iter().position(|s| s.stream_id == audio_stream_id) { + if let Some(client) = tcp_state.write().await.get_mut(¤t_client_id) + && let Some(audio_streams) = &mut client.audio_streams + && let Some(idx) = audio_streams.iter().position(|s| s.stream_id == audio_stream_id) { audio_streams.remove(idx); } - } - } let _ = global_event_tx_clone.send(ServerEvent::SendAllOtherStreams(current_client_id)); } }, @@ -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),