use core::any::Any;
use core::marker::PhantomData;
use std::io::{Read, Write};
use crate::Context;
use crate::de::{
Deserialize, DeserializeDriver, DeserializeOwned, Deserializer, StreamDeserializer,
};
use crate::error::{Error, ErrorKind};
use crate::ser::{Serialize, SerializeDriver, Serializer, StreamSerializer};
use crate::stream::{
DEFAULT_BUFFER_LIMIT, ElementReader, ElementStatus, InputBuffer, Part, Status,
};
pub struct Reader<R, D: StreamDeserializer> {
reader: R,
buffer: InputBuffer<D>,
pending: Option<Box<dyn Any + Send>>,
}
impl<R: Read, D: StreamDeserializer> Reader<R, D> {
pub fn new(reader: R, deserializer: D) -> Reader<R, D> {
Reader {
reader,
buffer: InputBuffer::new(deserializer),
pending: None,
}
}
pub fn set_context(&mut self, context: Context) {
self.buffer.set_context(context);
}
pub fn context(&self) -> &Context {
self.buffer.context()
}
fn ensure_idle(&self) -> Result<(), Error> {
match self.pending {
Some(_) => Err(Error::new(
ErrorKind::InvalidState,
"a value is being read with read_next",
)),
None => Ok(()),
}
}
fn read_more(&mut self) -> Result<(), Error> {
let buf = self.buffer.read_buf();
let read = loop {
match self.reader.read(buf) {
Ok(read) => break read,
Err(err) if err.kind() == std::io::ErrorKind::Interrupted => {}
Err(err) => return Err(err.into()),
}
};
if read == 0 {
self.buffer.set_eof();
} else {
self.buffer.filled(read);
}
Ok(())
}
fn fill(&mut self) -> Result<bool, Error> {
loop {
match self.buffer.poll()? {
Status::Ready => return Ok(true),
Status::End => return Ok(false),
Status::NeedInput => self.read_more()?,
}
}
}
fn drive_partial(&mut self, driver: &mut DeserializeDriver<'_, '_>) -> Result<bool, Error> {
loop {
match self.buffer.drive_partial(driver)? {
Status::Ready => return Ok(true),
Status::End => return Ok(false),
Status::NeedInput => self.read_more()?,
}
}
}
pub fn read<T: DeserializeOwned>(&mut self) -> Result<Option<T>, Error> {
self.read_with(|_| {})
}
pub fn read_with<T, F>(&mut self, setup: F) -> Result<Option<T>, Error>
where
T: DeserializeOwned,
F: FnOnce(&mut DeserializeDriver<'_, '_>),
{
self.ensure_idle()?;
if !self.buffer.supports_partial() {
if !self.fill()? {
return Ok(None);
}
return self
.buffer
.deserialize_with(|driver| setup(driver))
.map(Some);
}
let mut out = None::<T>;
{
let mut driver = DeserializeDriver::<'_, 'static>::new(&mut out);
setup(&mut driver);
if !self.drive_partial(&mut driver)? {
return Ok(None);
}
}
out.ok_or_else(|| Error::new(ErrorKind::EndOfFile, "empty input"))
.map(Some)
}
pub fn read_borrowed<'a, T: Deserialize<'a>>(&'a mut self) -> Result<Option<T>, Error> {
self.ensure_idle()?;
if !self.fill()? {
return Ok(None);
}
self.buffer.deserialize().map(Some)
}
pub fn read_next<T, E>(&mut self) -> Result<Option<Part<E, T>>, Error>
where
T: DeserializeOwned + 'static,
E: Send + 'static,
{
let mut reader = match self.pending.take() {
Some(pending) => match pending.downcast::<ElementReader<T, E>>() {
Ok(reader) => reader,
Err(pending) => {
self.pending = Some(pending);
return Err(Error::new(
ErrorKind::InvalidState,
"a value of another type is being read",
));
}
},
None => Box::new(ElementReader::<T, E>::new()),
};
loop {
match reader.poll(&mut self.buffer)? {
ElementStatus::Ready(next) => {
if reader.is_reading() {
self.pending = Some(reader);
}
return Ok(Some(next));
}
ElementStatus::End => return Ok(None),
ElementStatus::NeedInput => {
if let Err(err) = self.read_more() {
self.pending = Some(reader);
return Err(err);
}
}
}
}
}
pub fn is_end(&mut self) -> Result<bool, Error> {
self.ensure_idle()?;
loop {
match self.buffer.peek()? {
Status::Ready => return Ok(false),
Status::End => return Ok(true),
Status::NeedInput => self.read_more()?,
}
}
}
pub fn end(&mut self) -> Result<(), Error> {
self.ensure_idle()?;
if self.fill()? {
return Err(self.buffer.trailing_error());
}
Ok(())
}
pub fn iter<T: DeserializeOwned>(&mut self) -> Iter<'_, R, D, T> {
Iter {
reader: self,
failed: false,
_marker: PhantomData,
}
}
pub fn deserializer(&self) -> &D {
self.buffer.deserializer()
}
pub fn get_ref(&self) -> &R {
&self.reader
}
pub fn get_mut(&mut self) -> &mut R {
&mut self.reader
}
pub fn into_inner(self) -> R {
self.reader
}
pub fn into_parts(self) -> (R, D) {
(self.reader, self.buffer.into_parts().0)
}
}
impl<'de, R: Read, D: StreamDeserializer> Deserializer<'de> for Reader<R, D> {
fn drive(&mut self, driver: &mut DeserializeDriver<'_, 'de>) -> Result<(), Error> {
self.ensure_idle()?;
let found = if self.buffer.supports_partial() {
self.drive_partial(driver)?
} else if self.fill()? {
self.buffer.drive_transient(driver)?;
true
} else {
false
};
match found {
true => Ok(()),
false => Err(Error::new(ErrorKind::EndOfFile, "empty input")),
}
}
}
pub struct Iter<'r, R, D: StreamDeserializer, T> {
reader: &'r mut Reader<R, D>,
failed: bool,
_marker: PhantomData<fn() -> T>,
}
impl<R: Read, D: StreamDeserializer, T: DeserializeOwned> Iterator for Iter<'_, R, D, T> {
type Item = Result<T, Error>;
fn next(&mut self) -> Option<Self::Item> {
if self.failed {
return None;
}
match self.reader.read() {
Ok(Some(value)) => Some(Ok(value)),
Ok(None) => None,
Err(err) => {
self.failed = true;
Some(Err(err))
}
}
}
}
pub struct Writer<W, S: StreamSerializer> {
writer: W,
serializer: S,
limit: usize,
context: Context,
}
impl<W: Write, S: StreamSerializer> Writer<W, S> {
pub fn new(writer: W, serializer: S) -> Writer<W, S> {
Writer {
writer,
serializer,
limit: DEFAULT_BUFFER_LIMIT,
context: Context::new(),
}
}
pub fn set_context(&mut self, context: Context) {
self.context = context;
}
pub fn context(&self) -> &Context {
&self.context
}
pub fn set_buffer_limit(&mut self, limit: usize) {
self.limit = limit;
}
pub fn buffer_limit(&self) -> usize {
self.limit
}
pub fn write<T: Serialize + ?Sized>(&mut self, value: &T) -> Result<(), Error> {
self.write_driver(&mut SerializeDriver::new(&value))
}
pub fn write_with<T, F>(&mut self, value: &T, setup: F) -> Result<(), Error>
where
T: Serialize + ?Sized,
F: FnOnce(&mut SerializeDriver<'_>),
{
let mut driver = SerializeDriver::new(&value);
setup(&mut driver);
self.write_driver(&mut driver)
}
fn write_driver(&mut self, driver: &mut SerializeDriver<'_>) -> Result<(), Error> {
if !self.context.is_empty() {
driver.set_default_context(self.context.clone());
}
if self.serializer.in_progress() {
return Err(Error::in_progress());
}
write_output(&mut self.writer, &mut self.serializer)?;
let limit = match self.serializer.supports_partial() {
true => self.limit.max(1),
false => usize::MAX,
};
loop {
let done = self.serializer.drive_partial(driver, limit)?;
write_output(&mut self.writer, &mut self.serializer)?;
if done {
return Ok(());
}
}
}
pub fn flush(&mut self) -> Result<(), Error> {
self.writer.flush()?;
Ok(())
}
pub fn serializer(&self) -> &S {
&self.serializer
}
pub fn get_ref(&self) -> &W {
&self.writer
}
pub fn get_mut(&mut self) -> &mut W {
&mut self.writer
}
pub fn into_inner(self) -> W {
self.writer
}
pub fn into_parts(self) -> (W, S) {
(self.writer, self.serializer)
}
}
fn write_output<W: Write, S: StreamSerializer>(
writer: &mut W,
serializer: &mut S,
) -> Result<(), Error> {
let output = serializer.output();
if output.is_empty() {
return Ok(());
}
let rv = writer.write_all(output);
serializer.clear_output();
rv.map_err(Error::from)
}
impl<W: Write, S: StreamSerializer> Serializer for Writer<W, S> {
fn drive(&mut self, driver: &mut SerializeDriver<'_>) -> Result<(), Error> {
self.write_driver(driver)
}
}
pub fn from_reader<T, R, D>(reader: R, deserializer: D) -> Result<T, Error>
where
T: DeserializeOwned,
R: Read,
D: StreamDeserializer,
{
let mut reader = Reader::new(reader, deserializer);
let value = reader
.read()?
.ok_or_else(|| Error::new(ErrorKind::EndOfFile, "empty input"))?;
reader.end()?;
Ok(value)
}
pub fn to_writer<W, S, T>(writer: W, serializer: S, value: &T) -> Result<(), Error>
where
W: Write,
S: StreamSerializer,
T: Serialize + ?Sized,
{
Writer::new(writer, serializer).write(value)
}