From 84dc1a983fd7f25f3c0a6eaeca329e314f9b1e5a Mon Sep 17 00:00:00 2001
From: _ <>
Date: Sun, 14 Nov 2021 19:31:54 +0000
Subject: [PATCH] :heavy_plus_sign: HTTP playback works, for Ogg/Vorbis only
 for now

This is really cool. Since I own the whole I/O stack, I could also make it run
over QUIC or some compressed format or anything I like.

And I can probably sniff the first packet by hand to suport both Vorbis and
Opus. That would cover most of my music library.
---
 src/decoder.rs |  56 +++++++++++-----
 src/main.rs    | 178 +++++++++++++++++++++++++------------------------
 2 files changed, 131 insertions(+), 103 deletions(-)

diff --git a/src/decoder.rs b/src/decoder.rs
index c3d31dc..9cdc6fe 100644
--- a/src/decoder.rs
+++ b/src/decoder.rs
@@ -90,8 +90,8 @@ impl PcmBuffers {
 	}
 }
 
-pub struct Decoder {
-	demuxer: FfmpegDemuxer,
+pub struct Decoder <T> {
+	demuxer: T,
 	
 	decoder: DecodeContext,
 	resampler: ResamplingContext,
@@ -99,10 +99,8 @@ pub struct Decoder {
 	dummy_frame: Option <FfAudioFrame>,
 }
 
-impl Decoder {
-	pub fn new (filename: &str) -> Result <Self> {
-		let (demuxer, codec) = FfmpegDemuxer::new (filename)?;
-		
+impl <T: Demuxer> Decoder <T> {
+	pub fn new (demuxer: T, codec: ffmpeg_next::codec::Context) -> Result <Self> {
 		let decoder = codec.decoder ().audio ()?;
 		let resampler = decoder.resampler (
 			Sample::F32 (sample::Type::Packed),
@@ -228,11 +226,11 @@ impl Decoder {
 	}
 }
 
-trait Demuxer {
-	
+pub trait Demuxer {
+	fn pump (&mut self) -> Result <Option <ffmpeg_next::Packet>>;
 }
 
-struct FfmpegDemuxer {
+pub struct FfmpegDemuxer {
 	input_ctx: DemuxContext,
 	best_stream_idx: usize,
 }
@@ -255,11 +253,10 @@ impl FfmpegDemuxer {
 	}
 }
 
-impl FfmpegDemuxer {
+impl Demuxer for FfmpegDemuxer {
 	fn pump (&mut self) -> Result <Option <ffmpeg_next::Packet>> {
 		while let Some ((stream, packet)) = self.input_ctx.packets ().next () {
 			if stream.index () == self.best_stream_idx {
-				// tracing::trace! ("demuxed packet");
 				return Ok (Some (packet));
 			}
 		}
@@ -270,19 +267,48 @@ impl FfmpegDemuxer {
 
 use crate::net_reader::NetReader;
 
-struct HttpOggDemuxer {
+pub struct HttpOggDemuxer {
 	ogg_rdr: ogg::reading::PacketReader <std::io::BufReader <NetReader>>,
 }
 
 impl HttpOggDemuxer {
-	pub fn new (url: String) -> Result <Self> {
+	pub fn new (url: String) -> Result <(Self, ffmpeg_next::codec::Context)> {
 		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);
 		
-		Ok (Self {
+		let codec = {
+			// 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)))
 	}
 }
 
diff --git a/src/main.rs b/src/main.rs
index 20b1ba7..9d2fa27 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -1,6 +1,5 @@
 use std::{
 	fs::File,
-	io::Write,
 	sync::{
 		Arc,
 		Condvar,
@@ -45,9 +44,7 @@ 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"),
@@ -55,71 +52,67 @@ fn main () -> Result <()> {
 }
 
 fn cmd_debug_net (args: &[String]) -> Result <()> {
-	use std::{
-		io::{
-			self,
-			Read,
-			Seek,
-			SeekFrom,
-		},
-	};
-	
 	tracing_subscriber::fmt::init ();
 	
-	let url = args.get (1).ok_or_else (|| anyhow! ("Need a URL"))?
-	.to_string ();
+	let mut urls: Vec <_> = args.iter ().skip (1).map (|s| s.to_string ()).collect ();
+	urls.reverse ();
 	
-	let net_rdr = net_reader::NetReader::new (url)?;
-	let buf_rdr = std::io::BufReader::new (net_rdr);
-	let mut ogg_rdr = ogg::reading::PacketReader::new (buf_rdr);
+	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, 300, 300, None)
+	.with_label (&window_title);
+	wind.make_resizable (true);
+	wind.end ();
+	wind.show ();
 	
-	let mut packet_count = 0;
+	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 ()));
 	
-	while let Ok (Some (packet)) = ogg_rdr.read_packet () {
-		packet_count += 1;
+	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");
 	}
 	
-	tracing::info! ("Read {} packets", packet_count);
-	
-	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 ())?;
+	while app.wait () {
+		tracing::trace! ("Main thread is waiting on messages...");
+		let msg = match fltk_rx.recv () {
+			None => continue,
+			Some (x) => x,
+		};
+		
+		match msg {
+			Message::DecoderStatus (x) => {
+				if ! x {
+					tracing::debug! ("decoder ended");
+					if let Some (url) = urls.pop () {
+						thread_decoder.tx.send (MessageToDecoder::SetUrl (url))?;
+					}
+					else {
+						break;
+					}
+				}
+			},
+			_ => {},
+		}
 	}
 	
-	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);
+	audio_output.pause ()?;
+	thread_decoder.tx.send (MessageToDecoder::Quit)?;
+	thread_decoder.join ()?;
 	
 	Ok (())
 }
@@ -164,7 +157,6 @@ impl Gui {
 
 #[derive (Clone, Copy, Debug)]
 enum Message {
-	AudioStreamStatus (bool),
 	DecoderStatus (bool),
 	RawPlay,
 	RawPause,
@@ -258,17 +250,15 @@ 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),
-		fltk_tx.clone (),
+		Some (fltk_tx.clone ()),
 	);
 	let mut audio_output = AudioOutput::new (
 		Arc::clone (&shared_state),
 		thread_decoder.tx.clone (),
-		fltk_tx.clone (),
 	)?;
 	
 	// Doesn't work.
@@ -292,9 +282,6 @@ fn cmd_gui (args: &[String]) -> Result <()> {
 		};
 		
 		match msg {
-			Message::AudioStreamStatus (x) => {
-				
-			},
 			Message::DecoderStatus (x) => {
 				if ! x {
 					tracing::info! ("Decoder ran dry");
@@ -312,11 +299,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 (Some (format! ("/home/user/music/{}", filename))))?;
+				thread_decoder.tx.send (MessageToDecoder::SetFile (format! ("/home/user/music/{}", filename)))?;
 				current_file = Some (filename);
 			},
 			Message::RawSkip => {
-				thread_decoder.tx.send (MessageToDecoder::SetFile (None))?;
+				thread_decoder.tx.send (MessageToDecoder::Clear)?;
 			},
 		}
 	}
@@ -391,14 +378,21 @@ struct DecoderThread {
 
 enum MessageToDecoder {
 	NeedPcm,
-	SetFile (Option <String>),
+	SetFile (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: fltk::app::Sender <Message>,
+		main_tx: Option <fltk::app::Sender <Message>>,
 	) -> Self
 	{
 		let (tx, rx) = mpsc::sync_channel (4);
@@ -406,9 +400,17 @@ impl DecoderThread {
 		let self_tx = tx.clone ();
 		
 		let join_handle = thread::spawn (move|| {
-			let mut decoder: Option <decoder::Decoder> = None;
+			let mut decoder: Option <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;
@@ -416,11 +418,10 @@ impl DecoderThread {
 					
 					'pipeline: while bytes_produced < BUFFER_SIZE * 4 * 2 {
 						// tracing::trace! ("Pumping multi-file decode pipeline");
-						let frame = if let Some (decoder) = decoder.as_mut () {
-							decoder.next ()?
-						}
-						else {
-							None
+						let frame = match &mut decoder {
+							None => None,
+							Some (Decoder::File (x)) => x.next ()?,
+							Some (Decoder::Http (x)) => x.next ()?,
 						};
 						
 						if let Some (frame) = frame {
@@ -433,7 +434,8 @@ impl DecoderThread {
 						}
 						else if decoder.is_some () {
 							decoder = None;
-							main_tx.send (Message::DecoderStatus (false));
+							tracing::trace! ("Decoder ended, telling main thread");
+							main_tx.as_ref ().map (|x| x.send (Message::DecoderStatus (false)));
 						}
 						
 						{
@@ -450,20 +452,22 @@ impl DecoderThread {
 							continue 'pipeline;
 						}
 					}
-					// tracing::trace! ("Left NeedPcm handler");
-					// main_tx.send (Message::DecoderStatus (status));
 				},
 				MessageToDecoder::SetFile (x) => {
-					if let Some (x) = x.as_ref () {
-						decoder = Some (decoder::Decoder::new (x)?);
-					}
-					else {
-						decoder = None;
-					}
-					// main_tx.send (Message::DecoderStatus (x.is_some ()));
+					let (demuxer, codec) = decoder::FfmpegDemuxer::new (&x)?;
+					decoder = Some (Decoder::File (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;
+				},
+				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;
 				},
@@ -499,7 +503,6 @@ 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 ();
@@ -553,7 +556,6 @@ 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);