substrate_data/wire/
stream.rs

1//! [`Stream`]: a list kept in the binary form — its items written as they
2//! come (no `Vec` of them on the way) and read back one at a time, when
3//! asked. For the big ones: a table's rows, a picture's pixels, a log.
4//!
5//! ```
6//! use substrate_data::{Schema, Stream};
7//!
8//! #[derive(Schema)]
9//! struct Sheet {
10//!     name: String,
11//!     cells: Stream<u16>,
12//! }
13//!
14//! let sheet = Sheet { name: "q3".into(), cells: (0..1000u16).map(|i| i % 3).collect() };
15//! let back: Sheet = substrate_data::read_bytes(&substrate_data::to_bytes(&sheet)).unwrap();
16//! assert_eq!(back.cells.len(), 1000);
17//! assert_eq!(back.cells.iter().filter(|c| *c == 2).count(), 333);
18//! ```
19
20use std::fmt;
21use std::marker::PhantomData;
22
23use crate::binary::{Decoder, Writer};
24use crate::frame::{Packer, Unpacker};
25use crate::model::borrowed::ValueRef;
26use crate::model::value::{Seg, Value};
27use crate::schema::{Reader, Schema, Shape};
28
29/// A list of `T` kept written: its body in the binary form, its strings its own.
30pub struct Stream<T> {
31    /// The items, written one after another.
32    items: Vec<u8>,
33    len: usize,
34    item: PhantomData<fn() -> T>,
35}
36
37impl<T: Schema> Stream<T> {
38    /// Writes items one by one ([`StreamWriter::push`]).
39    #[must_use]
40    pub fn writer() -> StreamWriter<T> {
41        StreamWriter {
42            w: Writer::body(),
43            len: 0,
44            item: PhantomData,
45        }
46    }
47
48    /// How many items.
49    #[must_use]
50    pub fn len(&self) -> usize {
51        self.len
52    }
53
54    /// Whether it has none.
55    #[must_use]
56    pub fn is_empty(&self) -> bool {
57        self.len == 0
58    }
59
60    /// The items, read as they're asked for; it stops at one it can't read.
61    pub fn iter(&self) -> impl Iterator<Item = T> + '_ {
62        let mut d = Decoder::body(&self.items);
63        (0..self.len).map_while(move |_| T::decode(&mut d))
64    }
65
66    /// Every item, read now.
67    #[must_use]
68    pub fn to_vec(&self) -> Vec<T> {
69        self.iter().collect()
70    }
71
72    /// How many bytes it takes.
73    #[must_use]
74    pub fn bytes_len(&self) -> usize {
75        self.items.len()
76    }
77}
78
79/// Writes a [`Stream`]'s items as they come.
80#[derive(Debug)]
81pub struct StreamWriter<T> {
82    w: Writer,
83    len: usize,
84    item: PhantomData<fn() -> T>,
85}
86
87impl<T: Schema> StreamWriter<T> {
88    /// Writes `item`.
89    pub fn push(&mut self, item: &T) {
90        item.encode(&mut self.w);
91        self.len += 1;
92    }
93
94    /// The stream written.
95    #[must_use]
96    pub fn finish(self) -> Stream<T> {
97        Stream {
98            items: self.w.finish(),
99            len: self.len,
100            item: PhantomData,
101        }
102    }
103}
104
105impl<T: Schema> FromIterator<T> for Stream<T> {
106    fn from_iter<I: IntoIterator<Item = T>>(items: I) -> Self {
107        let mut w = Stream::writer();
108        for item in items {
109            w.push(&item);
110        }
111        w.finish()
112    }
113}
114
115impl<T: Schema> From<Vec<T>> for Stream<T> {
116    fn from(items: Vec<T>) -> Self {
117        items.into_iter().collect()
118    }
119}
120
121impl<T> Clone for Stream<T> {
122    fn clone(&self) -> Self {
123        Stream {
124            items: self.items.clone(),
125            len: self.len,
126            item: PhantomData,
127        }
128    }
129}
130
131impl<T> PartialEq for Stream<T> {
132    fn eq(&self, other: &Self) -> bool {
133        self.len == other.len && self.items == other.items
134    }
135}
136
137impl<T> Default for Stream<T> {
138    fn default() -> Self {
139        Stream {
140            items: Vec::new(),
141            len: 0,
142            item: PhantomData,
143        }
144    }
145}
146
147impl<T> fmt::Debug for Stream<T> {
148    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
149        write!(f, "Stream({} items, {} bytes)", self.len, self.items.len())
150    }
151}
152
153impl<T: Schema> Schema for Stream<T> {
154    fn shape() -> Shape {
155        Shape::List(T::shape)
156    }
157    /// The slow way (a text file, a value): read whole, then written.
158    fn read(v: &ValueRef<'_>, cx: &mut Reader<'_>) -> Option<Self> {
159        let ValueRef::List(items) = v else {
160            cx.expected(crate::schema::Expected::List, v);
161            return None;
162        };
163        let mut w = Stream::writer();
164        let mut ok = true;
165        for (i, item) in items.iter().enumerate() {
166            match cx.at(Seg::Index(i), |cx| T::read(item, cx)) {
167                Some(x) => w.push(&x),
168                None => ok = false,
169            }
170        }
171        ok.then(|| w.finish())
172    }
173    fn write(&self) -> Value {
174        Value::List(self.iter().map(|item| item.write()).collect())
175    }
176    fn encode(&self, w: &mut Writer) {
177        w.stream_body(self.len, &self.items);
178    }
179    fn decode(d: &mut Decoder<'_>) -> Option<Self> {
180        let mut body = Decoder::body(d.stream()?);
181        let len = body.stream_count()?;
182        Some(Stream {
183            items: body.rest().to_vec(),
184            len,
185            item: PhantomData,
186        })
187    }
188    /// In a frame it's a list, as its shape says.
189    fn pack(&self, p: &mut Packer) {
190        T::pack_list(&self.to_vec(), p);
191    }
192    fn unpack(u: &mut Unpacker<'_>) -> Option<Self> {
193        T::unpack_list(u).map(Stream::from)
194    }
195}
substrate_data/wire/stream.rs

substrate_data/wire/
stream.rs
