//!
//! the writers are free functions over any [`Write`]. the header and the end byte can go in zstd
//! frames of their own.
use std::io::{self, Read, Write};
use std::io::{self, BufRead, Write};
use ipld_core::ipld::Ipld;
use serde::Serialize;
buf[len] = n as u8;
len + 1
}
/// yields the pairs in archive order. the first error ends it, and the pairs before an error are
/// from an archive that isn't valid, so a caller that can't take them back should collect first
/// the first error ends it. pairs before the error come from an invalid archive.
/// collect them all first if you can't take them back
pub struct Reader<R> {
inner: R,
meta: Vec<u8>,
/// the current pair, key then value
buf: Vec<u8>,
done: bool,
}
impl<R: Read> Reader<R> {
/// reads and checks the magic and the header. `inner` is read a byte at a time between keys
/// and values, so it should be buffered
impl<R: BufRead> Reader<R> {
pub fn new(mut inner: R) -> Result<Self, Error> {
let mut magic = [0; 4];
inner.read_exact(&mut magic)?;
if magic != MAGIC {
return Err(Error::Magic);
}
let len = read_varint(&mut inner)?;
let meta = read_len(&mut inner, len)?;
let mut meta = Vec::new();
read_into(&mut inner, &mut meta, len)?;
check_meta(&meta)?;
Ok(Self {
inner,
meta,
buf: Vec::new(),
done: false,
})
}
pub fn meta(&self) -> &[u8] {
&self.meta
}
fn pair(&mut self) -> Result<Option<Pair>, Error> {
let mut control = [0];
self.inner.read_exact(&mut control)?;
match control[0] {
// llvm stops inlining the read path once a program uses both this and the iterator, and short
// pairs then read twice as slow in `benches/star.rs`
#[inline(always)]
#[allow(clippy::type_complexity)]
pub fn next_pair(&mut self) -> Result<Option<(&[u8], &[u8])>, Error> {
if self.done {
return Ok(None);
}
let key_len = self.pair();
self.done = !matches!(key_len, Ok(Some(_)));
Ok(key_len?.map(|at| self.buf.split_at(at)))
}
/// the key's length, with the key and value in `buf`
#[inline(always)]
fn pair(&mut self) -> Result<Option<usize>, Error> {
match read_byte(&mut self.inner)? {
PAIR => {
self.buf.clear();
let len = read_varint(&mut self.inner)?;
let key = read_len(&mut self.inner, len)?;
read_into(&mut self.inner, &mut self.buf, len)?;
let key_len = self.buf.len();
let len = read_varint(&mut self.inner)?;
let value = read_len(&mut self.inner, len)?;
Ok(Some((key, value)))
read_into(&mut self.inner, &mut self.buf, len)?;
Ok(Some(key_len))
}
END => match self.inner.read_exact(&mut control) {
Ok(()) => Err(Error::TrailingData),
Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => Ok(None),
Err(e) => Err(Error::Io(e)),
},
END if self.inner.fill_buf()?.is_empty() => Ok(None),
END => Err(Error::TrailingData),
other => Err(Error::Control(other)),
}
}
}
impl<R: Read> Iterator for Reader<R> {
impl<R: BufRead> Iterator for Reader<R> {
type Item = Result<Pair, Error>;
fn next(&mut self) -> Option<Self::Item> {
if self.done {
return None;
}
let pair = self.pair();
self.done = !matches!(pair, Ok(Some(_)));
pair.transpose()
let pair = self.next_pair().transpose()?;
Some(pair.map(|(key, value)| (key.to_vec(), value.to_vec())))
}
}
#[inline(always)]
fn read_byte(r: &mut impl BufRead) -> Result<u8, Error> {
let byte = *r.fill_buf()?.first().ok_or(Error::Truncated)?;
r.consume(1);
Ok(byte)
}
#[inline(always)]
fn read_varint(r: &mut impl Read) -> Result<u64, Error> {
fn read_varint(r: &mut impl BufRead) -> Result<u64, Error> {
let mut n = 0;
for i in 0..VARINT_MAX_BYTES {
let mut byte = [0];
r.read_exact(&mut byte)?;
n |= u64::from(byte[0] & 0x7f) << (7 * i);
if byte[0] & 0x80 == 0 {
if byte[0] == 0 && i > 0 {
let byte = read_byte(r)?;
n |= u64::from(byte & 0x7f) << (7 * i);
if byte & 0x80 == 0 {
if byte == 0 && i > 0 {
return Err(Error::Varint("that isn't minimally encoded"));
}
return Ok(n);
}
}
Err(Error::Varint("longer than 9 bytes"))
}
/// grows with what arrives, so a huge length on a short input can't allocate it up front
fn read_len(r: &mut impl Read, len: u64) -> Result<Vec<u8>, Error> {
let mut out = Vec::new();
r.take(len).read_to_end(&mut out)?;
if out.len() as u64 != len {
return Err(Error::Truncated);
/// grows `buf` as bytes arrive and never ahead of them, because a huge length on a short input
/// would otherwise allocate it all up front
#[inline(always)]
fn read_into(r: &mut impl BufRead, buf: &mut Vec<u8>, len: u64) -> Result<(), Error> {
let mut left = usize::try_from(len).unwrap_or(usize::MAX);
while left > 0 {
let available = r.fill_buf()?;
if available.is_empty() {
return Err(Error::Truncated);
}
let n = available.len().min(left);
if buf.capacity() - buf.len() < n {
buf.reserve_exact(left.min(buf.len().max(n)));
}
buf.extend_from_slice(&available[..n]);
r.consume(n);
left -= n;
}
Ok(out)
Ok(())
}
/// a value that changes when decoded and re-encoded isn't DRISL, which has one encoding per value
fn check_meta(meta: &[u8]) -> Result<(), Error> {
let reader = Reader::new(&star[..]).unwrap();
assert_eq!(reader.meta(), serde_ipld_dagcbor::to_vec(&meta).unwrap());
let read: Vec<_> = reader.collect::<Result<_, _>>().unwrap();
assert_eq!(read, pairs);
let mut reader = Reader::new(std::io::BufReader::with_capacity(1, &star[..])).unwrap();
for (key, value) in &pairs {
assert_eq!(reader.next_pair().unwrap(), Some((&key[..], &value[..])));
}
assert_eq!(reader.next_pair().unwrap(), None);
}
#[test]
fn a_header_without_a_text_type_is_not_written() {