1use std::fs::File;
2use std::io::{BufReader, Cursor, Read, Seek};
3use std::path::PathBuf;
4use std::sync::Arc;
5use std::time::Duration;
6
7use lewton::inside_ogg::OggStreamReader;
8
9use crate::channel::ChannelId;
10use crate::clip::Clip;
11use crate::decode::ahead_limit;
12use crate::interp::StreamResampler;
13use crate::mix::{Level, Mixer, PcmQueue, VoiceId};
14
15trait Source: Read + Seek + Send {}
16
17impl<T: Read + Seek + Send> Source for T {}
18
19type Reader = OggStreamReader<Box<dyn Source>>;
20
21#[derive(Clone, Debug)]
22pub enum VorbisSource {
23 Bytes(Arc<[u8]>),
24 File(PathBuf),
25}
26
27impl VorbisSource {
28 fn open(&self) -> Result<Reader, String> {
29 let source: Box<dyn Source> = match self {
30 VorbisSource::Bytes(bytes) => Box::new(Cursor::new(Arc::clone(bytes))),
31 VorbisSource::File(path) => {
32 let file = File::open(path).map_err(|e| format!("{}: {e}", path.display()))?;
33 Box::new(BufReader::new(file))
34 }
35 };
36 OggStreamReader::new(source).map_err(|e| format!("vorbis: {e}"))
37 }
38}
39
40fn unit(sample: i16) -> f32 {
41 f32::from(sample) / 32768.0
42}
43
44impl Clip {
45 pub fn from_vorbis(bytes: &[u8]) -> Result<Clip, String> {
46 let mut reader =
47 OggStreamReader::new(Cursor::new(bytes)).map_err(|e| format!("vorbis: {e}"))?;
48 let rate = reader.ident_hdr.audio_sample_rate;
49 let channels = u16::from(reader.ident_hdr.audio_channels);
50 if rate == 0 || channels == 0 {
51 return Err("vorbis: no audio".into());
52 }
53 let mut samples = Vec::new();
54 while let Some(packet) = reader
55 .read_dec_packet_itl()
56 .map_err(|e| format!("vorbis: {e}"))?
57 {
58 samples.extend(packet.into_iter().map(unit));
59 }
60 if let Some(end) = reader.get_last_absgp() {
61 let keep = usize::try_from(end).unwrap_or(usize::MAX);
62 samples.truncate(keep.saturating_mul(usize::from(channels)));
63 }
64 Ok(Clip {
65 rate,
66 channels,
67 samples,
68 })
69 }
70}
71
72struct Pump {
73 source: VorbisSource,
74 reader: Reader,
75 looping: Option<(u64, u64)>,
76 channels: usize,
77 ahead: usize,
78 queue: Arc<PcmQueue>,
79 resampler: StreamResampler,
80 held: Option<Vec<i16>>,
81 done: u64,
82 discard_to: u64,
83 emitted_this_pass: bool,
84}
85
86impl Pump {
87 fn run(mut self) {
88 loop {
89 if Arc::strong_count(&self.queue) <= 1 {
90 return;
91 }
92 if self.queue.len() >= self.ahead {
93 std::thread::sleep(Duration::from_millis(10));
94 continue;
95 }
96 match self.reader.read_dec_packet_itl() {
97 Ok(Some(packet)) => {
98 if let Some(previous) = self.held.replace(packet) {
99 self.take(&previous);
100 }
101 if self.looping.is_some_and(|(_, end)| self.done >= end) && !self.restart() {
102 return;
103 }
104 }
105 Ok(None) | Err(_) => {
106 if let Some(last) = self.held.take() {
107 let last = self.trimmed(last);
108 self.take(&last);
109 }
110 if self.looping.is_some() && self.restart() {
111 continue;
112 }
113 self.conclude();
114 return;
115 }
116 }
117 }
118 }
119
120 fn trimmed(&self, mut last: Vec<i16>) -> Vec<i16> {
121 let Some(end) = self.reader.get_last_absgp() else {
122 return last;
123 };
124 let frames = (last.len() / self.channels) as u64;
125 let excess = (self.done + frames).saturating_sub(end).min(frames);
126 last.truncate((frames - excess) as usize * self.channels);
127 last
128 }
129
130 fn take(&mut self, packet: &[i16]) {
131 let frames = (packet.len() / self.channels) as u64;
132 let from = self.discard_to.saturating_sub(self.done).min(frames);
133 let to = match self.looping {
134 Some((_, end)) => end.saturating_sub(self.done).min(frames),
135 None => frames,
136 };
137 if to > from {
138 let slice = &packet[from as usize * self.channels..to as usize * self.channels];
139 let floats: Vec<f32> = slice.iter().copied().map(unit).collect();
140 let mut out = Vec::new();
141 self.resampler.push(&floats, &mut out);
142 self.queue.push(&out);
143 self.emitted_this_pass = true;
144 }
145 self.done += frames;
146 }
147
148 fn conclude(&mut self) {
149 let mut out = Vec::new();
150 self.resampler.finish(&mut out);
151 self.queue.push(&out);
152 self.queue.end();
153 }
154
155 fn restart(&mut self) -> bool {
156 if !self.emitted_this_pass {
157 self.conclude();
158 return false;
159 }
160 let Ok(reader) = self.source.open() else {
161 self.conclude();
162 return false;
163 };
164 self.reader = reader;
165 self.held = None;
166 self.done = 0;
167 self.discard_to = self.looping.map_or(0, |(start, _)| start);
168 self.emitted_this_pass = false;
169 true
170 }
171}
172
173pub(crate) fn open_stream(
174 source: &VorbisSource,
175 rate: u32,
176 looping: Option<(u64, u64)>,
177) -> Result<(Arc<PcmQueue>, u16), String> {
178 let reader = source.open()?;
179 let src_rate = reader.ident_hdr.audio_sample_rate;
180 let channels = u16::from(reader.ident_hdr.audio_channels);
181 if rate == 0 || src_rate == 0 || channels == 0 {
182 return Err("vorbis: no audio".into());
183 }
184 let looping = looping.filter(|(start, end)| start < end);
185 let queue = PcmQueue::new();
186 let pump = Pump {
187 source: source.clone(),
188 reader,
189 looping,
190 channels: usize::from(channels),
191 ahead: ahead_limit(rate, channels),
192 queue: Arc::clone(&queue),
193 resampler: StreamResampler::new(src_rate, rate, usize::from(channels)),
194 held: None,
195 done: 0,
196 discard_to: 0,
197 emitted_this_pass: false,
198 };
199 std::thread::spawn(move || pump.run());
200 Ok((queue, channels))
201}
202
203impl Mixer {
204 pub fn stream_vorbis(
205 &mut self,
206 channel: ChannelId,
207 source: &VorbisSource,
208 looping: Option<(u64, u64)>,
209 level: Level,
210 ) -> Result<VoiceId, String> {
211 let (queue, channels) = open_stream(source, self.rate(), looping)?;
212 Ok(self.stream_on(channel, queue, channels, level))
213 }
214
215 pub fn play_music_vorbis(
216 &mut self,
217 source: &VorbisSource,
218 looping: Option<(u64, u64)>,
219 fade_seconds: f32,
220 level: Level,
221 ) -> Result<VoiceId, String> {
222 let (queue, channels) = open_stream(source, self.rate(), looping)?;
223 Ok(self.stream_music(queue, channels, fade_seconds, level))
224 }
225}
226
227#[cfg(test)]
228mod tests {
229 use super::*;
230 use std::time::Instant;
231
232 const OGG: &[u8] = include_bytes!("../tests/fixtures/sine.ogg");
233
234 fn bytes() -> VorbisSource {
235 VorbisSource::Bytes(Arc::from(OGG))
236 }
237
238 fn fill(queue: &PcmQueue, samples: usize) {
239 let until = Instant::now() + Duration::from_secs(20);
240 while queue.len() < samples && !queue.reader_ended() && Instant::now() < until {
241 std::thread::sleep(Duration::from_millis(2));
242 }
243 assert!(queue.len() >= samples || queue.reader_ended());
244 }
245
246 fn finish(queue: &PcmQueue) {
247 let until = Instant::now() + Duration::from_secs(20);
248 while !queue.reader_ended() && Instant::now() < until {
249 std::thread::sleep(Duration::from_millis(2));
250 }
251 assert!(queue.reader_ended());
252 }
253
254 fn drain(queue: &PcmQueue, frames: usize, channels: usize) -> Vec<f32> {
255 queue.pop_frames(frames, channels).samples
256 }
257
258 #[test]
259 fn a_stream_yields_the_same_samples_as_the_clip() {
260 let clip = Clip::from_vorbis(OGG).unwrap();
261 let (queue, channels) = open_stream(&bytes(), 44_100, None).unwrap();
262 assert_eq!(channels, 2);
263 finish(&queue);
264 assert_eq!(queue.len(), clip.samples.len());
265 let streamed = drain(&queue, clip.frames() as usize, 2);
266 assert_eq!(streamed, clip.samples);
267 }
268
269 #[test]
270 fn a_file_source_streams_too() {
271 let path = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/sine.ogg");
272 let (queue, _) = open_stream(&VorbisSource::File(path), 44_100, None).unwrap();
273 finish(&queue);
274 assert_eq!(queue.len(), 88_200);
275 }
276
277 #[test]
278 fn a_looping_stream_repeats_between_its_loop_points() {
279 let clip = Clip::from_vorbis(OGG).unwrap();
280 let (queue, _) = open_stream(&bytes(), 44_100, Some((10_000, 30_000))).unwrap();
281 fill(&queue, 2 * 70_000);
282 assert!(!queue.reader_ended());
283 let out = drain(&queue, 70_000, 2);
284 assert_eq!(out[..60_000], clip.samples[..60_000]);
285 assert_eq!(out[60_000..100_000], clip.samples[20_000..60_000]);
286 assert_eq!(out[100_000..140_000], clip.samples[20_000..60_000]);
287 }
288
289 #[test]
290 fn a_stream_resamples_to_the_output_rate() {
291 let clip = Clip::from_vorbis(OGG).unwrap();
292 let (queue, _) = open_stream(&bytes(), 48_000, None).unwrap();
293 finish(&queue);
294 let frames = queue.len() / 2;
295 assert!((47_998..=48_002).contains(&frames), "{frames}");
296 let streamed = drain(&queue, frames, 2);
297 let power =
298 |samples: &[f32]| samples.iter().map(|s| s * s).sum::<f32>() / samples.len() as f32;
299 let drift = 10.0 * (power(&streamed) / power(&clip.samples)).log10();
300 assert!(drift.abs() < 0.2, "{drift} dB");
301 }
302
303 #[test]
304 fn a_dropped_voice_stops_the_decoder_thread() {
305 let (queue, _) = open_stream(&bytes(), 44_100, Some((0, 10_000))).unwrap();
306 fill(&queue, 1000);
307 let weak = Arc::downgrade(&queue);
308 drop(queue);
309 let until = Instant::now() + Duration::from_secs(20);
310 while weak.strong_count() > 0 && Instant::now() < until {
311 std::thread::sleep(Duration::from_millis(5));
312 }
313 assert_eq!(weak.strong_count(), 0);
314 }
315}