lavende_core/audio/
flow.rs

1pub mod controller {
2    use crate::audio::{
3        AudioFrame,
4        buffer::{PooledBuffer, acquire_buffer},
5        constants::FRAME_SIZE_SAMPLES,
6        effects::{
7            crossfade::CrossfadeController, fade::FadeEffect, tape::TapeEffect,
8            volume::VolumeEffect,
9        },
10        error::AudioError,
11    };
12    use flume::{Receiver, Sender};
13    pub struct FlowController {
14        pub tape: TapeEffect,
15        pub volume: VolumeEffect,
16        pub fade: FadeEffect,
17        pub crossfade: CrossfadeController,
18        pending_pcm: Vec<i16>,
19        decoder_done: bool,
20        frame_rx: Receiver<AudioFrame>,
21        frame_tx: Option<Sender<AudioFrame>>,
22        latest_opus: Option<Vec<u8>>,
23    }
24    impl FlowController {
25        pub fn new(
26            frame_rx: Receiver<AudioFrame>,
27            frame_tx: Sender<AudioFrame>,
28            sample_rate: u32,
29            channels: usize,
30            volume: f32,
31        ) -> Self {
32            Self::build(frame_rx, Some(frame_tx), sample_rate, channels, volume)
33        }
34        pub fn for_mixer(
35            frame_rx: Receiver<AudioFrame>,
36            sample_rate: u32,
37            channels: usize,
38            volume: f32,
39        ) -> Self {
40            Self::build(frame_rx, None, sample_rate, channels, volume)
41        }
42        fn build(
43            frame_rx: Receiver<AudioFrame>,
44            frame_tx: Option<Sender<AudioFrame>>,
45            sample_rate: u32,
46            channels: usize,
47            volume: f32,
48        ) -> Self {
49            Self {
50                tape: TapeEffect::new(sample_rate, channels),
51                volume: VolumeEffect::new(volume, sample_rate, channels),
52                fade: FadeEffect::new(1.0, channels),
53                crossfade: CrossfadeController::new(sample_rate, channels),
54                pending_pcm: Vec::with_capacity(FRAME_SIZE_SAMPLES * 2),
55                decoder_done: false,
56                frame_rx,
57                frame_tx,
58                latest_opus: None,
59            }
60        }
61        pub fn run(&mut self) {
62            while let Ok(frame_data) = self.frame_rx.recv() {
63                match frame_data {
64                    AudioFrame::Pcm(pooled) => {
65                        if pooled.is_empty() {
66                            self.pending_pcm.clear();
67                            continue;
68                        }
69                        self.pending_pcm.extend_from_slice(&pooled);
70                        while self.pending_pcm.len() >= FRAME_SIZE_SAMPLES {
71                            let mut frame = acquire_buffer(FRAME_SIZE_SAMPLES);
72                            frame.extend(self.pending_pcm.drain(..FRAME_SIZE_SAMPLES));
73                            self.process_frame(&mut frame);
74                            if self
75                                .frame_tx
76                                .as_ref()
77                                .is_some_and(|tx| tx.send(AudioFrame::Pcm(frame)).is_err())
78                            {
79                                return;
80                            }
81                        }
82                    }
83                    AudioFrame::Opus(packet) => {
84                        if let Some(tx) = &self.frame_tx
85                            && tx.send(AudioFrame::Opus(packet)).is_err()
86                        {
87                            return;
88                        }
89                    }
90                }
91            }
92        }
93        pub fn try_pop_frame(&mut self) -> Result<Option<PooledBuffer>, AudioError> {
94            if !self.decoder_done {
95                while self.pending_pcm.len() < FRAME_SIZE_SAMPLES {
96                    match self.frame_rx.try_recv() {
97                        Ok(AudioFrame::Pcm(chunk)) if chunk.is_empty() => {
98                            self.pending_pcm.clear();
99                            self.decoder_done = false;
100                        }
101                        Ok(AudioFrame::Pcm(chunk)) => {
102                            self.pending_pcm.extend_from_slice(&chunk);
103                            crate::audio::buffer::release_buffer(chunk);
104                        }
105                        Ok(AudioFrame::Opus(packet)) => {
106                            self.latest_opus = Some(packet);
107                        }
108                        Err(flume::TryRecvError::Empty) => break,
109                        Err(flume::TryRecvError::Disconnected) => {
110                            self.decoder_done = true;
111                            break;
112                        }
113                    }
114                }
115            }
116            if self.pending_pcm.len() >= FRAME_SIZE_SAMPLES {
117                let mut frame = acquire_buffer(FRAME_SIZE_SAMPLES);
118                frame.extend(self.pending_pcm.drain(..FRAME_SIZE_SAMPLES));
119                self.process_frame(&mut frame);
120                Ok(Some(frame))
121            } else if self.decoder_done {
122                if !self.pending_pcm.is_empty() {
123                    let mut frame = acquire_buffer(FRAME_SIZE_SAMPLES);
124                    frame.extend(self.pending_pcm.drain(..));
125                    frame.resize(FRAME_SIZE_SAMPLES, 0);
126                    self.process_frame(&mut frame);
127                    Ok(Some(frame))
128                } else {
129                    Err(AudioError::DecoderFinished)
130                }
131            } else {
132                Ok(None)
133            }
134        }
135        pub fn process_frame(&mut self, frame: &mut [i16]) {
136            self.tape.process(frame);
137            self.volume.process(frame);
138            self.fade.process(frame);
139            self.crossfade.fill_buffer();
140            if self.crossfade.is_active() {
141                self.crossfade.process(frame);
142            }
143        }
144        pub fn take_opus(&mut self) -> Option<Vec<u8>> {
145            self.latest_opus.take()
146        }
147    }
148}
149pub use controller::FlowController;
lavende_core/audio/flow.rs

lavende_core/audio/
flow.rs
