src/decoder.rs

@@ -90,18 +90,25 @@ impl PcmBuffers {
 	}
 }
 
-pub struct Decoder <T> {
+pub struct Decoder {
-	demuxer: T,
+	input_ctx: DemuxContext,
-	
 	decoder: DecodeContext,
 	resampler: ResamplingContext,
 	
+	best_stream_idx: usize,
 	dummy_frame: Option <FfAudioFrame>,
 }
 
-impl <T: Demuxer> Decoder <T> {
+impl Decoder {
-	pub fn new (demuxer: T, codec: ffmpeg_next::codec::Context) -> Result <Self> {
+	pub fn new (filename: &str) -> Result <Self> {
-		let decoder = codec.decoder ().audio ()?;
+		let input_ctx = ffmpeg_next::format::input (&filename)?;
+		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 decoder = stream.codec ().decoder ().audio ()?;
 		let resampler = decoder.resampler (
 			Sample::F32 (sample::Type::Packed),
 			ChannelLayout::STEREO,
@@ -109,10 +116,11 @@ impl <T: Demuxer> Decoder <T> {
 		)?;
 		
 		Ok (Self {
-			demuxer,
+			input_ctx,
 			decoder,
 			resampler,
 			
+			best_stream_idx,
 			dummy_frame: None,
 		})
 	}
@@ -215,28 +223,30 @@ impl <T: Demuxer> Decoder <T> {
 	}
 	
 	pub fn pump_demuxer (&mut self) -> Result <bool> {
-		let packet = self.demuxer.pump ()?;
+		while let Some ((stream, packet)) = self.input_ctx.packets ().next () {
-		let packet = match packet {
+			if stream.index () == self.best_stream_idx {
-			Some (x) => x,
+				// tracing::trace! ("demuxed packet");
-			None => return Ok (false),
-		};
-		
 				self.decoder.send_packet (&packet)?;
-		Ok (true)
+				return Ok (true);
+			}
+		}
+		
+		Ok (false)
 	}
 }
 
-pub trait Demuxer {
+trait Demuxer {
-	fn pump (&mut self) -> Result <Option <ffmpeg_next::Packet>>;
+	
 }
 
-pub struct FfmpegDemuxer {
+struct FfmpegDemuxer {
 	input_ctx: DemuxContext,
 	best_stream_idx: usize,
+	decoder: DecodeContext,
 }
 
 impl FfmpegDemuxer {
-	pub fn new (filename: &str) -> Result <(Self, ffmpeg_next::codec::Context)> {
+	pub fn new (filename: &str) -> Result <Self> {
 		let input_ctx = ffmpeg_next::format::input (&filename)?;
 		let stream = input_ctx
 		.streams ()
@@ -244,71 +254,35 @@ impl FfmpegDemuxer {
 		.ok_or_else (|| anyhow! ("can't find good audio stream"))?;
 		let best_stream_idx = stream.index ();
 		
-		let codec = stream.codec ();
+		let decoder = stream.codec ().decoder ().audio ()?;
 		
-		Ok ((Self {
+		Ok (Self {
 			input_ctx,
 			best_stream_idx,
-		}, codec))
+			decoder,
+		})
 	}
 }
 
-impl Demuxer for FfmpegDemuxer {
+impl FfmpegDemuxer {
-	fn pump (&mut self) -> Result <Option <ffmpeg_next::Packet>> {
+	fn pump (&mut self) -> Result <
-		while let Some ((stream, packet)) = self.input_ctx.packets ().next () {
-			if stream.index () == self.best_stream_idx {
-				return Ok (Some (packet));
-			}
-		}
-		
-		Ok (None)
-	}
 }
 
 use crate::net_reader::NetReader;
 
-pub struct HttpOggDemuxer {
+struct HttpOggDemuxer {
 	ogg_rdr: ogg::reading::PacketReader <std::io::BufReader <NetReader>>,
 }
 
 impl HttpOggDemuxer {
-	pub fn new (url: String) -> Result <(Self, ffmpeg_next::codec::Context)> {
+	pub fn new (url: String) -> Result <Self> {
 		let net_rdr = NetReader::new (url)?;
 		let buf_rdr = std::io::BufReader::new (net_rdr);
 		let mut ogg_rdr = ogg::reading::PacketReader::new (buf_rdr);
 		
-		let codec = {
+		Ok (Self {
-			// Open a local file as a placeholder for figuring out the remote file's codec
-			let input_ctx = ffmpeg_next::format::input (&"test.ogg")?;
-			let stream = input_ctx
-			.streams ()
-			.best (ffmpeg_next::media::Type::Audio)
-			.ok_or_else (|| anyhow! ("can't find good audio stream"))?;
-			
-			stream.codec ()
-		};
-		
-		Ok ((Self {
 			ogg_rdr,
-		}, codec))
+		})
-	}
-}
-
-impl Demuxer for HttpOggDemuxer {
-	fn pump (&mut self) -> Result <Option <ffmpeg_next::Packet>> {
-		let packet = match self.ogg_rdr.read_packet () {
-			Err (e) => {
-				tracing::error! ("Ogg read error: {:?}", e);
-				return Ok (None);
-			},
-			Ok (x) => x,
-		};
-		let packet = match packet {
-			None => return Ok (None),
-			Some (x) => x,
-		};
-		
-		Ok (Some (ffmpeg_next::Packet::copy (&packet.data)))
 	}
 }
 

src/main.rs

@@ -1,5 +1,6 @@
 use std::{
 	fs::File,
+	io::Write,
 	sync::{
 		Arc,
 		Condvar,
@@ -44,7 +45,9 @@ fn main () -> Result <()> {
 	
 	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-net") => cmd_debug_net (&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"),
@@ -52,67 +55,71 @@ fn main () -> Result <()> {
 }
 
 fn cmd_debug_net (args: &[String]) -> Result <()> {
-	tracing_subscriber::fmt::init ();
+	use std::{
-	
+		io::{
-	let mut urls: Vec <_> = args.iter ().skip (1).map (|s| s.to_string ()).collect ();
+			self,
-	urls.reverse ();
+			Read,
-	
+			Seek,
-	let (fltk_tx, fltk_rx) = app::channel::<Message> ();
+			SeekFrom,
-	let app = app::App::default ();
+		},
-	let window_title = "Music Player".to_string ();
-	let mut wind = Window::new (100, 100, 300, 300, None)
-	.with_label (&window_title);
-	wind.make_resizable (true);
-	wind.end ();
-	wind.show ();
-	
-	let mut shared_state = SharedState::default ();
-	let silence = vec! [0; 12_000 * 4 * 2];
-	shared_state.pcm_buffers.produce_bytes (silence);
-	let shared_state = Arc::new ((Mutex::new (shared_state), Condvar::new ()));
-	
-	let thread_decoder = DecoderThread::new (
-		Arc::clone (&shared_state),
-		Some (fltk_tx.clone ()),
-	);
-	let mut audio_output = AudioOutput::new (
-		Arc::clone (&shared_state),
-		thread_decoder.tx.clone (),
-	)?;
-	
-	if let Some (url) = urls.pop () {
-		thread_decoder.tx.send (MessageToDecoder::SetUrl (url))?;
-	}
-	else {
-		bail! ("No URLs to play");
-	}
-	
-	while app.wait () {
-		tracing::trace! ("Main thread is waiting on messages...");
-		let msg = match fltk_rx.recv () {
-			None => continue,
-			Some (x) => x,
 	};
 	
-		match msg {
+	tracing_subscriber::fmt::init ();
-			Message::DecoderStatus (x) => {
+	
-				if ! x {
+	let url = args.get (1).ok_or_else (|| anyhow! ("Need a URL"))?
-					tracing::debug! ("decoder ended");
+	.to_string ();
-					if let Some (url) = urls.pop () {
+	
-						thread_decoder.tx.send (MessageToDecoder::SetUrl (url))?;
+	let net_rdr = net_reader::NetReader::new (url)?;
-					}
+	let buf_rdr = std::io::BufReader::new (net_rdr);
-					else {
+	let mut ogg_rdr = ogg::reading::PacketReader::new (buf_rdr);
-						break;
+	
-					}
+	let mut packet_count = 0;
-				}
+	
-			},
+	while let Ok (Some (packet)) = ogg_rdr.read_packet () {
-			_ => {},
+		packet_count += 1;
-		}
 	}
 	
-	audio_output.pause ()?;
+	tracing::info! ("Read {} packets", packet_count);
-	thread_decoder.tx.send (MessageToDecoder::Quit)?;
+	
-	thread_decoder.join ()?;
+	Ok (())
+}
+
+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 (())
 }
@@ -157,6 +164,7 @@ impl Gui {
 
 #[derive (Clone, Copy, Debug)]
 enum Message {
+	AudioStreamStatus (bool),
 	DecoderStatus (bool),
 	RawPlay,
 	RawPause,
@@ -250,15 +258,17 @@ fn cmd_gui (args: &[String]) -> Result <()> {
 	let mut shared_state = SharedState::default ();
 	let silence = vec! [0; 12_000 * 4 * 2];
 	shared_state.pcm_buffers.produce_bytes (silence);
+	
 	let shared_state = Arc::new ((Mutex::new (shared_state), Condvar::new ()));
 	
 	let thread_decoder = DecoderThread::new (
 		Arc::clone (&shared_state),
-		Some (fltk_tx.clone ()),
+		fltk_tx.clone (),
 	);
 	let mut audio_output = AudioOutput::new (
 		Arc::clone (&shared_state),
 		thread_decoder.tx.clone (),
+		fltk_tx.clone (),
 	)?;
 	
 	// Doesn't work.
@@ -282,6 +292,9 @@ fn cmd_gui (args: &[String]) -> Result <()> {
 		};
 		
 		match msg {
+			Message::AudioStreamStatus (x) => {
+				
+			},
 			Message::DecoderStatus (x) => {
 				if ! x {
 					tracing::info! ("Decoder ran dry");
@@ -299,11 +312,11 @@ fn cmd_gui (args: &[String]) -> Result <()> {
 				let idx = rand::thread_rng ().gen_range (0..playlist.len ());
 				let filename = &playlist [idx];
 				tracing::info! ("Set decoder to {}", filename);
-				thread_decoder.tx.send (MessageToDecoder::SetFile (format! ("/home/user/music/{}", filename)))?;
+				thread_decoder.tx.send (MessageToDecoder::SetFile (Some (format! ("/home/user/music/{}", filename))))?;
 				current_file = Some (filename);
 			},
 			Message::RawSkip => {
-				thread_decoder.tx.send (MessageToDecoder::Clear)?;
+				thread_decoder.tx.send (MessageToDecoder::SetFile (None))?;
 			},
 		}
 	}
@@ -378,21 +391,14 @@ struct DecoderThread {
 
 enum MessageToDecoder {
 	NeedPcm,
-	SetFile (String),
+	SetFile (Option <String>),
-	SetUrl (String),
-	Clear,
 	Quit,
 }
 
-enum Decoder {
-	File (decoder::Decoder <decoder::FfmpegDemuxer>),
-	Http (decoder::Decoder <decoder::HttpOggDemuxer>),
-}
-
 impl DecoderThread {
 	pub fn new (
 		shared_state: Arc <(Mutex <SharedState>, Condvar)>,
-		main_tx: Option <fltk::app::Sender <Message>>,
+		main_tx: fltk::app::Sender <Message>,
 	) -> Self
 	{
 		let (tx, rx) = mpsc::sync_channel (4);
@@ -400,17 +406,9 @@ impl DecoderThread {
 		let self_tx = tx.clone ();
 		
 		let join_handle = thread::spawn (move|| {
-			let mut decoder: Option <Decoder> = None;
+			let mut decoder: Option <decoder::Decoder> = None;
 			
 			'thread: while let Ok (msg) = rx.recv () { match msg {
-				MessageToDecoder::Clear => {
-					decoder = None;
-					let new_buffers = decoder::PcmBuffers::default ();
-					self_tx.send (MessageToDecoder::NeedPcm)?;
-					
-					let mut shared_state = shared_state.0.lock ().unwrap ();
-					shared_state.pcm_buffers = new_buffers;
-				},
 				MessageToDecoder::NeedPcm => {
 					// tracing::trace! ("DecoderThread recv'd NeedPcm");
 					let mut bytes_produced = 0;
@@ -418,10 +416,11 @@ impl DecoderThread {
 					
 					'pipeline: while bytes_produced < BUFFER_SIZE * 4 * 2 {
 						// tracing::trace! ("Pumping multi-file decode pipeline");
-						let frame = match &mut decoder {
+						let frame = if let Some (decoder) = decoder.as_mut () {
-							None => None,
+							decoder.next ()?
-							Some (Decoder::File (x)) => x.next ()?,
+						}
-							Some (Decoder::Http (x)) => x.next ()?,
+						else {
+							None
 						};
 						
 						if let Some (frame) = frame {
@@ -434,8 +433,7 @@ impl DecoderThread {
 						}
 						else if decoder.is_some () {
 							decoder = None;
-							tracing::trace! ("Decoder ended, telling main thread");
+							main_tx.send (Message::DecoderStatus (false));
-							main_tx.as_ref ().map (|x| x.send (Message::DecoderStatus (false)));
 						}
 						
 						{
@@ -452,22 +450,20 @@ impl DecoderThread {
 							continue 'pipeline;
 						}
 					}
+					// tracing::trace! ("Left NeedPcm handler");
+					// main_tx.send (Message::DecoderStatus (status));
 				},
 				MessageToDecoder::SetFile (x) => {
-					let (demuxer, codec) = decoder::FfmpegDemuxer::new (&x)?;
+					if let Some (x) = x.as_ref () {
-					decoder = Some (Decoder::File (decoder::Decoder::new (demuxer, codec)?));
+						decoder = Some (decoder::Decoder::new (x)?);
-					
+					}
+					else {
+						decoder = None;
+					}
+					// main_tx.send (Message::DecoderStatus (x.is_some ()));
 					let new_buffers = decoder::PcmBuffers::default ();
 					self_tx.send (MessageToDecoder::NeedPcm)?;
-					let mut shared_state = shared_state.0.lock ().unwrap ();
-					shared_state.pcm_buffers = new_buffers;
-				},
-				MessageToDecoder::SetUrl (x) => {
-					let (demuxer, codec) = decoder::HttpOggDemuxer::new (x)?;
-					decoder = Some (Decoder::Http (decoder::Decoder::new (demuxer, codec)?));
 					
-					let new_buffers = decoder::PcmBuffers::default ();
-					self_tx.send (MessageToDecoder::NeedPcm)?;
 					let mut shared_state = shared_state.0.lock ().unwrap ();
 					shared_state.pcm_buffers = new_buffers;
 				},
@@ -503,6 +499,7 @@ impl AudioOutput {
 	pub fn new (
 		shared_state: Arc <(Mutex <SharedState>, Condvar)>,
 		decoder_tx: mpsc::SyncSender <MessageToDecoder>,
+		main_tx: fltk::app::Sender <Message>,
 	) -> Result <Self> 
 	{
 		let host = cpal::default_host ();
@@ -556,6 +553,7 @@ impl AudioOutput {
 					decoder_tx.try_send (MessageToDecoder::NeedPcm).ok ();
 					// tracing::trace! ("AudioOutput sent NeedPcm");
 				}
+				// main_tx.send (Message::AudioStreamStatus (status));
 			},
 			move |err| {
 				tracing::error! ("{:?}", err);