1use std::fmt;
21use std::marker::PhantomData;
22
23use crate::binary::{Decoder, Writer};
24use crate::borrowed::ValueRef;
25use crate::frame::{Packer, Unpacker};
26use crate::schema::{Reader, Schema, Shape};
27use crate::value::{Seg, Value};
28
29pub struct Stream<T> {
31 items: Vec<u8>,
33 len: usize,
34 item: PhantomData<fn() -> T>,
35}
36
37impl<T: Schema> Stream<T> {
38 pub fn writer() -> StreamWriter<T> {
40 StreamWriter {
41 w: Writer::body(),
42 len: 0,
43 item: PhantomData,
44 }
45 }
46
47 pub fn len(&self) -> usize {
48 self.len
49 }
50
51 pub fn is_empty(&self) -> bool {
52 self.len == 0
53 }
54
55 pub fn iter(&self) -> impl Iterator<Item = T> + '_ {
57 let mut d = Decoder::body(&self.items);
58 (0..self.len).map_while(move |_| T::decode(&mut d))
59 }
60
61 pub fn to_vec(&self) -> Vec<T> {
63 self.iter().collect()
64 }
65
66 pub fn bytes_len(&self) -> usize {
68 self.items.len()
69 }
70}
71
72pub struct StreamWriter<T> {
74 w: Writer,
75 len: usize,
76 item: PhantomData<fn() -> T>,
77}
78
79impl<T: Schema> StreamWriter<T> {
80 pub fn push(&mut self, item: &T) {
81 item.encode(&mut self.w);
82 self.len += 1;
83 }
84
85 pub fn finish(self) -> Stream<T> {
86 Stream {
87 items: self.w.finish(),
88 len: self.len,
89 item: PhantomData,
90 }
91 }
92}
93
94impl<T: Schema> FromIterator<T> for Stream<T> {
95 fn from_iter<I: IntoIterator<Item = T>>(items: I) -> Self {
96 let mut w = Stream::writer();
97 for item in items {
98 w.push(&item);
99 }
100 w.finish()
101 }
102}
103
104impl<T: Schema> From<Vec<T>> for Stream<T> {
105 fn from(items: Vec<T>) -> Self {
106 items.into_iter().collect()
107 }
108}
109
110impl<T> Clone for Stream<T> {
111 fn clone(&self) -> Self {
112 Stream {
113 items: self.items.clone(),
114 len: self.len,
115 item: PhantomData,
116 }
117 }
118}
119
120impl<T> PartialEq for Stream<T> {
121 fn eq(&self, other: &Self) -> bool {
122 self.len == other.len && self.items == other.items
123 }
124}
125
126impl<T> Default for Stream<T> {
127 fn default() -> Self {
128 Stream {
129 items: Vec::new(),
130 len: 0,
131 item: PhantomData,
132 }
133 }
134}
135
136impl<T> fmt::Debug for Stream<T> {
137 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
138 write!(f, "Stream({} items, {} bytes)", self.len, self.items.len())
139 }
140}
141
142impl<T: Schema> Schema for Stream<T> {
143 fn shape() -> Shape {
144 Shape::List(T::shape)
145 }
146 fn read(v: &ValueRef<'_>, cx: &mut Reader) -> Option<Self> {
148 let ValueRef::List(items) = v else {
149 cx.expected("a list", v);
150 return None;
151 };
152 let mut w = Stream::writer();
153 let mut ok = true;
154 for (i, item) in items.iter().enumerate() {
155 match cx.at(Seg::Index(i), |cx| T::read(item, cx)) {
156 Some(x) => w.push(&x),
157 None => ok = false,
158 }
159 }
160 ok.then(|| w.finish())
161 }
162 fn write(&self) -> Value {
163 Value::List(self.iter().map(|item| item.write()).collect())
164 }
165 fn encode(&self, w: &mut Writer) {
166 w.stream_body(self.len, &self.items);
167 }
168 fn decode(d: &mut Decoder<'_>) -> Option<Self> {
169 let mut body = Decoder::body(d.stream()?);
170 let len = body.stream_count()?;
171 Some(Stream {
172 items: body.rest().to_vec(),
173 len,
174 item: PhantomData,
175 })
176 }
177 fn pack(&self, p: &mut Packer) {
179 T::pack_list(&self.to_vec(), p);
180 }
181 fn unpack(u: &mut Unpacker<'_>) -> Option<Self> {
182 T::unpack_list(u).map(Stream::from)
183 }
184}