1pub mod layer {
2 use crate::audio::{RingBuffer, buffer::PooledBuffer, constants::LAYER_BUFFER_SIZE};
3 use flume::Receiver;
4 pub struct MixLayer {
5 pub id: String,
6 pub rx: Receiver<PooledBuffer>,
7 pub ring_buffer: RingBuffer,
8 pub volume: f32,
9 pub finished: bool,
10 }
11 impl MixLayer {
12 pub fn new(id: String, rx: Receiver<PooledBuffer>, volume: f32) -> Self {
13 Self {
14 id,
15 rx,
16 ring_buffer: RingBuffer::new(LAYER_BUFFER_SIZE),
17 volume: volume.clamp(0.0, 1.0),
18 finished: false,
19 }
20 }
21 pub fn fill(&mut self) {
22 while let Ok(pooled) = self.rx.try_recv() {
23 let bytes = unsafe {
24 std::slice::from_raw_parts(pooled.as_ptr() as *const u8, pooled.len() * 2)
25 };
26 self.ring_buffer.write(bytes);
27 crate::audio::buffer::release_buffer(pooled);
28 }
29 if self.rx.is_disconnected() {
30 self.finished = true;
31 }
32 }
33 pub fn is_dead(&self) -> bool {
34 self.finished && self.ring_buffer.is_empty()
35 }
36 pub fn accumulate(&mut self, acc: &mut [i32]) {
37 let byte_count = acc.len() * 2;
38 if let Some(bytes) = self.ring_buffer.read(byte_count) {
39 let samples = unsafe {
40 std::slice::from_raw_parts(bytes.as_ptr() as *const i16, bytes.len() / 2)
41 };
42 for (acc_val, &s) in acc.iter_mut().zip(samples.iter()) {
43 *acc_val += (s as f32 * self.volume).round() as i32;
44 }
45 }
46 }
47 }
48}
49pub mod mixer {
50 use super::layer::MixLayer;
51 use crate::{
52 audio::{
53 AudioFrame,
54 buffer::PooledBuffer,
55 constants::{MAX_LAYERS, MIXER_CHANNELS, TARGET_SAMPLE_RATE},
56 flow::FlowController,
57 playback::{StuckDetector, handle::PlaybackState},
58 },
59 config::player::PlayerConfig,
60 };
61 use flume::Receiver;
62 use std::{
63 collections::HashMap,
64 sync::{
65 Arc,
66 atomic::{AtomicBool, AtomicU8, AtomicU32, AtomicU64, Ordering},
67 },
68 };
69 pub struct AudioMixer {
70 pub layers: HashMap<String, MixLayer>,
71 pub max_layers: usize,
72 pub enabled: bool,
73 acc_buf: Vec<i32>,
74 }
75 impl Default for AudioMixer {
76 fn default() -> Self {
77 Self::new()
78 }
79 }
80 impl AudioMixer {
81 pub fn new() -> Self {
82 Self {
83 layers: HashMap::new(),
84 max_layers: MAX_LAYERS,
85 enabled: true,
86 acc_buf: Vec::with_capacity(1920),
87 }
88 }
89 pub fn add_layer(
90 &mut self,
91 id: String,
92 rx: Receiver<PooledBuffer>,
93 volume: f32,
94 ) -> Result<(), &'static str> {
95 if self.layers.len() >= MAX_LAYERS {
96 return Err("Maximum mix layers reached");
97 }
98 self.layers
99 .insert(id.clone(), MixLayer::new(id, rx, volume));
100 Ok(())
101 }
102 pub fn remove_layer(&mut self, id: &str) {
103 self.layers.remove(id);
104 }
105 pub fn set_layer_volume(&mut self, id: &str, volume: f32) {
106 if let Some(layer) = self.layers.get_mut(id) {
107 layer.volume = volume.clamp(0.0, 1.0);
108 }
109 }
110 pub fn mix(&mut self, main_frame: &mut [i16]) {
111 if !self.enabled || self.layers.is_empty() {
112 return;
113 }
114 let out_len = main_frame.len();
115 if self.acc_buf.len() != out_len {
116 self.acc_buf.resize(out_len, 0);
117 }
118 for (acc, &sample) in self.acc_buf.iter_mut().zip(main_frame.iter()) {
119 *acc = sample as i32;
120 }
121 self.layers.retain(|_, layer| {
122 layer.fill();
123 !layer.is_dead()
124 });
125 for layer in self.layers.values_mut() {
126 layer.accumulate(&mut self.acc_buf);
127 }
128 for (out, &sum) in main_frame.iter_mut().zip(self.acc_buf.iter()) {
129 *out = sum.clamp(i16::MIN as i32, i16::MAX as i32) as i16;
130 }
131 }
132 }
133 pub struct Mixer {
134 tracks: Vec<MixerTrack>,
135 mix_buf: Vec<i32>,
136 pub audio_mixer: AudioMixer,
137 opus_passthrough_track: Option<usize>,
138 final_pcm_buf: Vec<i16>,
139 pub stuck_detector: Arc<StuckDetector>,
140 }
141 struct MixerTrack {
142 flow: FlowController,
143 pending: Vec<i16>,
144 pending_pos: usize,
145 state: Arc<AtomicU8>,
146 volume: Arc<AtomicU32>,
147 position: Arc<AtomicU64>,
148 is_buffering: Arc<AtomicBool>,
149 config: PlayerConfig,
150 finished: bool,
151 }
152 impl Mixer {
153 pub fn new(_sample_rate: u32) -> Self {
154 Self {
155 tracks: Vec::new(),
156 mix_buf: Vec::with_capacity(1920),
157 audio_mixer: AudioMixer::new(),
158 opus_passthrough_track: None,
159 final_pcm_buf: Vec::with_capacity(1920),
160 stuck_detector: Arc::new(StuckDetector::new(10_000)),
161 }
162 }
163 pub fn add_track(
164 &mut self,
165 rx: Receiver<AudioFrame>,
166 state: Arc<AtomicU8>,
167 volume: Arc<AtomicU32>,
168 position: Arc<AtomicU64>,
169 is_buffering: Arc<AtomicBool>,
170 config: PlayerConfig,
171 ) {
172 let vol_raw = f32::from_bits(volume.load(Ordering::Acquire));
173 let mut flow =
174 FlowController::for_mixer(rx, TARGET_SAMPLE_RATE, MIXER_CHANNELS, vol_raw);
175 flow.volume.set_volume_instant(vol_raw);
176 self.tracks.push(MixerTrack {
177 flow,
178 pending: Vec::new(),
179 pending_pos: 0,
180 state,
181 volume,
182 position,
183 is_buffering,
184 config,
185 finished: false,
186 });
187 }
188 pub fn set_passthrough_track(&mut self, track_index: usize) {
189 self.opus_passthrough_track = Some(track_index);
190 }
191 pub fn take_opus_frame(&mut self) -> Option<Vec<u8>> {
192 for track in self.tracks.iter_mut() {
193 let state = PlaybackState::from(track.state.load(Ordering::Acquire));
194 if matches!(
195 state,
196 PlaybackState::Paused
197 | PlaybackState::Stopped
198 | PlaybackState::Stopping
199 | PlaybackState::Starting
200 ) {
201 continue;
202 }
203 if let Some(packet) = track.flow.take_opus() {
204 track.position.fetch_add(960, Ordering::Relaxed);
205 return Some(packet);
206 }
207 }
208 None
209 }
210 pub fn stop_all(&mut self) {
211 for track in self.tracks.iter_mut() {
212 track
213 .state
214 .store(PlaybackState::Stopped as u8, Ordering::Release);
215 }
216 self.tracks.clear();
217 self.audio_mixer.enabled = false;
218 }
219 pub fn mix(&mut self, buf: &mut [i16]) -> bool {
220 let out_len = buf.len();
221 if self.mix_buf.len() != out_len {
222 self.mix_buf.resize(out_len, 0);
223 }
224 self.mix_buf.fill(0);
225 self.tracks
226 .retain(|t| t.state.load(Ordering::Acquire) != PlaybackState::Stopped as u8);
227 let mut has_audio = false;
228 for track in self.tracks.iter_mut() {
229 let state = PlaybackState::from(track.state.load(Ordering::Acquire));
230 if matches!(state, PlaybackState::Paused | PlaybackState::Stopped) {
231 continue;
232 }
233 let vol_f = f32::from_bits(track.volume.load(Ordering::Acquire));
234 if (vol_f - track.flow.volume.current_volume()).abs() > 0.001 {
235 track.flow.volume.set_volume(vol_f);
236 }
237 if state == PlaybackState::Stopping && !track.flow.tape.is_ramping() {
238 track.flow.tape.tape_to(
239 track.config.tape.tape_stop_duration_ms as f32,
240 false,
241 track.config.tape.curve,
242 );
243 } else if state == PlaybackState::Starting && !track.flow.tape.is_ramping() {
244 track.flow.tape.tape_to(
245 track.config.tape.tape_stop_duration_ms as f32,
246 true,
247 track.config.tape.curve,
248 );
249 }
250 let mut filled = 0usize;
251 if track.pending_pos < track.pending.len() {
252 let n = (out_len - filled).min(track.pending.len() - track.pending_pos);
253 for (acc, &s) in self.mix_buf[filled..filled + n]
254 .iter_mut()
255 .zip(&track.pending[track.pending_pos..track.pending_pos + n])
256 {
257 *acc += s as i32;
258 }
259 track.pending_pos += n;
260 filled += n;
261 if track.pending_pos >= track.pending.len() {
262 track.pending.clear();
263 track.pending_pos = 0;
264 }
265 }
266 'pull: while filled < out_len && !track.finished {
267 match track.flow.try_pop_frame() {
268 Ok(Some(frame)) => {
269 let n = frame.len().min(out_len - filled);
270 for (acc, &s) in
271 self.mix_buf[filled..filled + n].iter_mut().zip(&frame[..n])
272 {
273 *acc += s as i32;
274 }
275 if n < frame.len() {
276 track.pending.extend_from_slice(&frame[n..]);
277 track.pending_pos = 0;
278 }
279 filled += n;
280 crate::audio::buffer::release_buffer(frame);
281 }
282 Ok(None) => break 'pull,
283 Err(_) => {
284 track.finished = true;
285 break 'pull;
286 }
287 }
288 }
289 if filled > 0 {
290 has_audio = true;
291 track
292 .position
293 .fetch_add(filled as u64 / MIXER_CHANNELS as u64, Ordering::Relaxed);
294 track.is_buffering.store(false, Ordering::Release);
295 self.stuck_detector.record_frame_received();
296 } else if !track.finished {
297 track.is_buffering.store(true, Ordering::Release);
298 }
299 if track.finished && track.pending.is_empty() && !track.flow.tape.is_active() {
300 track
301 .state
302 .store(PlaybackState::Stopped as u8, Ordering::Release);
303 }
304 if track.flow.tape.check_ramp_completed() {
305 match state {
306 PlaybackState::Stopping => {
307 track
308 .state
309 .store(PlaybackState::Paused as u8, Ordering::Release);
310 }
311 PlaybackState::Starting => {
312 track
313 .state
314 .store(PlaybackState::Playing as u8, Ordering::Release);
315 }
316 _ => {}
317 }
318 }
319 }
320 if self.final_pcm_buf.len() != out_len {
321 self.final_pcm_buf.resize(out_len, 0);
322 }
323 for (final_pcm, &sum) in self.final_pcm_buf.iter_mut().zip(self.mix_buf.iter()) {
324 *final_pcm = sum.clamp(i16::MIN as i32, i16::MAX as i32) as i16;
325 }
326 self.audio_mixer.mix(&mut self.final_pcm_buf);
327 if !self.audio_mixer.layers.is_empty() {
328 has_audio = true;
329 }
330 buf.copy_from_slice(&self.final_pcm_buf);
331 has_audio
332 }
333 }
334}
335pub use layer::MixLayer;
336pub use mixer::{AudioMixer, Mixer};