use std::fmt;
use std::marker::PhantomData;
use super::item_error::ItemError;
use super::items::Items;
use super::stream_writer::StreamWriter;
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};
pub struct Stream<T> {
items: Vec<u8>,
len: usize,
item: PhantomData<fn() -> T>,
}
impl<T> Stream<T> {
pub(super) fn from_written(items: Vec<u8>, len: usize) -> Self {
Stream {
items,
len,
item: PhantomData,
}
}
#[must_use]
pub fn len(&self) -> usize {
self.len
}
#[must_use]
pub fn is_empty(&self) -> bool {
self.len == 0
}
#[must_use]
pub fn bytes_len(&self) -> usize {
self.items.len()
}
}
impl<T: Schema> Stream<T> {
#[must_use]
pub fn writer() -> StreamWriter<T> {
StreamWriter {
w: Writer::body(),
len: 0,
item: PhantomData,
}
}
#[must_use]
pub fn iter(&self) -> Items<'_, T> {
Items::new(&self.items, self.len)
}
pub fn to_vec(&self) -> Result<Vec<T>, ItemError> {
self.iter().collect()
}
pub fn validate(&self) -> Result<(), ItemError> {
self.iter().try_for_each(|item| item.map(drop))
}
}
impl<'a, T: Schema> IntoIterator for &'a Stream<T> {
type Item = Result<T, ItemError>;
type IntoIter = Items<'a, T>;
fn into_iter(self) -> Items<'a, T> {
self.iter()
}
}
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::from_written(self.items.clone(), self.len)
}
}
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::from_written(Vec::new(), 0)
}
}
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)
}
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_while(Result::ok)
.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()?;
Some(Stream::from_written(body.rest().to_vec(), len))
}
fn pack(&self, p: &mut Packer) {
let items: Vec<T> = self.iter().map_while(Result::ok).collect();
T::pack_list(&items, p);
}
fn unpack(u: &mut Unpacker<'_>) -> Option<Self> {
T::unpack_list(u).map(Stream::from)
}
}