music_player/src/main.rs

use std::{
	fs::File,
	io::Write,
	sync::{
		Arc,
		Condvar,
		Mutex,
	},
	thread::{
		self,
		sleep,
	},
	time::{Duration, Instant},
};

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

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

use fltk::{
	app,
	button::Button,
	enums::CallbackTrigger,
	frame::Frame,
	group::Flex,
	prelude::*,
	window::Window,
};

mod decoder;

#[derive (Clone, Copy)]
enum Message {
	Play,
	Pause,
}

fn main () -> Result <()> {
	let args: Vec <_> = std::env::args ().collect ();
	
	match args.get (1).map (|s| &s[..]) {
		None => bail! ("First argument must be a subcommand like `play`"),
		Some ("debug-dump") => cmd_debug_dump (&args [1..]),
		Some ("debug-pipe") => cmd_debug_pipe (&args [1..]),
		Some ("gui") => cmd_gui (&args [1..]),
		Some ("play") => cmd_play (&args [1..]),
		Some (_) => bail! ("Unrecognized subcommand"),
	}
}

fn cmd_debug_dump (args: &[String]) -> Result <()> {
	tracing_subscriber::fmt::init ();
	
	let filename = args.get (1)
	.map (|s| s.to_string ())
	.unwrap_or_else (|| "test-short.m4a".to_string ());
	
	let mut decoder = decoder::Decoder::new (&filename)?;
	
	tracing::debug! ("Decoding...");
	
	let mut f = File::create ("pcm-dump.data")?;
	
	while let Some (frame) = decoder.next ()? {
		f.write_all (frame.data ())?;
	}
	
	Ok (())
}

fn cmd_debug_pipe (args: &[String]) -> Result <()> {
	tracing_subscriber::fmt::init ();
	
	let filename = args.get (1)
	.map (|s| s.to_string ())
	.unwrap_or_else (|| "test-short.m4a".to_string ());
	
	let mut decoder = decoder::Decoder::new (&filename)?;
	let mut sample_count = 0;
	
	while let Some (frame) = decoder.next ()? {
		sample_count += frame.samples ();
		// dbg! (frame, sample_count);
	}
	
	dbg! (sample_count);
	
	Ok (())
}

#[derive (Default)]
pub struct SharedState {
	pub pcm_buffers: decoder::PcmBuffers,
	pub quit: bool,
}

fn cmd_gui (_args: &[String]) -> Result <()> {
	let (fltk_tx, fltk_rx) = app::channel::<Message> ();
	
	let app = app::App::default ();
	let window_title = "Music Player".to_string ();
	let mut wind = Window::new (100, 100, 600, 600, None)
	.with_label (&window_title);
	wind.make_resizable (true);
	
	let mut row = Flex::default ().row ().size_of_parent ();
	
	let mut but_play = Button::default ().with_label ("▶️");
	but_play.set_trigger (CallbackTrigger::Release);
	but_play.emit (fltk_tx, Message::Play);
	row.set_size (&mut but_play, 30);
	
	let mut but_pause = Button::default ().with_label ("■");
	but_pause.set_trigger (CallbackTrigger::Release);
	but_pause.emit (fltk_tx, Message::Pause);
	row.set_size (&mut but_pause, 30);
	
	row.end ();
	
	wind.end ();
	wind.show ();
	
	while app.wait () {
		match fltk_rx.recv () {
			Some (Message::Play) => {
				tracing::info! ("play");
			},
			Some (Message::Pause) => {
				tracing::info! ("pause");
			},
			None => (),
		}
	}
	
	Ok (())
}
	
fn cmd_play (args: &[String]) -> Result <()> {
	tracing_subscriber::fmt::init ();
	
	let filenames: Vec <_> = args.iter ()
	.skip (1)
	.map (|s| s.to_string ())
	.collect ();
	
	let shared_state = Arc::new ((Mutex::new (SharedState::default ()), Condvar::new ()));
	let pcm_quit = Arc::new ((Mutex::new (false), Condvar::new ()));
	
	let thread_decoder = DecoderThread::new (Arc::clone (&shared_state), filenames);
	let audio_output = AudioOutput::new (Arc::clone (&pcm_quit), Arc::clone (&shared_state))?;
	
	tracing::debug! ("Joining decoder thread...");
	
	thread_decoder.join ()?;
	
	tracing::debug! ("Joining PCM thread...");
	
	let (lock, cvar) = &*pcm_quit;
	let _ = cvar.wait (lock.lock ().unwrap ()).unwrap ();
	
	drop (audio_output);
	
	sleep (Duration::from_secs (1));
	
	tracing::info! ("Exiting cleanly.");
	Ok (())
}

struct DecoderThread {
	join_handle: thread::JoinHandle <Result <()>>,
}

impl DecoderThread {
	pub fn new (
		shared_state: Arc <(Mutex <SharedState>, Condvar)>,
		filenames: Vec <String>,
	) -> Self
	{
		let join_handle = thread::spawn (move|| {
			let (lock, cvar) = &*shared_state;
			
			'many_files: for filename in &filenames {
				let mut decoder = decoder::Decoder::new (&filename)?;
				
				'one_file: loop {
					let frame = match decoder.next ()? {
						Some (x) => x,
						None => {
							tracing::debug! ("Decoder thread finished a file");
							break 'one_file;
						},
					};
					
					let mut decoder_state = cvar.wait_while (lock.lock ().unwrap (), |decoder_state| {
						decoder_state.pcm_buffers.samples_available () >= 12_000 &&
						! decoder_state.quit
					}).unwrap ();
					
					if decoder_state.quit {
						break 'many_files;
					}
					
					decoder_state.pcm_buffers.produce_bytes (frame.data ().into ());
				}
			}
			
			Ok::<_, anyhow::Error> (())
		});
		
		Self {
			join_handle,
		}
	}
	
	pub fn join (self) -> Result <()> {
		self.join_handle.join ().unwrap ()
	}
}

struct AudioOutput {
	host: cpal::Host,
	device: cpal::Device,
	stream: cpal::Stream,
}

impl AudioOutput {
	pub fn new (
		pcm_quit: Arc <(Mutex <bool>, Condvar)>,
		shared_state: Arc <(Mutex <SharedState>, Condvar)>,
	) -> Result <Self> 
	{
		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 (decoder::SAMPLE_RATE))
		.config ();
		
		let stream = device.build_output_stream (
			&config,
			move |data: &mut [f32], _: &cpal::OutputCallbackInfo| {
				let (lock, cvar) = &*shared_state;
				
				let time_start = Instant::now ();
				let mut decoder_state = match lock.lock () {
					Ok (x) => x,
					Err (_) => return,
				};
				let time_stop = Instant::now ();
				let dur = time_stop - time_start;
				if dur.as_micros () > 100 {
					tracing::warn! ("PCM thread waited {} us for a lock", dur.as_micros ());
				}
				
				let pcm_buffers = &mut decoder_state.pcm_buffers;
				
				if ! pcm_buffers.consume_exact (data) {
					tracing::warn! ("PCM buffer underflow");
					
					for x in data {
						*x = 0.0;
					}
					
					let (lock, cvar) = &*pcm_quit;
					
					let mut pcm_quit = lock.lock ().unwrap ();
					*pcm_quit = true;
					cvar.notify_one ();
				}
				
				cvar.notify_one ();
			},
			move |_err| {
				// react to errors here.
			},
		)?;
		
		Ok (Self {
			host,
			device,
			stream,
		})
	}
	
	pub fn play (&mut self) -> Result <()> {
		Ok (self.stream.play ()?)
	}
	
	pub fn pause (&mut self) -> Result <()> {
		Ok (self.stream.pause ()?)
	}
}

#[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;
			}
		}
		
		Ok (())
	}
}