Something went wrong. Try again.
STreaming ARchive format: simpler, stricter, smaller than CARs for a subset of use-cases.
star car archive atproto
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485use crate::error::Result;use crate::parser::StarParser;use crate::types::StarItem;use cid::Cid;use std::collections::VecDeque;use std::io::Read;
pub struct StarIterator<R> { reader: R, parser: StarParser, buffer: VecDeque<u8>,}
impl<R: Read> StarIterator<R> { pub fn new(reader: R) -> Self { Self { reader, parser: StarParser::new(), buffer: VecDeque::new(), } }
/// Returns an iterator over all items (Nodes, Records, Header) in the tree (order preserved). /// This is the identity iterator. pub fn iter_tree(self) -> Self { self }
/// Returns an iterator over triples of (key, cid, Option<record>). /// Skips Commit header and MST Nodes. pub fn iter(self) -> impl Iterator<Item = Result<(Vec<u8>, Cid, Option<Vec<u8>>)>> { self.filter_map(|res| match res { Ok(StarItem::Record { key, cid, content }) => Some(Ok((key, cid, content))), Ok(_) => None, Err(e) => Some(Err(e)), }) }
/// Returns an iterator over (key, cid) pairs. pub fn iter_keys(self) -> impl Iterator<Item = Result<(Vec<u8>, Cid)>> { self.filter_map(|res| match res { Ok(StarItem::Record { key, cid, .. }) => Some(Ok((key, cid))), Ok(_) => None, Err(e) => Some(Err(e)), }) }}
impl<R: Read> Iterator for StarIterator<R> { type Item = Result<StarItem>;
fn next(&mut self) -> Option<Self::Item> { loop { // VecDeque must be contiguous to be parsed as a slice. // make_contiguous() allows us to get a single slice, but might involve a memory copy/move // if the buffer is wrapped. This is amortized O(1) in many cases but can be O(N). // However, it solves the "head removal" problem perfectly (pop_front is O(1)).
self.buffer.make_contiguous(); let (slice, _) = self.buffer.as_slices();
match self.parser.parse(slice) { Ok((consumed, Some(item))) => { self.buffer.drain(..consumed); return Some(Ok(item)); } Ok((consumed, None)) => { self.buffer.drain(..consumed);
// Read more // We can't read directly into VecDeque easily without temporary buffer // because it doesn't expose a mutable slice to uninitialized memory. let mut temp = [0u8; 8192]; match self.reader.read(&mut temp) { Ok(0) => return None, // EOF Ok(n) => self.buffer.extend(&temp[..n]), Err(e) => return Some(Err(e.into())), } } Err(e) => return Some(Err(e)), } } }}