music_player/src/main.rs

use std::{
	io::Cursor,
	sync::{
		Arc,
		Condvar,
		Mutex,
	},
	thread,
};

use anyhow::{
	anyhow,
	//bail,
	Result,
};

use byteorder::{
	LittleEndian,
	ReadBytesExt,
};

use cpal::traits::{
	DeviceTrait,
	HostTrait,
	StreamTrait,
};

const SAMPLE_RATE: u32 = 48000;

#[derive (Default)]
struct PcmBuffers {
	buffers: Vec <Vec <f32>>,
	
	// Always points into the first buffer, if any
	consumer_cursor: usize,
}

impl PcmBuffers {
	fn samples_available (&self) -> usize {
		self.buffers.iter ().map (|b| b.len ()).sum::<usize> () - self.consumer_cursor
	}
	
	#[warn(unused_must_use)]
	fn consume_exact (&mut self, data: &mut [f32]) -> bool {
		if data.len () > self.samples_available () {
			return false;
		}
		
		for x in data {
			if self.consumer_cursor >= self.buffers [0].len () {
				self.buffers.remove (0);
				self.consumer_cursor = 0;
			}
			
			*x = self.buffers [0][self.consumer_cursor];
			self.consumer_cursor += 1;
		}
		
		true
	}
	
	fn produce (&mut self, new_buffer: Vec <f32>) {
		self.buffers.push (new_buffer);
	}
	
	fn produce_bytes (&mut self, new_buffer: &[u8]) {
		let mut b = vec! [0.0f32; new_buffer.len () / 4];
		let mut rdr = Cursor::new (new_buffer);
		
		rdr.read_f32_into::<LittleEndian> (&mut b).unwrap ();
		
		self.produce (b);
	}
}

fn main () -> Result <()> {
	let host = cpal::default_host ();
	let device = host.default_output_device ().ok_or_else (|| anyhow! ("can't open cpal device"))?;
	
	let mut supported_configs_range = device.supported_output_configs ()?
	.filter (|c| c.channels () == 2 && c.sample_format () == cpal::SampleFormat::F32);
	
	let config = supported_configs_range.next ()
	.ok_or_else (|| anyhow! ("can't get stereo f32 audio output"))?
	.with_sample_rate (cpal::SampleRate (SAMPLE_RATE))
	.config ();
	
	#[derive (Default)]
	struct DecoderState {
		pcm_buffers: PcmBuffers,
		quit: bool,
	}
	
	let pair = Arc::new ((Mutex::new (DecoderState::default ()), Condvar::new ()));
	let pair2 = Arc::clone (&pair);
	let pair3 = Arc::clone (&pair);
	
	let thread_decoder = thread::spawn (move|| {
		let (lock, cvar) = &*pair2;
		
		let mut input_ctx = ffmpeg_next::format::input (&"test.m4a")?;
		let stream = input_ctx
		.streams ()
		.best (ffmpeg_next::media::Type::Audio)
		.ok_or_else (|| anyhow! ("can't find good audio stream"))?;
		let best_stream_idx = stream.index ();
		
		let mut decoder = stream.codec ().decoder ().audio ()?;
		let mut resampler = decoder.resampler (
			ffmpeg_next::util::format::sample::Sample::F32 (
				ffmpeg_next::util::format::sample::Type::Packed,
			),
			ffmpeg_next::util::channel_layout::ChannelLayout::STEREO,
			48000,
		)?;
		
		let mut frame_src = ffmpeg_next::util::frame::Audio::empty ();
		let mut frame_resampled = ffmpeg_next::util::frame::Audio::empty ();
		let mut packets = input_ctx.packets ();
		
		'decoder_thread: loop {
			// eprintln! ("decode thread parking");
			
			let mut decoder_state = cvar.wait_while (lock.lock ().unwrap (), |decoder_state| {
				decoder_state.pcm_buffers.samples_available () >= 24_000 &&
				! decoder_state.quit
			}).unwrap ();
			
			if decoder_state.quit {
				break 'decoder_thread;
			}
			
			//dbg! (resampler.delay ());
			
			let pcm_buffers = &mut decoder_state.pcm_buffers;
			
			'fill_buffer: while pcm_buffers.samples_available () < 48_000 {
				//eprintln! ("Decoder is trying to work...");
				
				match resampler.delay () {
					Some (x) if x.milliseconds > 500 => {
						eprintln! ("flushing resampler");
						if let Some (_) = resampler.flush (&mut frame_resampled)? {
							pcm_buffers.produce_bytes (frame_resampled.data (0));
							continue 'fill_buffer;
						}
					},
					_ => {},
				}
				
				if decoder.receive_frame (&mut frame_src).is_ok () {
					//eprintln! ("decoder.receive_frame");
					resampler.run (&frame_src, &mut frame_resampled)?;
					pcm_buffers.produce_bytes (frame_resampled.data (0));
					continue 'fill_buffer;
				}
				
				//eprintln! ("Decoder is dry, fetching a new packet...");
				
				'get_packet: while let Some ((stream, packet)) = packets.next () {
					if stream.index () == best_stream_idx {
						//eprintln! ("decoder.send_packet");
						decoder.send_packet (&packet)?;
						continue 'fill_buffer;
					}
				}
				
				//eprintln! ("Decoder ran out of work");
				
				if resampler.delay ().is_some () {
					eprintln! ("flushing resampler");
					if let Some (_) = resampler.flush (&mut frame_resampled)? {
						//eprintln! ("resampler.flush");
						pcm_buffers.produce_bytes (frame_resampled.data (0));
						continue 'fill_buffer;
					}
				}
				
				break 'fill_buffer;
			}
		}
		
		Ok::<_, anyhow::Error> (())
	});
	
	{
		let stream = device.build_output_stream (
			&config,
			move |data: &mut [f32], _: &cpal::OutputCallbackInfo| {
				let (lock, cvar) = &*pair;
				
				let mut decoder_state = match lock.lock () {
					Ok (x) => x,
					Err (_) => return,
				};
				
				let pcm_buffers = &mut decoder_state.pcm_buffers;
				
				if ! pcm_buffers.consume_exact (data) {
					eprintln! ("PCM buffer underflow");
				}
				
				cvar.notify_one ();
			},
			move |err| {
				// react to errors here.
			},
		);
		
		std::thread::sleep (std::time::Duration::from_millis (180_000));
	}
	
	eprintln! ("Joining decoder thread");
	
	{
		let mut decoder_state = pair3.0.lock ().unwrap ();
		decoder_state.quit = true;
		pair3.1.notify_one ();
	}
	
	thread_decoder.join ().unwrap ()?;
	
	eprintln! ("Joined decoder thread");
	
	Ok (())
}

#[cfg (test)]
mod test {
	use super::*;
	
	#[test]
	fn decode () -> Result <()> {
		ffmpeg_next::init ()?;
		
		let mut input_ctx = ffmpeg_next::format::input (&"test.opus")?;
		let stream = input_ctx
		.streams ()
		.best (ffmpeg_next::media::Type::Audio)
		.ok_or_else (|| anyhow! ("can't find good audio stream"))?;
		let best_stream_idx = stream.index ();
		
		let mut decoder = stream.codec ().decoder ().audio ()?;
		
		let mut packet_count = 0;
		let mut frame_count = 0;
		let mut frame = ffmpeg_next::util::frame::Audio::empty ();
		
		let mut packets = input_ctx.packets ();
		
		loop {
			let mut did_anything = false;
			
			while decoder.receive_frame (&mut frame).is_ok () {
				frame_count += 1;
				did_anything = true;
			}
			
			match packets.next () {
				None => {},
				Some ((stream, packet)) => {
					did_anything = true;
					if stream.index () == best_stream_idx {
						packet_count += 1;
						decoder.send_packet (&packet)?;
					}
				},
			}
			
			if ! did_anything {
				break;
			}
		}
		
		assert_eq! (packet_count, 5496);
		assert_eq! (frame_count, 5496);
		
		Ok (())
	}
}