1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
|
use crate::network::VoiceStreamType;
use log::*;
use cpal::{OutputCallbackInfo, Sample};
use mumble_protocol::voice::VoicePacketPayload;
use opus::Channels;
use std::collections::{HashMap, VecDeque, hash_map::Entry};
use std::ops::AddAssign;
use std::sync::{Arc, Mutex};
use tokio::sync::watch;
type ClientStreamKey = Option<(VoiceStreamType, u32)>;
type ConsumerInput<'a> = std::collections::hash_map::IterMut<'a, ClientStreamKey, VecDeque<f32>>;
pub struct ClientStream {
buffer: HashMap<ClientStreamKey, VecDeque<f32>>, //TODO ring buffer?
opus_decoder: opus::Decoder,
}
impl ClientStream {
pub fn new(sample_rate: u32, channels: u16) -> Self {
let mut buffer = HashMap::new();
//None is the system audio effects
buffer.insert(None, VecDeque::new());
Self {
buffer,
opus_decoder: opus::Decoder::new(
sample_rate,
match channels {
1 => Channels::Mono,
2 => Channels::Stereo,
_ => unimplemented!("Only 1 or 2 channels supported, got {}", channels),
},
)
.unwrap(),
}
}
pub fn decode_packet(&mut self, client: ClientStreamKey, payload: VoicePacketPayload, channels: usize) {
match payload {
VoicePacketPayload::Opus(bytes, _eot) => {
let mut out: Vec<f32> = vec![0.0; 720 * channels * 4]; //720 is because that is the max size of packet we can get that we want to decode
let parsed = self
.opus_decoder
.decode_float(&bytes, &mut out, false)
.expect("Error decoding");
out.truncate(parsed);
self.extend(client, &out);
}
_ => {
unimplemented!("Payload type not supported");
}
}
}
pub fn consume<'a, F>(&mut self, consumer: F)
where
F: FnOnce(ConsumerInput) {
let iter = self.buffer.iter_mut();
consumer(iter);
}
pub fn extend(&mut self, key: ClientStreamKey, values: &[f32]) {
match self.buffer.entry(key) {
Entry::Occupied(mut entry) => {
entry.get_mut().extend(values.iter().copied());
}
Entry::Vacant(_) => {
match key {
None => warn!("Can't find session None"),
Some(key) => warn!("Can't find session id {}", key.1),
}
}
}
}
pub fn add_client(&mut self, session_id: u32) {
for stream_type in [VoiceStreamType::TCP, VoiceStreamType::UDP].iter() {
match self.buffer.entry(Some((*stream_type, session_id))) {
Entry::Occupied(_) => {
warn!("Session id {} already exists", session_id);
}
Entry::Vacant(entry) => {
entry.insert(VecDeque::new());
}
}
}
}
pub fn remove_client(&mut self, session_id: u32) {
for stream_type in [VoiceStreamType::TCP, VoiceStreamType::UDP].iter() {
match self.buffer.entry(Some((*stream_type, session_id))) {
Entry::Occupied(entry) => {
entry.remove();
}
Entry::Vacant(_) => {
warn!(
"Tried to remove session id {} that doesn't exist",
session_id
);
}
}
}
}
pub fn clear_clients(&mut self) {
self.buffer.retain(|k , _| k.is_none());
}
}
pub trait SaturatingAdd {
fn saturating_add(self, rhs: Self) -> Self;
}
impl SaturatingAdd for f32 {
fn saturating_add(self, rhs: Self) -> Self {
match self + rhs {
a if a < -1.0 => -1.0,
a if a > 1.0 => 1.0,
a => a,
}
}
}
impl SaturatingAdd for i16 {
fn saturating_add(self, rhs: Self) -> Self {
i16::saturating_add(self, rhs)
}
}
impl SaturatingAdd for u16 {
fn saturating_add(self, rhs: Self) -> Self {
u16::saturating_add(self, rhs)
}
}
pub fn curry_callback<T: Sample + AddAssign + SaturatingAdd + std::fmt::Display>(
user_bufs: Arc<Mutex<ClientStream>>,
output_volume_receiver: watch::Receiver<f32>,
user_volumes: Arc<Mutex<HashMap<u32, (f32, bool)>>>,
) -> impl FnMut(&mut [T], &OutputCallbackInfo) + Send + 'static {
move |data: &mut [T], _info: &OutputCallbackInfo| {
for sample in data.iter_mut() {
*sample = Sample::from(&0.0);
}
let volume = *output_volume_receiver.borrow();
let mut user_bufs = user_bufs.lock().unwrap();
let user_volumes = user_volumes.lock().unwrap();
let mut saturating_add = |input: ConsumerInput| {
for (k, v) in input {
let (user_volume, muted) = match k {
Some((_, id)) => user_volumes.get(id).cloned().unwrap_or((1.0, false)),
None => (1.0, false),
};
for sample in data.iter_mut() {
if !muted {
*sample = sample.saturating_add(Sample::from(
&(v.pop_front().unwrap_or(0.0) * volume * user_volume),
));
}
}
}
};
user_bufs.consume(&mut saturating_add);
}
}
|
