1 use std::time::Duration;
2 use std::thread::JoinHandle;
3 use std::sync::{Arc, Mutex};
4 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
5 use std::sync::mpsc::{Receiver, SyncSender, TrySendError};
8 use sdl2::AudioSubsystem;
9 use sdl2::audio::{AudioDevice, AudioCallback, AudioSpecDesired};
11 use nihav_core::formats::*;
12 use nihav_core::codecs::*;
13 use nihav_core::soundcvt::*;
15 use super::{DecoderStuff, DecoderType, DecoderState, DecodingState, PktSendEvent};
17 static ADEC_STATE: DecoderState = DecoderState::new();
18 static AUDIO_VOLUME: AtomicUsize = AtomicUsize::new(100);
19 static CUR_QUEUE_FILL: AtomicUsize = AtomicUsize::new(0);
20 static SAMPLE_RATE: AtomicUsize = AtomicUsize::new(0);
21 static CHANNELS: AtomicUsize = AtomicUsize::new(0);
23 static CURRENT_TIME: AtomicUsize = AtomicUsize::new(0);
24 static CURRENT_TIME_SET: AtomicBool = AtomicBool::new(false);
26 const QUEUE_INITIAL_SIZE: usize = 16384;
27 const QUEUE_REFILL_LIMIT: usize = 262144;
39 fn new(srate: usize, chans: usize) -> Self {
40 SAMPLE_RATE.store(srate, Ordering::Relaxed);
41 CHANNELS.store(chans, Ordering::Relaxed);
42 CUR_QUEUE_FILL.store(0, Ordering::Relaxed);
44 queue: Vec::with_capacity(QUEUE_INITIAL_SIZE),
52 if self.start == 0 { return; }
53 let fill = self.fill();
55 if fill < self.start {
56 let (dst, src) = self.queue.split_at_mut(self.start);
57 dst[..fill].copy_from_slice(&src[..fill]);
60 self.queue[i] = self.queue[self.start + i];
67 fn set_time(&mut self) {
68 let fill = self.fill();
69 let samp_time = self.spos.saturating_sub(fill / self.chans);
70 CURRENT_TIME.store(samp_time * 1000 / self.srate, Ordering::Relaxed);
71 CURRENT_TIME_SET.store(true, Ordering::Relaxed);
72 CUR_QUEUE_FILL.store(self.fill(), Ordering::Relaxed);
74 fn add(&mut self, src: &[i16], samplepos: usize) {
75 if self.end + src.len() > self.queue.len() {
78 if self.end + src.len() > self.queue.len() {
79 self.queue.resize(self.end + src.len(), 0);
81 self.queue[self.end..][..src.len()].copy_from_slice(src);
82 self.end += src.len();
83 self.spos = samplepos;
86 fn add_bytes(&mut self, src: &[u8], samplepos: usize) {
87 let srclen = src.len() / 2;
88 if self.end + srclen > self.queue.len() {
91 if self.end + srclen > self.queue.len() {
92 self.queue.resize(self.end + srclen, 0);
94 for (dst, src) in self.queue[self.end..][..srclen].iter_mut().zip(src.chunks_exact(2)) {
95 *dst = (u16::from(src[0]) + u16::from(src[1]) * 256) as i16;
98 self.spos = samplepos;
101 fn drain(&mut self, size: usize) {
102 let fill = self.fill();
110 fn fill(&self) -> usize { self.end - self.start }
111 fn flush(&mut self) {
117 pub struct AudioOutput {
118 queue: Arc<Mutex<AudioQueue>>,
121 impl AudioCallback for AudioOutput {
124 fn callback(&mut self, out: &mut [Self::Channel]) {
125 let mut queue = self.queue.lock().expect("audio queue should be accessible");
126 let dstlen = out.len();
127 let copylen = queue.fill().min(dstlen);
128 let volume = AUDIO_VOLUME.load(Ordering::Relaxed) as i32;
130 out[..copylen].copy_from_slice(&queue.queue[queue.start..][..copylen]);
132 for (dst, &src) in out[..copylen].iter_mut().zip(queue.queue[queue.start..].iter()) {
133 *dst = (i32::from(src) * volume / 100).max(-32768).min(32767) as i16;
136 queue.drain(copylen);
137 for el in out[copylen..].iter_mut() { *el = 0; }
141 fn dummy_audio_thread(aprecv: Receiver<PktSendEvent>) -> JoinHandle<()> {
142 std::thread::Builder::new().name("acontrol-dummy".to_string()).spawn(move ||{
144 match aprecv.recv() {
145 Ok(PktSendEvent::End) => break,
146 Ok(PktSendEvent::ImmediateEnd) => break,
156 type AudioPlaybackType = Option<AudioDevice<AudioOutput>>;
158 fn start_audio_decoding(asystem: &AudioSubsystem, ainfo: NAAudioInfo, sbr_hack: bool, mut audio_dec: DecoderStuff, aprecv: Receiver<PktSendEvent>) -> (AudioPlaybackType, JoinHandle<()>) {
159 let ch = ainfo.channels.max(2);
160 let sample_rate = if !sbr_hack { ainfo.sample_rate } else { ainfo.sample_rate * 2 };
161 let desired_spec = AudioSpecDesired {
162 freq: Some(sample_rate as i32),
166 let dst_info = NAAudioInfo {
169 format: SND_S16_FORMAT,
172 let queue = Arc::new(Mutex::new(AudioQueue::new(sample_rate as usize, ch as usize)));
173 let qclone = queue.clone();
174 let ret = asystem.open_playback(None, &desired_spec, |_spec| {
180 return (None, dummy_audio_thread(aprecv))
182 let adevice = ret.unwrap();
183 (Some(adevice), std::thread::Builder::new().name("acontrol".to_string()).spawn(move ||{
184 let adec = if let DecoderType::Audio(ref mut dec) = audio_dec.dec { dec } else { panic!("not an audio decoder!"); };
185 let mut samplepos = 0usize;
186 let dst_chmap = if dst_info.channels == 2 {
187 NAChannelMap::from_str("L,R").expect("should be able to create stereo channel map")
189 NAChannelMap::from_str("C").expect("should be able to create single-channel map")
191 ADEC_STATE.set_state(DecodingState::Normal);
193 match aprecv.recv() {
194 Ok(PktSendEvent::Packet(pkt)) => {
196 if CUR_QUEUE_FILL.load(Ordering::Relaxed)
197 < QUEUE_REFILL_LIMIT || ADEC_STATE.is_flushing() {
200 std::thread::sleep(Duration::from_millis(100));
202 if !ADEC_STATE.is_flushing() {
203 if let Ok(frm) = adec.decode(&mut audio_dec.dsupp, &pkt) {
204 let buf = frm.get_buffer();
205 if let Some(pts) = frm.get_pts() {
206 samplepos = NATimeInfo::ts_to_time(pts, u64::from(dst_info.sample_rate), frm.ts.tb_num, frm.ts.tb_den) as usize;
211 samplepos += buf.get_audio_length();
212 if let Ok(out_buf) = convert_audio_frame(&buf, &dst_info, &dst_chmap) {
214 NABufferType::AudioI16(abuf) => {
215 let mut qdata = queue.lock().expect("audio queue should be accessible");
216 qdata.add(abuf.get_data(), samplepos);
219 NABufferType::AudioPacked(abuf) => {
220 let mut qdata = queue.lock().expect("audio queue should be accessible");
221 qdata.add_bytes(abuf.get_data(), samplepos);
230 Ok(PktSendEvent::GetFrames) => {},
231 Ok(PktSendEvent::Flush) => {
233 let mut qdata = queue.lock().expect("audio queue should be accessible");
235 ADEC_STATE.set_state(DecodingState::Waiting);
237 Ok(PktSendEvent::End) => break,
238 Ok(PktSendEvent::ImmediateEnd) => {
239 let mut qdata = queue.lock().expect("audio queue should be accessible");
243 Ok(PktSendEvent::HurryUp) => {},
250 let qdata = queue.lock().expect("audio queue should be accessible");
251 if qdata.fill() == 0 || ADEC_STATE.is_flushing() {
255 ADEC_STATE.set_state(DecodingState::End);
259 pub struct AudioControl {
260 aqueue: Vec<PktSendEvent>,
261 apsend: SyncSender<PktSendEvent>,
262 adevice: AudioPlaybackType,
263 athread: JoinHandle<()>,
267 pub fn new(audio_dec: Option<DecoderStuff>, ainfo: Option<NAAudioInfo>, sbr_hack: bool, asystem: &AudioSubsystem) -> Self {
268 let (apsend, aprecv) = std::sync::mpsc::sync_channel::<PktSendEvent>(20);
269 let (adevice, athread) = if let Some(audio_dec) = audio_dec {
270 start_audio_decoding(asystem, ainfo.expect("audio info should be present"), sbr_hack, audio_dec, aprecv)
272 (None, dummy_audio_thread(aprecv))
274 ADEC_STATE.set_state(DecodingState::Normal);
283 pub fn has_audio(&self) -> bool { self.adevice.is_some() }
284 pub fn pause(&mut self) {
285 if let Some(ref device) = self.adevice {
289 pub fn resume(&mut self) {
290 if let Some(ref device) = self.adevice {
294 pub fn set_volume(&mut self, volume: usize) {
295 AUDIO_VOLUME.store(volume, Ordering::Relaxed);
297 pub fn get_volume(&self) -> usize {
298 AUDIO_VOLUME.load(Ordering::Relaxed)
300 pub fn is_audio_end(&self) -> bool {
301 matches!(ADEC_STATE.get_state(), DecodingState::End | DecodingState::Error)
303 pub fn get_fill(&self) -> usize { CUR_QUEUE_FILL.load(Ordering::Relaxed) }
304 pub fn get_time(&self) -> Option<u64> {
305 if CURRENT_TIME_SET.load(Ordering::Relaxed) {
306 Some(CURRENT_TIME.load(Ordering::Relaxed) as u64)
311 pub fn get_time_left(&self) -> u64 {
312 let srate = SAMPLE_RATE.load(Ordering::Relaxed);
313 let chans = CHANNELS.load(Ordering::Relaxed);
314 if srate != 0 && chans != 0{
315 let fill = self.get_fill();
316 (fill * 1000 / srate / chans) as u64
322 pub fn get_queue_size(&self) -> usize { self.aqueue.len() }
323 pub fn try_send_audio(&mut self, evt: PktSendEvent) -> bool {
324 if !self.aqueue.is_empty() {
325 self.aqueue.push(evt);
328 self.try_send_event(evt)
331 fn try_send_event(&mut self, evt: PktSendEvent) -> bool {
332 if let Err(TrySendError::Full(evt)) = self.apsend.try_send(evt) {
333 self.aqueue.insert(0, evt);
339 pub fn try_send_queued(&mut self) -> bool {
340 while !self.aqueue.is_empty() {
341 let pkt = self.aqueue.remove(0);
342 if !self.try_send_event(pkt) {
349 pub fn flush(&mut self) {
352 ADEC_STATE.set_state(DecodingState::Flush);
353 CURRENT_TIME_SET.store(false, Ordering::Release);
354 let _ = self.apsend.send(PktSendEvent::Flush);
356 pub fn finish(self) {
357 ADEC_STATE.set_state(DecodingState::Flush);
358 let _ = self.apsend.send(PktSendEvent::ImmediateEnd);
359 self.athread.join().unwrap();