substrate_data/wire/

stream.rs

//! [`Stream`]: a list kept in the binary form — its items written as they
//! come (no `Vec` of them on the way), checked when the stream is decoded
//! and read back one at a time, when asked. For the big ones: a table's rows, a picture's pixels, a log.
//!
//! ```
//! use substrate_data::{Schema, Stream};
//!
//! #[derive(Schema)]
//! struct Sheet {
//!     name: String,
//!     cells: Stream<u16>,
//! }
//!
//! let sheet = Sheet { name: "q3".into(), cells: (0..1000u16).map(|i| i % 3).collect() };
//! let back: Sheet = substrate_data::read_bytes(&substrate_data::to_bytes(&sheet)).unwrap();
//! assert_eq!(back.cells.len(), 1000);
//! assert_eq!(back.cells.iter().filter(|c| *c == 2).count(), 333);
//! ```

use std::fmt;
use std::marker::PhantomData;

use crate::binary::{Decoder, Writer};
use crate::frame::{Packer, Unpacker};
use crate::model::borrowed::ValueRef;
use crate::model::value::{Seg, Value};
use crate::schema::{Reader, Schema, Shape};

/// A list of `T` kept written: its body in the binary form, its strings its own.
pub struct Stream<T> {
    /// The items, written one after another.
    items: Vec<u8>,
    len: usize,
    item: PhantomData<fn() -> T>,
}

impl<T: Schema> Stream<T> {
    /// Writes items one by one ([`StreamWriter::push`]).
    #[must_use]
    pub fn writer() -> StreamWriter<T> {
        StreamWriter {
            w: Writer::body(),
            len: 0,
            item: PhantomData,
        }
    }

    /// How many items.
    #[must_use]
    pub fn len(&self) -> usize {
        self.len
    }

    /// Whether it has none.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.len == 0
    }

    /// The items, read as they're asked for. A stream that was decoded has had
    /// every item checked, so none is unreadable.
    pub fn iter(&self) -> impl Iterator<Item = T> + '_ {
        let mut d = Decoder::body(&self.items);
        (0..self.len).map_while(move |_| T::decode(&mut d))
    }

    /// Every item, read now.
    #[must_use]
    pub fn to_vec(&self) -> Vec<T> {
        self.iter().collect()
    }

    /// How many bytes it takes.
    #[must_use]
    pub fn bytes_len(&self) -> usize {
        self.items.len()
    }
}

/// Writes a [`Stream`]'s items as they come.
#[derive(Debug)]
pub struct StreamWriter<T> {
    w: Writer,
    len: usize,
    item: PhantomData<fn() -> T>,
}

impl<T: Schema> StreamWriter<T> {
    /// Writes `item`.
    pub fn push(&mut self, item: &T) {
        item.encode(&mut self.w);
        self.len += 1;
    }

    /// The stream written.
    #[must_use]
    pub fn finish(self) -> Stream<T> {
        Stream {
            items: self.w.finish(),
            len: self.len,
            item: PhantomData,
        }
    }
}

impl<T: Schema> FromIterator<T> for Stream<T> {
    fn from_iter<I: IntoIterator<Item = T>>(items: I) -> Self {
        let mut w = Stream::writer();
        for item in items {
            w.push(&item);
        }
        w.finish()
    }
}

impl<T: Schema> From<Vec<T>> for Stream<T> {
    fn from(items: Vec<T>) -> Self {
        items.into_iter().collect()
    }
}

impl<T> Clone for Stream<T> {
    fn clone(&self) -> Self {
        Stream {
            items: self.items.clone(),
            len: self.len,
            item: PhantomData,
        }
    }
}

impl<T> PartialEq for Stream<T> {
    fn eq(&self, other: &Self) -> bool {
        self.len == other.len && self.items == other.items
    }
}

impl<T> Default for Stream<T> {
    fn default() -> Self {
        Stream {
            items: Vec::new(),
            len: 0,
            item: PhantomData,
        }
    }
}

impl<T> fmt::Debug for Stream<T> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "Stream({} items, {} bytes)", self.len, self.items.len())
    }
}

impl<T: Schema> Schema for Stream<T> {
    fn shape() -> Shape {
        Shape::List(T::shape)
    }
    /// The slow way (a text file, a value): read whole, then written.
    fn read(v: &ValueRef<'_>, cx: &mut Reader<'_>) -> Option<Self> {
        let ValueRef::List(items) = v else {
            cx.expected(crate::schema::Expected::List, v);
            return None;
        };
        let mut w = Stream::writer();
        let mut ok = true;
        for (i, item) in items.iter().enumerate() {
            match cx.at(Seg::Index(i), |cx| T::read(item, cx)) {
                Some(x) => w.push(&x),
                None => ok = false,
            }
        }
        ok.then(|| w.finish())
    }
    fn write(&self) -> Value {
        Value::List(self.iter().map(|item| item.write()).collect())
    }
    fn encode(&self, w: &mut Writer) {
        w.stream_body(self.len, &self.items);
    }
    fn decode(d: &mut Decoder<'_>) -> Option<Self> {
        let mut body = Decoder::body(d.stream()?);
        let len = body.stream_count()?;
        // Every item is read once now, so a stream that decodes is one whose
        // items all read: an item this `T` cannot read (a newer item type,
        // a corrupt byte) fails the stream, as it does in the text form and
        // in a frame, and the slow path names its index.
        let items = body.rest();
        let mut check = Decoder::body(items);
        for _ in 0..len {
            T::decode(&mut check)?;
        }
        check.at_end().then(|| Stream {
            items: items.to_vec(),
            len,
            item: PhantomData,
        })
    }
    /// In a frame it's a list, as its shape says.
    fn pack(&self, p: &mut Packer) {
        T::pack_list(&self.to_vec(), p);
    }
    fn unpack(u: &mut Unpacker<'_>) -> Option<Self> {
        T::unpack_list(u).map(Stream::from)
    }
}