Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Simple header search and remove dependency to io::Seek #18

Open
wants to merge 6 commits into
base: master
Choose a base branch
from
Open

Simple header search and remove dependency to io::Seek #18

Changes from 5 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
5790448
Simplify the logic of finding a new page header
TonalidadeHidrica Nov 21, 2020
22c3137
Remove requirement for io::Seek for reader
TonalidadeHidrica Nov 21, 2020
a197365
Add additional notes about the reader on docs
TonalidadeHidrica Nov 21, 2020
5f41374
Return None if no new page header is fully available
TonalidadeHidrica Nov 23, 2020
8373a62
Modify the code so that it compiles on 1.27.2
TonalidadeHidrica Nov 23, 2020
c1fe4dd
Apply suggestions from code review
est31 May 10, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
223 changes: 41 additions & 182 deletions src/reading.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,15 +12,15 @@ Reading logic

use std::error;
use std::io;
use std::io::{Cursor, Read, Write, SeekFrom, Error, ErrorKind};
use std::io::{Cursor, Write, SeekFrom, Error, ErrorKind};
use byteorder::{ReadBytesExt, LittleEndian};
use std::collections::HashMap;
use std::collections::hash_map::Entry;
use std::fmt::{Display, Formatter, Error as FmtError};
use std::mem::replace;
use crc::vorbis_crc32_update;
use Packet;
use std::io::Seek;
use std::io::Read;

/// Error that can be raised when decoding an Ogg transport.
#[derive(Debug)]
Expand Down Expand Up @@ -59,7 +59,7 @@ impl error::Error for OggReadError {

fn cause(&self) -> Option<&dyn error::Error> {
match *self {
OggReadError::ReadError(ref err) => Some(err as &error::Error),
OggReadError::ReadError(ref err) => Some(err as &dyn error::Error),
_ => None
}
}
Expand Down Expand Up @@ -528,177 +528,20 @@ impl BasePacketReader {
}
}

#[derive(Clone, Copy)]
enum UntilPageHeaderReaderMode {
Searching,
FoundWithNeeded(u8),
SeekNeeded(i32),
Found,
#[derive(Default)]
struct MagicFinder {
len: usize,
est31 marked this conversation as resolved.
Show resolved Hide resolved
}

enum UntilPageHeaderResult {
Eof,
Found,
ReadNeeded,
SeekNeeded,
}

struct UntilPageHeaderReader {
mode :UntilPageHeaderReaderMode,
/// Capture pattern offset. Needed so that if we only partially
/// recognized the capture pattern, we later on only check the
/// remaining part.
cpt_of :u8,
/// The return buffer.
ret_buf :[u8; 27],
read_amount :usize,
}

impl UntilPageHeaderReader {
pub fn new() -> Self {
UntilPageHeaderReader {
mode : UntilPageHeaderReaderMode::Searching,
cpt_of : 0,
ret_buf : [0; 27],
read_amount : 0,
impl MagicFinder {
fn feed(&mut self, b: u8){
est31 marked this conversation as resolved.
Show resolved Hide resolved
match (b, self.len) {
(b'O', _) => self.len = 1,
(b'g', 1..=2) | (b'S', 3) => self.len += 1,
_ => self.len = 0,
}
}
/// Returns Some(off), where off is the offset of the last byte
/// of the capture pattern if it's found, None if the capture pattern
/// is not inside the passed slice.
///
/// Changes the capture pattern offset accordingly
fn check_arr(&mut self, arr :&[u8]) -> Option<usize> {
for (i, ch) in arr.iter().enumerate() {
match *ch {
b'O' => self.cpt_of = 1,
b'g' if self.cpt_of == 1 || self.cpt_of == 2 => self.cpt_of += 1,
b'S' if self.cpt_of == 3 => return Some(i),
_ => self.cpt_of = 0,
}
}
return None;
}
/// Do one read exactly, and if it was successful,
/// return Ok(true) if the full header has been read and can be extracted with
///
/// or return Ok(false) if the
pub fn do_read<R :Read>(&mut self, mut rdr :R)
-> Result<UntilPageHeaderResult, OggReadError> {
use self::UntilPageHeaderReaderMode::*;
use self::UntilPageHeaderResult as Res;
// The array's size is freely choseable, but must be > 27,
// and must well fit into an i32 (needs to be stored in SeekNeeded)
let mut buf :[u8; 1024] = [0; 1024];

let rd_len = tri!(rdr.read(if self.read_amount < 27 {
// This is an optimisation for the most likely case:
// the next page directly follows the current read position.
// Then it would be a waste to read more than the needed amount.
&mut buf[0 .. 27 - self.read_amount]
} else {
match self.mode {
Searching => &mut buf,
FoundWithNeeded(amount) => &mut buf[0 .. amount as usize],
SeekNeeded(_) => return Ok(Res::SeekNeeded),
Found => return Ok(Res::Found),
}
}));

if rd_len == 0 {
// Reached EOF.
if self.read_amount == 0 {
// If we have read nothing yet, there is no data
// but ogg data, meaning the stream ends legally
// and without corruption.
return Ok(Res::Eof);
} else {
// There is most likely a corruption here.
// I'm not sure, but the ogg spec doesn't say that
// random data past the last ogg page is allowed,
// so we just assume it's not allowed.
tri!(Err(OggReadError::NoCapturePatternFound));
}
}
self.read_amount += rd_len;

// 150 kb gives us a bit of safety: we can survive
// up to one page with a corrupted capture pattern
// after having seeked right after a capture pattern
// of an earlier page.
let read_amount_max = 150 * 1024;
if self.read_amount > read_amount_max {
// Exhaustive searching for the capture pattern
// has returned no ogg capture pattern.
tri!(Err(OggReadError::NoCapturePatternFound));
}

let rd_buf = &buf[0 .. rd_len];

use std::cmp::min;
let (off, needed) = match self.mode {
Searching => match self.check_arr(rd_buf) {
// Capture pattern found
Some(off) => {
self.ret_buf[0] = b'O';
self.ret_buf[1] = b'g';
self.ret_buf[2] = b'g';
self.ret_buf[3] = b'S'; // (Not actually needed)
(off, 24)
},
// Nothing found
None => return Ok(Res::ReadNeeded),
},
FoundWithNeeded(needed) => {
(0, needed as usize)
},
_ => unimplemented!(),
};

let fnd_buf = &rd_buf[off..];

let copy_amount = min(needed, fnd_buf.len());
let start_fill = 27 - needed;
(&mut self.ret_buf[start_fill .. copy_amount + start_fill])
.copy_from_slice(&fnd_buf[0 .. copy_amount]);
if fnd_buf.len() == needed {
// Capture pattern found!
self.mode = Found;
return Ok(Res::Found);
} else if fnd_buf.len() < needed {
// We still have to read some content.
let needed_new = needed - copy_amount;
self.mode = FoundWithNeeded(needed_new as u8);
return Ok(Res::ReadNeeded);
} else {
// We have read too much content (exceeding the header).
// Seek back so that we are at the position
// right after the header.

self.mode = SeekNeeded(needed as i32 - fnd_buf.len() as i32);
return Ok(Res::SeekNeeded);
}
}
pub fn do_seek<S :Seek>(&mut self, mut skr :S)
-> Result<UntilPageHeaderResult, OggReadError> {
use self::UntilPageHeaderReaderMode::*;
use self::UntilPageHeaderResult as Res;
match self.mode {
Searching | FoundWithNeeded(_) => Ok(Res::ReadNeeded),
SeekNeeded(offs) => {
tri!(skr.seek(SeekFrom::Current(offs as i64)));
self.mode = Found;
Ok(Res::Found)
},
Found => Ok(Res::Found),
}
}
pub fn into_header(self) -> [u8; 27] {
use self::UntilPageHeaderReaderMode::*;
match self.mode {
Found => self.ret_buf,
_ => panic!("wrong mode"),
}
fn found(&self) -> bool {
self.len == 4
}
}

Expand All @@ -711,14 +554,17 @@ This reader is not async ready. It does not keep its internal state
consistent when it encounters the `WouldBlock` error kind.
If you desire async functionality, consider enabling the `async` feature
and look into the async module.

The reader passed to this packet reader should be buffered properly,
since `Read::read` will be called many times.
*/
pub struct PacketReader<T :io::Read + io::Seek> {
pub struct PacketReader<T :io::Read> {
rdr :T,

base_pck_rdr :BasePacketReader,
}

impl<T :io::Read + io::Seek> PacketReader<T> {
impl<T :io::Read> PacketReader<T> {
/// Constructs a new `PacketReader` with a given `Read`.
pub fn new(rdr :T) -> PacketReader<T> {
PacketReader { rdr, base_pck_rdr : BasePacketReader::new() }
Expand Down Expand Up @@ -767,18 +613,29 @@ impl<T :io::Read + io::Seek> PacketReader<T> {
/// Ok(None) is returned if the stream has ended without an uncompleted page
/// or non page data after the last page (if any) present.
fn read_until_pg_header(&mut self) -> Result<Option<[u8; 27]>, OggReadError> {
let mut r = UntilPageHeaderReader::new();
use self::UntilPageHeaderResult::*;
let mut res = tri!(r.do_read(&mut self.rdr));
let mut ret = [0u8; 27];
ret[..4].copy_from_slice(b"OggS");
let mut finder = MagicFinder::default();
while !finder.found() {
let next = match self.rdr.by_ref().bytes().next() {
Some(b) => tri!(b),
None => return Ok(None),
};
finder.feed(next);
}
let mut pos = 4;
loop {
res = match res {
Eof => return Ok(None),
Found => break,
ReadNeeded => tri!(r.do_read(&mut self.rdr)),
SeekNeeded => tri!(r.do_seek(&mut self.rdr))
let read = tri!(self.rdr.read(&mut ret[pos..]));
if read == 0 {
break;
}
pos += read;
}
if pos == ret.len() {
Ok(Some(ret))
} else {
Ok(None)
}
Ok(Some(r.into_header()))
}

/// Parses and reads a new OGG page
Expand All @@ -803,7 +660,9 @@ impl<T :io::Read + io::Seek> PacketReader<T> {

Ok(Some(tri!(pg_prs.parse_packet_data(packet_data))))
}
}

impl<T :io::Read + io::Seek> PacketReader<T> {
/// Seeks the underlying reader
///
/// Seeks the reader that this PacketReader bases on by the specified
Expand Down