How to decode an async stream?

[ababo]

Since Symphonia lacks async support I would like to find a proper way to proactively feed with incoming chunks rather than via Read::read().

Indeed when I try to use a ring buffer that acts like a reader it's drained eventually making FormatReader return std::io::Error. After the ring buffer is topped up the format reader can be recovered, but at cost of skipping some data (an ogg page in my case) which is unacceptable.

I need a way to feed the format reader proactively with data or at least make its state fully recoverable.

[jBernavaPrah]

Hi @ababo

Did you find a way?

[ababo]

@jBernavaPrah No, I used an async packet decoder from ogg crate.

[jBernavaPrah]

Hi,

I have found a way to do it.

I'm not sure is the best way to do it, but it's working. :)

You need to implement the MediaSource to read the data and then use tokio::task::blocking_spawn to read the data without blocking the main thread.

Happy to help if needed.

[ababo]

Hi,
The problem of this approach is that it requires an entire OS thread to be allocated per each connection. For high-load server this doesn't look like a good solution.

[aschey]

I'm not sure if this is exactly the same use case you're asking for, but I made a crate to handle streaming data asynchronously https://github.com/aschey/stream-download-rs. It's not specific to audio data, but it will take care of treating an asynchronous stream like a normal Read + Seek source.

[Enitoni]

Here's how I did it in my project.

/// Bridges an async [Loadable] with a synchronous [MediaSource].
struct LoadableMediaSource {
    rt: Handle,
    loadable: BoxedLoadable,
}

impl MediaSource for LoadableMediaSource {
    fn is_seekable(&self) -> bool {
        self.rt.block_on(self.loadable.seekable())
    }

    fn byte_len(&self) -> Option<u64> {
        let result = self.rt.block_on(self.loadable.length());

        result.and_then(|l| match l {
            LoaderLength::Bytes(bytes) => Some(bytes as u64),
            LoaderLength::Time(_) => None,
        })
    }
}

impl Seek for LoadableMediaSource {
    fn seek(&mut self, pos: SeekFrom) -> std::io::Result<u64> {
        let result = self.rt.block_on(self.loadable.seek(pos));

        result
            .map_err(|e| std::io::Error::other(format!("Seek failed: {:?}", e)))
            .map(|seek| seek as u64)
    }
}

impl Read for LoadableMediaSource {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        let result = self.rt.block_on(self.loadable.read(buf));

        result
            .map_err(|e| std::io::Error::other(format!("Read failed: {:?}", e)))
            .map(|read| match read {
                ReadResult::More(bytes) => bytes,
                ReadResult::End(bytes) => bytes,
            })
    }
}