Merge pull request 'Wip' (#1) from wip into main
Reviewed-on: #1
This commit was merged in pull request #1.
This commit is contained in:
@@ -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
|
||||
@@ -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
|
||||
Generated
+137
@@ -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"
|
||||
|
||||
@@ -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"]
|
||||
@@ -1,24 +1,17 @@
|
||||
# smalltalk
|
||||
|
||||
## build and run client
|
||||
## running server
|
||||
|
||||
Usage:
|
||||
### using docker
|
||||
|
||||
```bash
|
||||
cargo run -p client --release -- --server-ip <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 <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 <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
|
||||
```
|
||||
|
||||
---
|
||||
## 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 <SERVER_IP>
|
||||
```
|
||||
|
||||
Options:
|
||||
```
|
||||
-c, --connect-with-defaults <CONNECT_WITH_DEFAULTS>
|
||||
-s, --server-ip <SERVER_IP>
|
||||
-h, --help Print help
|
||||
-V, --version Print version
|
||||
```
|
||||
|
||||
@@ -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"
|
||||
|
||||
+294
-101
@@ -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<T>(
|
||||
@@ -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::<T>();
|
||||
@@ -47,70 +51,189 @@ where
|
||||
};
|
||||
let err_fn = |err| eprintln!("stream error: {}", err);
|
||||
|
||||
return output_device
|
||||
output_device
|
||||
.build_output_stream(output_device_config, output_data_fn, err_fn, None)
|
||||
.expect("building output stream failed");
|
||||
.expect("building output stream failed")
|
||||
}
|
||||
|
||||
fn decode_sample(
|
||||
client_id: u32,
|
||||
stream_id: u8,
|
||||
stream_info: AudioStreamInfo,
|
||||
output_device_config: SupportedStreamConfig,
|
||||
active_input_audio_streams_decoders: &mut InputAudioStreamsDecoders,
|
||||
active_input_audio_streams_resamplers: &mut InputAudioStreamsResamplers,
|
||||
active_input_audio_streams_producers: &mut InputAudioStreamsProducers,
|
||||
active_input_audio_streams_consumers: &mut InputAudioStreamsConsumers,
|
||||
audio_packet: &[u8],
|
||||
decoded_samples: &mut Vec<f32>,
|
||||
resampled_samples: &mut Vec<f32>,
|
||||
decoded_samples_converted: &mut Vec<f32>,
|
||||
decode_mode: ropus::DecodeMode,
|
||||
) {
|
||||
let input_frames = ((20.0 / 1000.0) * stream_info.sample_rate as f32) as usize;
|
||||
let resampler_output_frames =
|
||||
((20.0 / 1000.0) * output_device_config.sample_rate() as f32) as usize;
|
||||
|
||||
if !active_input_audio_streams_decoders.contains_key(&(client_id, stream_id)) {
|
||||
let channels: ropus::Channels = {
|
||||
if stream_info.channels == 2 {
|
||||
ropus::Channels::Stereo
|
||||
} else {
|
||||
ropus::Channels::Mono
|
||||
}
|
||||
};
|
||||
|
||||
let decoder =
|
||||
Decoder::new(stream_info.sample_rate, channels).expect("failed to create decoder");
|
||||
active_input_audio_streams_decoders
|
||||
.insert((stream_info.client_id, stream_info.stream_id), decoder);
|
||||
}
|
||||
|
||||
active_input_audio_streams_resamplers
|
||||
.entry((client_id, stream_id))
|
||||
.or_insert_with(|| {
|
||||
Fft::<f32>::new(
|
||||
stream_info.sample_rate as usize,
|
||||
output_device_config.sample_rate() as usize,
|
||||
input_frames,
|
||||
stream_info.channels as usize,
|
||||
FixedSync::Both,
|
||||
)
|
||||
.unwrap()
|
||||
});
|
||||
|
||||
if let std::collections::hash_map::Entry::Vacant(e) =
|
||||
active_input_audio_streams_producers.entry((client_id, stream_id))
|
||||
{
|
||||
let ring_buffer_frames = (100.0 / 1000.0) * stream_info.sample_rate as f32;
|
||||
let ring_buffer_samples = ring_buffer_frames as usize * stream_info.channels as usize;
|
||||
let ring_buffer = HeapRb::<f32>::new(ring_buffer_samples);
|
||||
let (producer, consumer) = ring_buffer.split();
|
||||
|
||||
e.insert(producer);
|
||||
active_input_audio_streams_consumers
|
||||
.lock()
|
||||
.unwrap()
|
||||
.insert((client_id, stream_id), consumer);
|
||||
}
|
||||
|
||||
let decoder = active_input_audio_streams_decoders
|
||||
.get_mut(&(client_id, stream_id))
|
||||
.expect("failed to get decoder");
|
||||
|
||||
let _ = decoder
|
||||
.decode_float(audio_packet, decoded_samples, decode_mode)
|
||||
.expect("failed to decode opus");
|
||||
|
||||
let decoded_samples_adapter = InterleavedSlice::new(
|
||||
&decoded_samples,
|
||||
stream_info.channels as usize,
|
||||
input_frames,
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
let mut resampled_samples_adapter = InterleavedSlice::new_mut(
|
||||
resampled_samples,
|
||||
stream_info.channels as usize,
|
||||
resampler_output_frames,
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
let (_, frames_out) = active_input_audio_streams_resamplers
|
||||
.get_mut(&(client_id, stream_id))
|
||||
.expect("failed to get resampler")
|
||||
.process_into_buffer(
|
||||
&decoded_samples_adapter,
|
||||
&mut resampled_samples_adapter,
|
||||
None,
|
||||
)
|
||||
.expect("error during resampling");
|
||||
|
||||
if let Some(producer) = active_input_audio_streams_producers.get_mut(&(client_id, stream_id)) {
|
||||
for (&resampled_sample, output_frame) in resampled_samples
|
||||
.iter()
|
||||
.step_by(stream_info.channels as usize)
|
||||
.zip(
|
||||
decoded_samples_converted
|
||||
.chunks_exact_mut(output_device_config.channels() as usize),
|
||||
)
|
||||
{
|
||||
for output_sample in output_frame.iter_mut() {
|
||||
*output_sample = resampled_sample;
|
||||
}
|
||||
}
|
||||
|
||||
producer.push_slice(
|
||||
&decoded_samples_converted[..frames_out * output_device_config.channels() as usize],
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn audio_stream_receiver(
|
||||
output_device: Device,
|
||||
output_device_config: SupportedStreamConfig,
|
||||
active_input_audio_streams_producers: InputAudioStreamsProducers,
|
||||
active_input_audio_streams_consumers: InputAudioStreamsConsumers,
|
||||
active_input_audio_streams_encoders: InputAudioStreamsEncoders,
|
||||
active_input_audio_streams_info: InputAudioStreamsInfo,
|
||||
udp_socket: Arc<UdpSocket>,
|
||||
) {
|
||||
let mut active_input_audio_streams_producers: InputAudioStreamsProducers = HashMap::new();
|
||||
let active_input_audio_streams_consumers: InputAudioStreamsConsumers =
|
||||
Arc::new(Mutex::new(HashMap::new()));
|
||||
let mut active_input_audio_streams_decoders: InputAudioStreamsDecoders = HashMap::new();
|
||||
let mut active_input_audio_streams_resamplers: InputAudioStreamsResamplers = HashMap::new();
|
||||
|
||||
let mut last_frame_number: HashMap<AudioStreamKey, Wrapping<u8>> = HashMap::new();
|
||||
|
||||
let output_stream = match output_device_config.sample_format() {
|
||||
SampleFormat::F64 => cpal_build_output_stream::<f64>(
|
||||
&output_device,
|
||||
output_device_config.into(),
|
||||
active_input_audio_streams_consumers,
|
||||
active_input_audio_streams_consumers.clone(),
|
||||
),
|
||||
SampleFormat::I64 => cpal_build_output_stream::<i64>(
|
||||
&output_device,
|
||||
output_device_config.into(),
|
||||
active_input_audio_streams_consumers,
|
||||
active_input_audio_streams_consumers.clone(),
|
||||
),
|
||||
SampleFormat::U64 => cpal_build_output_stream::<u64>(
|
||||
&output_device,
|
||||
output_device_config.into(),
|
||||
active_input_audio_streams_consumers,
|
||||
active_input_audio_streams_consumers.clone(),
|
||||
),
|
||||
SampleFormat::F32 => cpal_build_output_stream::<f32>(
|
||||
&output_device,
|
||||
output_device_config.into(),
|
||||
active_input_audio_streams_consumers,
|
||||
active_input_audio_streams_consumers.clone(),
|
||||
),
|
||||
SampleFormat::I32 => cpal_build_output_stream::<i32>(
|
||||
&output_device,
|
||||
output_device_config.into(),
|
||||
active_input_audio_streams_consumers,
|
||||
active_input_audio_streams_consumers.clone(),
|
||||
),
|
||||
SampleFormat::U32 => cpal_build_output_stream::<u32>(
|
||||
&output_device,
|
||||
output_device_config.into(),
|
||||
active_input_audio_streams_consumers,
|
||||
active_input_audio_streams_consumers.clone(),
|
||||
),
|
||||
SampleFormat::I16 => cpal_build_output_stream::<i16>(
|
||||
&output_device,
|
||||
output_device_config.into(),
|
||||
active_input_audio_streams_consumers,
|
||||
active_input_audio_streams_consumers.clone(),
|
||||
),
|
||||
SampleFormat::U16 => cpal_build_output_stream::<u16>(
|
||||
&output_device,
|
||||
output_device_config.into(),
|
||||
active_input_audio_streams_consumers,
|
||||
active_input_audio_streams_consumers.clone(),
|
||||
),
|
||||
SampleFormat::I8 => cpal_build_output_stream::<i8>(
|
||||
&output_device,
|
||||
output_device_config.into(),
|
||||
active_input_audio_streams_consumers,
|
||||
active_input_audio_streams_consumers.clone(),
|
||||
),
|
||||
SampleFormat::U8 => cpal_build_output_stream::<u8>(
|
||||
&output_device,
|
||||
output_device_config.into(),
|
||||
active_input_audio_streams_consumers,
|
||||
active_input_audio_streams_consumers.clone(),
|
||||
),
|
||||
_ => {
|
||||
eprintln!("unsupported output stream format :(");
|
||||
@@ -122,23 +245,33 @@ pub async fn audio_stream_receiver(
|
||||
|
||||
let mut receive_buffer = vec![0u8; 4096];
|
||||
|
||||
let mut decoded_samples = [0.0f32; 1024];
|
||||
let 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,
|
||||
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],
|
||||
&mut resampled_samples,
|
||||
&mut decoded_samples_converted,
|
||||
ropus::DecodeMode::Fec,
|
||||
);
|
||||
} 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],
|
||||
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 {
|
||||
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],
|
||||
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<ClientEvent>,
|
||||
udp_socket: Arc<UdpSocket>,
|
||||
client_info: Arc<RwLock<ClientInfo>>,
|
||||
force_channels: u16,
|
||||
) {
|
||||
let current_stream_id: u8 = rand::random();
|
||||
|
||||
@@ -250,10 +390,10 @@ pub async fn audio_stream_spawner(
|
||||
};
|
||||
if connected {
|
||||
let _ = global_event_tx.send(ClientEvent::NewAudioStreamRequest(AudioStreamInfo {
|
||||
client_id: client_id,
|
||||
client_id,
|
||||
stream_id: current_stream_id,
|
||||
channels: input_device_config.channels(),
|
||||
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::<f32>::new(ring_buffer_size);
|
||||
let (producer, mut consumer) = ring_buffer.split();
|
||||
|
||||
@@ -319,43 +458,97 @@ pub async fn audio_stream_spawner(
|
||||
input_stream.play().expect("failed to play input stream");
|
||||
|
||||
let frames_needed = ((20.0 / 1000.0) * input_device_config.sample_rate() as f32) as usize;
|
||||
let mut frame_buffer = vec![0.0; frames_needed];
|
||||
|
||||
let channels: ropus::Channels = {
|
||||
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::<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(
|
||||
input_device_config.sample_rate(),
|
||||
channels,
|
||||
ropus::Application::Voip,
|
||||
)
|
||||
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::<u8>::with_capacity(1 + 4 + 1 + encoded_len * 4);
|
||||
|
||||
let (server_address, client_id) = {
|
||||
let info = client_info.read().unwrap();
|
||||
(info.server_address.clone(), info.client_id)
|
||||
};
|
||||
|
||||
payload.extend(client_id.to_be_bytes());
|
||||
payload.extend(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]);
|
||||
|
||||
|
||||
+5
-21
@@ -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::<ClientEvent>(16);
|
||||
|
||||
let active_input_audio_streams_producers: InputAudioStreamsProducers =
|
||||
Arc::new(Mutex::new(HashMap::new()));
|
||||
let active_input_audio_streams_consumers: InputAudioStreamsConsumers =
|
||||
Arc::new(Mutex::new(HashMap::new()));
|
||||
|
||||
let active_input_audio_streams_encoders: InputAudioStreamsEncoders =
|
||||
Arc::new(RwLock::new(HashMap::new()));
|
||||
|
||||
let active_input_audio_streams_info: InputAudioStreamsInfo =
|
||||
Arc::new(RwLock::new(HashMap::new()));
|
||||
|
||||
let udp_socket = Arc::new(UdpSocket::bind("0.0.0.0:0").await.unwrap());
|
||||
|
||||
let audio_control: Arc<AudioControl> = Arc::new(AudioControl {
|
||||
active_input_audio_streams_producers: active_input_audio_streams_producers.clone(),
|
||||
active_input_audio_streams_consumers: active_input_audio_streams_consumers.clone(),
|
||||
active_input_audio_streams_encoders: active_input_audio_streams_encoders.clone(),
|
||||
active_input_audio_streams_info: active_input_audio_streams_info.clone(),
|
||||
udp_socket: udp_socket.clone(),
|
||||
});
|
||||
@@ -86,21 +72,19 @@ async fn main() {
|
||||
|
||||
tokio::spawn(audio_stream_receiver(
|
||||
output_device,
|
||||
output_device_config.into(),
|
||||
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");
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
use clap::Parser;
|
||||
use ringbuf::{HeapRb, wrap::caching::Caching};
|
||||
use ropus::Decoder;
|
||||
use rubato::Fft;
|
||||
use shared::AudioStreamInfo;
|
||||
use std::{
|
||||
collections::HashMap,
|
||||
@@ -25,6 +26,7 @@ pub enum TuiScreen {
|
||||
Exit,
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
#[derive(Clone)]
|
||||
pub enum ClientEvent {
|
||||
NewConnection,
|
||||
@@ -38,21 +40,19 @@ pub struct ClientInfo {
|
||||
pub connected: bool,
|
||||
}
|
||||
|
||||
type AudioStreamKey = (u32, u8);
|
||||
pub type AudioStreamKey = (u32, u8);
|
||||
|
||||
pub type InputAudioStreamsProducers =
|
||||
Arc<Mutex<HashMap<AudioStreamKey, Caching<Arc<HeapRb<f32>>, true, false>>>>;
|
||||
HashMap<AudioStreamKey, Caching<Arc<HeapRb<f32>>, true, false>>;
|
||||
pub type InputAudioStreamsConsumers =
|
||||
Arc<Mutex<HashMap<AudioStreamKey, Caching<Arc<HeapRb<f32>>, false, true>>>>;
|
||||
|
||||
pub type InputAudioStreamsEncoders = Arc<RwLock<HashMap<AudioStreamKey, Mutex<Decoder>>>>;
|
||||
pub type InputAudioStreamsDecoders = HashMap<AudioStreamKey, Decoder>;
|
||||
pub type InputAudioStreamsResamplers = HashMap<AudioStreamKey, Fft<f32>>;
|
||||
|
||||
pub type InputAudioStreamsInfo = Arc<RwLock<HashMap<AudioStreamKey, AudioStreamInfo>>>;
|
||||
|
||||
pub struct AudioControl {
|
||||
pub active_input_audio_streams_producers: InputAudioStreamsProducers,
|
||||
pub active_input_audio_streams_consumers: InputAudioStreamsConsumers,
|
||||
pub active_input_audio_streams_encoders: InputAudioStreamsEncoders,
|
||||
pub active_input_audio_streams_info: InputAudioStreamsInfo,
|
||||
pub udp_socket: Arc<UdpSocket>,
|
||||
}
|
||||
|
||||
@@ -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::<TcpResponse>(&bytes) {
|
||||
Ok(response) => {
|
||||
println!("connection Ok: {}", response.status == ResponseStatus::Ok);
|
||||
// println!("connection Ok: {}", response.status == ResponseStatus::Ok);
|
||||
match response.data {
|
||||
TcpResponseData::NewConnection { client_id } => {
|
||||
let mut client_info_var = client_info.write().unwrap();
|
||||
client_info_var.client_id = client_id;
|
||||
client_info_var.connected = true;
|
||||
println!("fetched client id: {}", client_id);
|
||||
// println!("fetched client id: {}", client_id);
|
||||
}
|
||||
TcpResponseData::UserConnected { user_id } => {
|
||||
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::<f32>::new(ring_buffer_size);
|
||||
let (producer, consumer) = ring_buffer.split();
|
||||
|
||||
producer_map.insert(
|
||||
(audio_stream.client_id, audio_stream.stream_id),
|
||||
producer,
|
||||
);
|
||||
consumer_map.insert(
|
||||
(audio_stream.client_id, audio_stream.stream_id),
|
||||
consumer,
|
||||
);
|
||||
|
||||
let channels: ropus::Channels = {
|
||||
if audio_stream.channels == 1 {
|
||||
ropus::Channels::Mono
|
||||
} else {
|
||||
ropus::Channels::Stereo
|
||||
}
|
||||
};
|
||||
|
||||
let decoder = Decoder::new(audio_stream.sample_rate, channels).expect("failed to creat decoder");
|
||||
encoder_map.insert(
|
||||
(audio_stream.client_id, audio_stream.stream_id),
|
||||
Mutex::new(decoder),
|
||||
);
|
||||
|
||||
info_map.insert(
|
||||
(audio_stream.client_id, audio_stream.stream_id),
|
||||
audio_stream,
|
||||
);
|
||||
}
|
||||
println!("audio streams update");
|
||||
// println!("audio streams update");
|
||||
}
|
||||
TcpResponseData::NewAudioStreamAccepted { audio_stream_id } => {
|
||||
let _ = global_event_tx.send(ClientEvent::NewAudioStreamAccepted(audio_stream_id));
|
||||
@@ -135,13 +94,13 @@ pub async fn tcp_connection(
|
||||
})
|
||||
.expect("failed to parse initial request");
|
||||
|
||||
let _ = framed_tcp_stream.send(Bytes::from(request_json))
|
||||
framed_tcp_stream.send(Bytes::from(request_json))
|
||||
.await
|
||||
.expect("failed to send initial message");
|
||||
|
||||
}
|
||||
ClientEvent::NewAudioStreamRequest(audio_stream_info) => {
|
||||
println!("new audio stream request");
|
||||
// println!("new audio stream request");
|
||||
let session_client_id = client_info.read().unwrap().client_id;
|
||||
|
||||
let request_json = serde_json::to_string(&TcpRequest {
|
||||
@@ -151,7 +110,7 @@ pub async fn tcp_connection(
|
||||
}
|
||||
}).expect("failed to parse initial request");
|
||||
|
||||
let _ = framed_tcp_stream.send(Bytes::from(request_json))
|
||||
framed_tcp_stream.send(Bytes::from(request_json))
|
||||
.await
|
||||
.expect("failed to send initial message");
|
||||
}
|
||||
|
||||
+40
-97
@@ -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)
|
||||
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));
|
||||
}
|
||||
} else if let Ok(supported_configs) = default_input_device.supported_input_configs() {
|
||||
for supported_config in supported_configs {
|
||||
if let Some(input_config) =
|
||||
supported_config.try_with_sample_rate(SAMPLE_RATE_48K)
|
||||
{
|
||||
if input_config.channels() == 1 {
|
||||
input_devices.push((default_input_device, input_config));
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if let Ok(possible_input_devices) = host.input_devices() {
|
||||
for possible_input_device in possible_input_devices {
|
||||
if let Ok(default_config) = possible_input_device.default_input_config() {
|
||||
if default_config.sample_rate() == SAMPLE_RATE_48K
|
||||
&& (default_config.channels() == 1)
|
||||
{
|
||||
input_devices.push((possible_input_device.clone(), default_config));
|
||||
continue;
|
||||
}
|
||||
}
|
||||
if let Ok(supported_configs) = possible_input_device.supported_input_configs() {
|
||||
for supported_config in supported_configs {
|
||||
if let Some(input_config) =
|
||||
supported_config.try_with_sample_rate(SAMPLE_RATE_48K)
|
||||
{
|
||||
if input_config.channels() == 1 {
|
||||
input_devices.push((possible_input_device, input_config));
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let mut output_devices: Vec<(Device, SupportedStreamConfig)> = Vec::new();
|
||||
|
||||
if let Some(default_output_device) = host.default_output_device() {
|
||||
if let 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)
|
||||
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.into()));
|
||||
}
|
||||
} else if let Ok(supported_configs) = default_output_device.supported_output_configs() {
|
||||
for supported_config in supported_configs {
|
||||
if let Some(output_config) =
|
||||
supported_config.try_with_sample_rate(SAMPLE_RATE_48K)
|
||||
{
|
||||
if output_config.channels() == 1 || output_config.channels() == 2 {
|
||||
output_devices.push((default_output_device, output_config.into()));
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
output_devices.push((default_output_device.clone(), default_config));
|
||||
}
|
||||
|
||||
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,12 +195,19 @@ impl DeviceSelectScreen {
|
||||
let widths = [Constraint::Percentage(100), Constraint::Min(8)];
|
||||
|
||||
let mut input_device_rows: Vec<Row> = Vec::new();
|
||||
for (device, config) in self.input_devices.iter() {
|
||||
for (i, (device, config)) in self.input_devices.iter().enumerate() {
|
||||
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.sample_format().to_string(),
|
||||
config.channels().to_string(),
|
||||
]));
|
||||
}
|
||||
}
|
||||
let input_device_table = Table::new(input_device_rows, widths)
|
||||
.header(
|
||||
Row::new(vec!["device name", "channels"])
|
||||
@@ -296,12 +237,22 @@ impl DeviceSelectScreen {
|
||||
frame.render_stateful_widget(input_device_table, left_block, &mut self.input_device_state);
|
||||
|
||||
let mut output_device_rows: Vec<Row> = Vec::new();
|
||||
for (device, config) in self.output_devices.iter() {
|
||||
for (i, (device, config)) in self.output_devices.iter().enumerate() {
|
||||
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.sample_format().to_string(),
|
||||
config.channels().to_string(),
|
||||
]));
|
||||
}
|
||||
}
|
||||
let output_device_table = Table::new(output_device_rows, widths)
|
||||
.header(
|
||||
Row::new(vec!["device name", "channels"])
|
||||
@@ -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;
|
||||
|
||||
+17
-35
@@ -23,10 +23,9 @@ struct Client {
|
||||
audio_streams: Option<Vec<AudioStreamInfo>>,
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
#[derive(Clone)]
|
||||
enum ServerEvent {
|
||||
NewAudioStream(AudioStreamInfo),
|
||||
SendAllStreams,
|
||||
SendAllOtherStreams(u32),
|
||||
UserConnected(u32),
|
||||
UserDisconnected(u32),
|
||||
@@ -36,13 +35,13 @@ type State = Arc<RwLock<HashMap<u32, Client>>>;
|
||||
|
||||
fn scrape_all_client_streams_info(clients: &HashMap<u32, Client>) -> Vec<AudioStreamInfo> {
|
||||
let mut all_audio_streams_info: Vec<AudioStreamInfo> = Vec::new();
|
||||
for (_, data) in clients {
|
||||
for data in clients.values() {
|
||||
if let Some(audio_streams_info) = &data.audio_streams {
|
||||
all_audio_streams_info.extend(audio_streams_info.iter().cloned());
|
||||
}
|
||||
}
|
||||
|
||||
return all_audio_streams_info;
|
||||
all_audio_streams_info
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
@@ -70,25 +69,25 @@ async fn main() {
|
||||
let mut buf = vec![0u8; 4096];
|
||||
|
||||
loop {
|
||||
let (len, src_adddress) = udp_socket.recv_from(&mut buf).await.expect("receive error");
|
||||
let (len, src_address) = udp_socket.recv_from(&mut buf).await.expect("receive error");
|
||||
|
||||
if len < 4 {
|
||||
continue;
|
||||
}
|
||||
|
||||
let id_bytes: [u8; 4] = buf[0..4].try_into().unwrap();
|
||||
let client_id = u32::from_be_bytes(id_bytes);
|
||||
let client_id = u32::from_le_bytes(id_bytes);
|
||||
|
||||
let mut clients = udp_state.write().await;
|
||||
if let Some(client) = clients.get_mut(&client_id) {
|
||||
client.udp_addr = Some(src_adddress);
|
||||
client.udp_addr = Some(src_address);
|
||||
}
|
||||
|
||||
for (&other_client_id, other_client) in clients.iter() {
|
||||
if other_client_id != client_id {
|
||||
if 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),
|
||||
|
||||
Reference in New Issue
Block a user