➕ 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.main
_
parent
99b3bfe231
commit
84dc1a983f
|
|
@ -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)))
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
178
src/main.rs
178
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);
|
||||
|
|
|
|||
Loading…
Reference in New Issue