Added Opus decoder

Cargo.toml

@@ -11,6 +11,8 @@ bincode = "1.3.3"
 crc32fast = "1.3.2"
 crossterm = "0.27.0"
 interprocess = "1.2.1"
+opus = "0.3.0"
 ratatui = "0.23.0"
 rodio = {version = "0.17.3", features = [ "symphonia-all" ], default-features = false }
 serde = "1.0.196"
+symphonia = "0.5.4"

src/server/decoder.rs

@@ -4,50 +4,72 @@
 // - Mod
 // - XM
 
-use std::io::{Read, Seek};
+use std::{
+    fs::File,
+    io::{BufReader, Read, Seek},
+};
+use symphonia::core::{
+    codecs::CODEC_TYPE_OPUS,
+    formats::{FormatOptions, FormatReader, Track},
+    io::{MediaSource, MediaSourceStream},
+    meta::MetadataOptions,
+    probe::Hint,
+};
 
 mod opus;
 
+use crate::server::decoder::opus::OpusDecoder;
+
 pub struct Decoder;
 
 impl Decoder {
-    pub fn new<R>(mut data: R) -> Result<Box<dyn DecoderImpl<R, Item = i16>>, ()>
+    pub fn rodio<R>(data: R) -> Result<Box<dyn DecoderImpl<Item = i16>>, ()>
     where
-        R: Read + Seek + Send + Sync + 'static,
+        R: MediaSource + Read + Seek + Send + 'static,
     {
-        let start_position = data.stream_position().unwrap();
-
-        match opus::OpusDecoder::new(&mut data) {
-            Ok(source) => return Ok(Box::new(source)),
-            Err(err) => {
-                eprintln!("Custom decode error: {err:?}");
-            }
-        };
-
-        data.seek(std::io::SeekFrom::Start(start_position)).unwrap();
-
         match rodio::Decoder::new(data) {
-            Ok(source) => return Ok(Box::new(source)),
+            Ok(source) => Ok(Box::new(source)),
-            Err(err) => {
+            _ => Err(()),
-                eprintln!("Rodio decode error: {err:?}");
         }
+    }
+
+    pub fn custom<R>(data: R) -> Result<Box<dyn DecoderImpl<Item = i16>>, ()>
+    where
+        R: MediaSource + Read + Seek + Send + 'static,
+    {
+        let reader = match demux(data) {
+            Ok(track) => track,
+            _ => return Err(()),
         };
 
+        let tracks: Vec<_> = reader
+            .tracks()
+            .iter()
+            .cloned()
+            .map(|track| (track.codec_params.codec, track.id))
+            .collect();
+
+        for (codec, track_id) in tracks {
+            match codec {
+                CODEC_TYPE_OPUS => match OpusDecoder::new(Demuxer::new(reader, track_id)) {
+                    Ok(source) => return Ok(Box::new(source)),
+                    _ => return Err(()),
+                },
+                _ => {}
+            }
+        }
+
         Err(())
     }
 }
 
-pub trait DecoderImpl<R>: rodio::Source + Send + Sync + 'static
+pub trait DecoderImpl: rodio::Source + Send + 'static
 where
-    R: Read + Seek,
     <Self as Iterator>::Item: rodio::Sample,
 {
 }
 
-impl<R> rodio::Source for Box<dyn DecoderImpl<R, Item = i16>>
+impl rodio::Source for Box<dyn DecoderImpl<Item = i16>> {
-where
-    R: Read + Seek,
-{
     #[inline]
     fn current_frame_len(&self) -> Option<usize> {
         self.as_ref().current_frame_len()
@@ -69,4 +91,78 @@ where
     }
 }
 
-impl<R> DecoderImpl<R> for rodio::Decoder<R> where R: Read + Seek + Send + Sync + 'static {}
+impl<R> DecoderImpl for rodio::Decoder<R> where R: Read + Seek + Send + 'static {}
+
+pub struct SourceWrapper {
+    pub reader: BufReader<File>,
+    is_seekable: bool,
+    byte_len: Option<u64>,
+}
+
+impl SourceWrapper {
+    pub fn from_file(file: File) -> Self {
+        let is_seekable = file.is_seekable();
+        let byte_len = file.byte_len();
+        Self {
+            reader: BufReader::new(file),
+            is_seekable,
+            byte_len,
+        }
+    }
+}
+
+impl Read for SourceWrapper {
+    fn read(&mut self, buf: &mut [u8]) -> Result<usize, std::io::Error> {
+        self.reader.read(buf)
+    }
+}
+
+impl Seek for SourceWrapper {
+    fn seek(&mut self, pos: std::io::SeekFrom) -> Result<u64, std::io::Error> {
+        self.reader.seek(pos)
+    }
+}
+
+impl MediaSource for SourceWrapper {
+    fn is_seekable(&self) -> bool {
+        self.is_seekable
+    }
+
+    fn byte_len(&self) -> Option<u64> {
+        self.byte_len
+    }
+}
+
+pub struct Demuxer {
+    reader: Box<dyn FormatReader>,
+    track_id: u32,
+}
+
+impl Demuxer {
+    pub fn new(reader: Box<dyn FormatReader>, track_id: u32) -> Self {
+        Self { reader, track_id }
+    }
+
+    pub fn get_track(&self) -> Option<&Track> {
+        self.reader
+            .tracks()
+            .iter()
+            .find(|track| track.id == self.track_id)
+    }
+}
+
+fn demux<R>(data: R) -> Result<Box<dyn FormatReader>, ()>
+where
+    R: MediaSource + 'static,
+{
+    let stream = MediaSourceStream::new(Box::new(data), Default::default());
+    let hint = Hint::new();
+    let format_opts: FormatOptions = Default::default();
+    let metadata_opts: MetadataOptions = Default::default();
+    let probe = symphonia::default::get_probe()
+        .format(&hint, stream, &format_opts, &metadata_opts)
+        .map_err(|_| ())?;
+
+    let format = probe.format;
+    Ok(format)
+}

src/server/decoder/opus.rs

@@ -1,58 +1,117 @@
-use std::{
+use std::time::Duration;
-    io::{Read, Seek},
-    marker::PhantomData,
-};
 
-pub struct OpusDecoder<R>
+use super::{DecoderImpl, Demuxer};
-where
+
-    R: Read + Seek,
+pub struct OpusDecoder {
-{
+    demuxer: Demuxer,
-    _data: PhantomData<R>,
+    decoder: opus::Decoder,
+    channels: u16,
+    sample_rate: u32,
+    total_duration: Option<Duration>,
+    decoded_buffer: Vec<i16>,
+    decoded_cursor: usize,
+    decoded_length: usize,
 }
 
-impl<R> super::DecoderImpl<R> for OpusDecoder<R> where R: Read + Seek + Send + Sync + 'static {}
+impl OpusDecoder {
+    pub fn new(demuxer: Demuxer) -> Result<Self, ()> {
+        let track = demuxer.get_track().ok_or(())?;
+        let channels = track
+            .codec_params
+            .channels
+            .map(|c| c.count() as u16)
+            .ok_or(())?;
+        let sample_rate = track.codec_params.sample_rate.ok_or(())?;
+        let total_duration = match (track.codec_params.time_base, track.codec_params.n_frames) {
+            (Some(time_base), Some(n_frames)) => {
+                Some(Duration::from_secs(time_base.calc_time(n_frames).seconds))
+            }
+            _ => None,
+        };
 
-impl<R> OpusDecoder<R>
+        let channel_enum = match channels {
-where
+            1 => opus::Channels::Mono,
-    R: Read + Seek + Send + Sync + 'static,
+            2 => opus::Channels::Stereo,
-{
+            _ => return Err(()),
-    pub fn new(data: &mut R) -> Result<Self, ()> {
+        };
-        Err(())
+
+        let decoder = match opus::Decoder::new(sample_rate, channel_enum) {
+            Ok(decoder) => decoder,
+            Err(err) => {
+                eprintln!("Opus decoder creation error: {err:?}");
+                return Err(());
+            }
+        };
+
+        Ok(OpusDecoder {
+            demuxer,
+            decoder,
+            channels,
+            sample_rate,
+            total_duration,
+            decoded_buffer: vec![0; 10000], // TODO: Calculate better estimate
+            decoded_cursor: 0,
+            decoded_length: 0,
+        })
     }
 }
 
-impl<R> Iterator for OpusDecoder<R>
+impl DecoderImpl for OpusDecoder {}
-where
+
-    R: Read + Seek,
+impl Iterator for OpusDecoder {
-{
     type Item = i16;
 
     fn next(&mut self) -> Option<Self::Item> {
-        None
+        while self.decoded_cursor >= self.decoded_length {
+            // Fill buffer with the next decoded packet
+            match self.demuxer.reader.next_packet() {
+                Ok(packet) => {
+                    if packet.track_id() == self.demuxer.track_id {
+                        self.decoded_cursor = 0;
+                        self.decoded_length = 0;
+                        let result =
+                            self.decoder
+                                .decode(packet.buf(), &mut self.decoded_buffer, false);
+                        match result {
+                            Err(err) => {
+                                eprintln!("Opus packet decoding error: {err:?}");
+                                return None;
+                            }
+                            Ok(size) => {
+                                // opus_decode returns the number of decoded samples per channel,
+                                // so the actual written size is that times the number of channels
+                                self.decoded_length = size * self.channels as usize
+                            }
+                        };
+                    }
+                }
+                Err(_) => return None,
+            };
+        }
+        let item = self.decoded_buffer.get(self.decoded_cursor).cloned();
+        self.decoded_cursor += 1;
+        item
     }
 }
 
-impl<R> rodio::Source for OpusDecoder<R>
+impl rodio::Source for OpusDecoder {
-where
-    R: Read + Seek,
-{
     #[inline]
     fn current_frame_len(&self) -> Option<usize> {
-        todo!()
+        None
     }
 
     #[inline]
     fn channels(&self) -> u16 {
-        todo!()
+        self.channels
     }
 
     #[inline]
     fn sample_rate(&self) -> u32 {
-        todo!()
+        self.sample_rate
     }
 
     #[inline]
     fn total_duration(&self) -> Option<std::time::Duration> {
-        todo!()
+        self.total_duration
     }
 }

src/server/playback.rs

@@ -1,13 +1,15 @@
 use std::{
     fs::File,
-    io::BufReader,
     path::PathBuf,
     sync::{Arc, Mutex},
 };
 
 use rodio::{OutputStream, Sink};
 
-use super::{decoder::Decoder, Server};
+use super::{
+    decoder::{Decoder, DecoderImpl, SourceWrapper},
+    Server,
+};
 
 // HACK: hard-coded path to track
 static TRACK: &str = "";
@@ -46,15 +48,33 @@ pub fn playback_manager(server: Arc<Mutex<Server>>) {
         for command in server.lock().unwrap().playback.command_queue.drain(..) {
             match command {
                 PlaybackCommand::Play(track) => {
-                    let file = BufReader::new(File::open(&track).unwrap());
+                    let mut source: Option<Box<dyn DecoderImpl<Item = i16>>> = None;
-                    let source = match Decoder::new(file) {
+
-                        Ok(source) => source,
+                    {
-                        Err(_) => continue,
+                        let file = File::open(&track).unwrap();
+                        if let Ok(decoder) = Decoder::rodio(SourceWrapper::from_file(file)) {
+                            println!("rodio:\n\tsample_rate: {:?}\n\ttotal_duration: {:?}\n\tchannels: {:?}\n\tcurrent_frame_len: {:?}", decoder.sample_rate(), decoder.total_duration(), decoder.channels(), decoder.current_frame_len());
+                            source = Some(decoder);
                         };
+                    }
+
+                    if source.is_none() {
+                        let file = File::open(&track).unwrap();
+                        if let Ok(decoder) = Decoder::custom(SourceWrapper::from_file(file)) {
+                            println!("custom:\n\tsample_rate: {:?}\n\ttotal_duration: {:?}\n\tchannels: {:?}\n\tcurrent_frame_len: {:?}", decoder.sample_rate(), decoder.total_duration(), decoder.channels(), decoder.current_frame_len());
+                            source = Some(decoder);
+                        }
+                    }
+
+                    match source {
+                        Some(source) => {
                             sink.clear();
                             sink.append(source);
                             sink.play();
                         }
+                        None => println!("No handler found for '{track:?}'"),
+                    }
+                }
                 PlaybackCommand::TogglePause => {
                     if !sink.empty() {
                         if sink.is_paused() {