navy-nvim-rs 0.0.18

//! Borrowed msgpack readers for a future redraw notification fast path.

use std::fmt;
use std::io::{self, ErrorKind};

use rmp::{
    Marker,
    decode::{
        self, Bytes, DecodeStringError, NumValueReadError, RmpRead, ValueReadError,
        bytes::BytesReadError,
    },
};
use rmpv::{ValueRef, decode::read_value_ref};

use super::skip::{skip_value, skip_values};

pub trait Decode<'de>: Sized {
    fn decode(buf: &mut Bytes<'de>) -> io::Result<Self>;
}

impl<'de> Decode<'de> for &'de str {
    #[inline]
    fn decode(buf: &mut Bytes<'de>) -> io::Result<Self> {
        let (value, tail) =
            decode::read_str_from_slice(buf.remaining_slice()).map_err(decode_string_error)?;
        *buf = Bytes::new(tail);
        Ok(value)
    }
}

macro_rules! impl_decode_int {
    ($($ty:ty),+) => {
        $(
            impl<'de> Decode<'de> for $ty {
                #[inline]
                fn decode(buf: &mut Bytes<'de>) -> io::Result<Self> {
                    decode::read_int::<$ty, _>(buf).map_err(num_value_read_error)
                }
            }
        )+
    };
}

impl_decode_int!(u8, u32, u64, usize, i64);

impl<'de> Decode<'de> for bool {
    #[inline]
    fn decode(buf: &mut Bytes<'de>) -> io::Result<Self> {
        decode::read_bool(buf).map_err(value_read_error)
    }
}

impl<'de> Decode<'de> for f32 {
    #[inline]
    fn decode(buf: &mut Bytes<'de>) -> io::Result<Self> {
        decode::read_f32(buf).map_err(value_read_error)
    }
}

impl<'de> Decode<'de> for f64 {
    #[inline]
    fn decode(buf: &mut Bytes<'de>) -> io::Result<Self> {
        decode::read_f64(buf).map_err(value_read_error)
    }
}

impl<'de, T> Decode<'de> for Option<T>
where
    T: Decode<'de>,
{
    #[inline]
    fn decode(buf: &mut Bytes<'de>) -> io::Result<Self> {
        let mut bytes = *buf;
        if decode::read_marker(&mut bytes).map_err(marker_read_error)? == Marker::Null {
            *buf = bytes;
            Ok(None)
        } else {
            T::decode(buf).map(Some)
        }
    }
}

impl<'de> Decode<'de> for ValueRef<'de> {
    #[inline]
    fn decode(buf: &mut Bytes<'de>) -> io::Result<Self> {
        let mut bytes = buf.remaining_slice();
        let value = read_value_ref(&mut bytes).map_err(rmpv_decode_error)?;
        *buf = Bytes::new(bytes);
        Ok(value)
    }
}

#[inline]
fn bytes_read_error(err: BytesReadError) -> io::Error {
    match err {
        BytesReadError::InsufficientBytes { .. } => ErrorKind::UnexpectedEof.into(),
        _ => ErrorKind::InvalidData.into(),
    }
}

#[inline]
fn marker_read_error(err: decode::MarkerReadError<BytesReadError>) -> io::Error {
    bytes_read_error(err.0)
}

#[inline]
fn value_read_error(err: ValueReadError<BytesReadError>) -> io::Error {
    match err {
        ValueReadError::InvalidMarkerRead(err) | ValueReadError::InvalidDataRead(err) => {
            bytes_read_error(err)
        }
        ValueReadError::TypeMismatch(_) => ErrorKind::InvalidData.into(),
    }
}

#[inline]
fn num_value_read_error(err: NumValueReadError<BytesReadError>) -> io::Error {
    match err {
        NumValueReadError::InvalidMarkerRead(err) | NumValueReadError::InvalidDataRead(err) => {
            bytes_read_error(err)
        }
        NumValueReadError::TypeMismatch(_) | NumValueReadError::OutOfRange => {
            ErrorKind::InvalidData.into()
        }
    }
}

#[inline]
fn decode_string_error(err: DecodeStringError<'_, BytesReadError>) -> io::Error {
    match err {
        DecodeStringError::InvalidMarkerRead(err) | DecodeStringError::InvalidDataRead(err) => {
            bytes_read_error(err)
        }
        DecodeStringError::BufferSizeTooSmall(_) => ErrorKind::UnexpectedEof.into(),
        DecodeStringError::TypeMismatch(_) | DecodeStringError::InvalidUtf8(_, _) => {
            ErrorKind::InvalidData.into()
        }
    }
}

#[inline]
fn rmpv_decode_error(err: rmpv::decode::Error) -> io::Error {
    match err {
        rmpv::decode::Error::InvalidMarkerRead(err) | rmpv::decode::Error::InvalidDataRead(err) => {
            err
        }
        rmpv::decode::Error::DepthLimitExceeded => ErrorKind::InvalidData.into(),
    }
}

/// A complete owned `redraw` notification frame.
pub struct RedrawFrame {
    bytes: bytes::Bytes,
    params_offset: usize,
    params_len: u32,
}

pub struct RedrawFrameInfo {
    consumed: usize,
    params_offset: usize,
    params_len: u32,
}

impl RedrawFrameInfo {
    pub fn probe(bytes: &[u8]) -> io::Result<Option<Self>> {
        let mut reader = Bytes::new(bytes);
        let outer_len = match decode::read_array_len(&mut reader) {
            Ok(len) => len,
            Err(ValueReadError::TypeMismatch(_)) => return Ok(None),
            Err(err) => return Err(value_read_error(err)),
        };

        if outer_len < 3 {
            return Ok(None);
        }

        let msg_type = match decode::read_int::<u8, _>(&mut reader) {
            Ok(msg_type) => msg_type,
            Err(NumValueReadError::TypeMismatch(_)) | Err(NumValueReadError::OutOfRange) => {
                return Ok(None);
            }
            Err(err) => {
                return Err(num_value_read_error(err));
            }
        };

        if msg_type != 2 {
            return Ok(None);
        }

        match reader.read_str_eq("redraw") {
            Ok(true) => {}
            Ok(false) => return Ok(None),
            Err(err) => return Err(err),
        }

        let params_len = match decode::read_array_len(&mut reader) {
            Ok(len) => len,
            Err(ValueReadError::TypeMismatch(_)) => {
                return Ok(None);
            }
            Err(err) => {
                return Err(value_read_error(err));
            }
        };

        let params_offset = bytes.len() - reader.remaining_slice().len();
        reader.skip_values(params_len)?;

        for _ in 3..outer_len {
            reader.skip_value()?;
        }

        Ok(Some(Self {
            consumed: bytes.len() - reader.remaining_slice().len(),
            params_offset,
            params_len,
        }))
    }

    #[inline]
    #[must_use]
    pub fn consumed(&self) -> usize {
        self.consumed
    }

    #[inline]
    pub fn frame(&self, bytes: bytes::Bytes) -> RedrawFrame {
        debug_assert_eq!(bytes.len(), self.consumed);
        RedrawFrame {
            bytes,
            params_offset: self.params_offset,
            params_len: self.params_len,
        }
    }
}

impl fmt::Debug for RedrawFrame {
    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
        fmt.debug_struct("RedrawFrame")
            .field("len", &self.bytes.len())
            .finish()
    }
}

impl RedrawFrame {
    pub fn from_slice(bytes: &[u8]) -> io::Result<Self> {
        let info = RedrawFrameInfo::probe(bytes)?.expect("redraw frame");
        Ok(info.frame(bytes::Bytes::copy_from_slice(&bytes[..info.consumed])))
    }

    pub fn from_bytes(bytes: bytes::Bytes) -> io::Result<Self> {
        let info = RedrawFrameInfo::probe(&bytes)?.expect("redraw frame");
        Ok(info.frame(bytes))
    }

    #[inline]
    pub fn consumed(&self) -> usize {
        self.bytes.len()
    }

    #[inline]
    pub fn as_bytes(&self) -> &[u8] {
        &self.bytes
    }

    #[inline]
    pub fn notification(&self) -> io::Result<RedrawNotification<'_>> {
        Ok(RedrawNotification::new(ArrayReader {
            reader: Bytes::new(&self.bytes[self.params_offset..]),
            remaining: self.params_len,
        }))
    }
}

/// The params of a `redraw` notification.
pub struct RedrawNotification<'de> {
    params: ArrayReader<'de>,
}

impl<'de> RedrawNotification<'de> {
    #[inline]
    #[must_use]
    pub(crate) fn new(params: ArrayReader<'de>) -> Self {
        Self { params }
    }

    #[inline]
    #[must_use]
    pub fn batch_count(&self) -> u32 {
        self.params.remaining()
    }

    pub fn for_each_batch<F>(&mut self, mut f: F) -> io::Result<()>
    where
        F: FnMut(&mut RedrawBatch<'de>) -> io::Result<bool>,
    {
        while !self.params.is_empty() {
            self.params.ensure_remaining()?;

            let mut batch_items = self.params.reader.read_array_reader()?;
            self.params.remaining -= 1;

            let name = batch_items.read()?;
            let args = batch_items;
            let mut batch = RedrawBatch { name, args };

            let should_continue = f(&mut batch)?;
            batch.args.skip_remaining()?;
            self.params.reader = batch.args.reader;

            if !should_continue {
                break;
            }
        }

        Ok(())
    }
}

/// One redraw event batch, e.g. `["grid_line", [...], ...]`.
pub struct RedrawBatch<'de> {
    pub name: &'de str,
    pub args: ArrayReader<'de>,
}

/// A borrowed reader over msgpack array elements.
pub struct ArrayReader<'de> {
    reader: Bytes<'de>,
    remaining: u32,
}

impl<'de> ArrayReader<'de> {
    #[inline]
    pub fn new(input: &'de [u8]) -> io::Result<Self> {
        let mut reader = Bytes::new(input);
        reader.read_array_reader()
    }

    #[inline]
    #[must_use]
    pub fn remaining(&self) -> u32 {
        self.remaining
    }

    #[inline]
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.remaining == 0
    }

    #[inline]
    pub fn read<D>(&mut self) -> io::Result<D>
    where
        D: Decode<'de>,
    {
        self.ensure_remaining()?;
        let value = D::decode(&mut self.reader)?;
        self.remaining -= 1;
        Ok(value)
    }

    #[inline]
    pub fn read_as_string(&mut self) -> io::Result<Option<String>> {
        self.ensure_remaining()?;
        let value = self.reader.read_as_string()?;
        self.remaining -= 1;
        Ok(value)
    }

    #[inline]
    pub fn read_array<T>(
        &mut self,
        f: impl FnOnce(&mut ArrayReader<'de>) -> io::Result<T>,
    ) -> io::Result<T> {
        self.ensure_remaining()?;

        let mut values = self.reader.read_array_reader()?;
        let value = f(&mut values)?;
        values.skip_remaining()?;

        self.reader = values.reader;
        self.remaining -= 1;

        Ok(value)
    }

    pub fn read_map<T>(
        &mut self,
        f: impl FnOnce(&mut MapReader<'de>) -> io::Result<T>,
    ) -> io::Result<T> {
        self.ensure_remaining()?;

        let mut entries = self.reader.read_map_reader()?;
        let value = f(&mut entries)?;
        entries.skip_remaining()?;

        self.reader = entries.reader;
        self.remaining -= 1;

        Ok(value)
    }

    #[inline]
    pub fn skip_next(&mut self) -> io::Result<()> {
        self.ensure_remaining()?;
        self.reader.skip_value()?;
        self.remaining -= 1;
        Ok(())
    }

    #[inline]
    pub fn skip_remaining(&mut self) -> io::Result<()> {
        while self.remaining > 0 {
            self.skip_next()?;
        }
        Ok(())
    }

    #[inline]
    fn ensure_remaining(&self) -> io::Result<()> {
        if self.remaining != 0 {
            Ok(())
        } else {
            Err(ErrorKind::UnexpectedEof.into())
        }
    }
}

/// A borrowed reader over msgpack map entries.
#[derive(Clone)]
pub struct MapReader<'de> {
    reader: Bytes<'de>,
    remaining: u32,
}

impl<'de> MapReader<'de> {
    #[inline]
    #[must_use]
    pub fn remaining(&self) -> u32 {
        self.remaining
    }

    #[inline]
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.remaining == 0
    }

    #[inline]
    pub fn read<K, V>(&mut self) -> io::Result<(K, V)>
    where
        K: Decode<'de>,
        V: Decode<'de>,
    {
        self.ensure_remaining()?;
        let key = K::decode(&mut self.reader)?;
        let value = V::decode(&mut self.reader)?;
        self.remaining -= 1;
        Ok((key, value))
    }

    #[inline]
    pub fn skip_next(&mut self) -> io::Result<()> {
        self.ensure_remaining()?;
        self.reader.skip_value()?;
        self.reader.skip_value()?;
        self.remaining -= 1;
        Ok(())
    }

    #[inline]
    pub fn skip_remaining(&mut self) -> io::Result<()> {
        while self.remaining > 0 {
            self.skip_next()?;
        }
        Ok(())
    }

    #[inline]
    fn ensure_remaining(&self) -> io::Result<()> {
        if self.remaining != 0 {
            Ok(())
        } else {
            Err(ErrorKind::UnexpectedEof.into())
        }
    }
}

trait BytesReaderExt<'de> {
    fn read_str_eq(&mut self, expected: &str) -> io::Result<bool>;
    fn read_array_reader(&mut self) -> io::Result<ArrayReader<'de>>;
    fn read_map_reader(&mut self) -> io::Result<MapReader<'de>>;
    fn read_as_string(&mut self) -> io::Result<Option<String>>;
    fn skip_value(&mut self) -> io::Result<()>;
    fn skip_values(&mut self, count: u32) -> io::Result<()>;
}

impl<'de> BytesReaderExt<'de> for Bytes<'de> {
    fn read_str_eq(&mut self, expected: &str) -> io::Result<bool> {
        match decode::read_str_from_slice(self.remaining_slice()) {
            Ok((value, tail)) => {
                *self = Bytes::new(tail);
                Ok(value == expected)
            }
            Err(DecodeStringError::TypeMismatch(_)) => Ok(false),
            Err(err) => Err(decode_string_error(err)),
        }
    }

    #[inline]
    fn read_array_reader(&mut self) -> io::Result<ArrayReader<'de>> {
        let remaining = decode::read_array_len(self).map_err(value_read_error)?;
        Ok(ArrayReader {
            reader: *self,
            remaining,
        })
    }

    #[inline]
    fn read_map_reader(&mut self) -> io::Result<MapReader<'de>> {
        let remaining = decode::read_map_len(self).map_err(value_read_error)?;
        Ok(MapReader {
            reader: *self,
            remaining,
        })
    }

    fn read_as_string(&mut self) -> io::Result<Option<String>> {
        let start = *self;

        match decode::read_marker(self).map_err(marker_read_error)? {
            Marker::FixPos(value) => Ok(Some(value.to_string())),
            Marker::FixNeg(value) => Ok(Some(value.to_string())),
            Marker::False => Ok(Some(false.to_string())),
            Marker::True => Ok(Some(true.to_string())),
            Marker::U8 => Ok(Some(
                self.read_data_u8().map_err(value_read_error)?.to_string(),
            )),
            Marker::U16 => Ok(Some(
                self.read_data_u16().map_err(value_read_error)?.to_string(),
            )),
            Marker::U32 => Ok(Some(
                self.read_data_u32().map_err(value_read_error)?.to_string(),
            )),
            Marker::U64 => Ok(Some(
                self.read_data_u64().map_err(value_read_error)?.to_string(),
            )),
            Marker::I8 => Ok(Some(
                self.read_data_i8().map_err(value_read_error)?.to_string(),
            )),
            Marker::I16 => Ok(Some(
                self.read_data_i16().map_err(value_read_error)?.to_string(),
            )),
            Marker::I32 => Ok(Some(
                self.read_data_i32().map_err(value_read_error)?.to_string(),
            )),
            Marker::I64 => Ok(Some(
                self.read_data_i64().map_err(value_read_error)?.to_string(),
            )),
            Marker::F32 => Ok(Some(
                self.read_data_f32().map_err(value_read_error)?.to_string(),
            )),
            Marker::F64 => Ok(Some(
                self.read_data_f64().map_err(value_read_error)?.to_string(),
            )),
            Marker::FixStr(_) | Marker::Str8 | Marker::Str16 | Marker::Str32 => {
                *self = start;
                let value: &str = Decode::decode(self)?;
                Ok(Some(value.to_owned()))
            }
            _ => {
                *self = start;
                self.skip_value()?;
                Ok(None)
            }
        }
    }

    #[inline]
    fn skip_value(&mut self) -> io::Result<()> {
        let remaining = self.remaining_slice();
        let consumed = skip_value(remaining)?;
        *self = Bytes::new(&remaining[consumed..]);
        Ok(())
    }

    #[inline]
    fn skip_values(&mut self, count: u32) -> io::Result<()> {
        let remaining = self.remaining_slice();
        let consumed = skip_values(remaining, count)?;
        *self = Bytes::new(&remaining[consumed..]);
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use std::assert_matches;

    use super::*;
    use rmpv::encode::write_value_ref;

    fn encode_value(value: ValueRef<'_>) -> Vec<u8> {
        let mut bytes = vec![];
        write_value_ref(&mut bytes, &value).unwrap();
        bytes
    }

    fn redraw_notification(params: Vec<ValueRef<'_>>) -> Vec<u8> {
        rpc_message(vec![2_u64.into(), "redraw".into(), ValueRef::Array(params)])
    }

    fn rpc_message(fields: Vec<ValueRef<'_>>) -> Vec<u8> {
        encode_value(ValueRef::Array(fields))
    }

    fn read<'de, D>(reader: &mut Bytes<'de>) -> io::Result<D>
    where
        D: Decode<'de>,
    {
        D::decode(reader)
    }

    fn read_redraw_notification(bytes: &[u8]) -> RedrawNotification<'_> {
        let mut reader = Bytes::new(bytes);
        decode::read_array_len(&mut reader).unwrap();
        reader.skip_value().unwrap();
        assert_eq!(read::<&str>(&mut reader).unwrap(), "redraw");
        RedrawNotification::new(reader.read_array_reader().unwrap())
    }

    #[track_caller]
    fn assert_reader_error_kind<T>(result: io::Result<T>, kind: ErrorKind) {
        let err = result.err().expect("expected reader error");
        assert_eq!(err.kind(), kind);
    }

    #[track_caller]
    fn assert_incomplete<T>(result: io::Result<T>) {
        assert_reader_error_kind(result, ErrorKind::UnexpectedEof);
    }

    #[track_caller]
    fn skip_ok(bytes: Vec<u8>) {
        let mut reader = Bytes::new(&bytes);
        reader.skip_value().unwrap();
        assert!(reader.remaining_slice().is_empty());
    }

    fn push_u16(bytes: &mut Vec<u8>, value: u16) {
        bytes.extend_from_slice(&value.to_be_bytes());
    }

    fn push_u32(bytes: &mut Vec<u8>, value: u32) {
        bytes.extend_from_slice(&value.to_be_bytes());
    }

    fn fixed_payload(marker: Marker, len: usize) -> Vec<u8> {
        let mut bytes = vec![marker.to_u8(), 0];
        bytes.extend(std::iter::repeat_n(1, len));
        bytes
    }

    fn len8_payload(marker: Marker, len: u8, includes_ext_type: bool) -> Vec<u8> {
        let mut bytes = vec![marker.to_u8(), len];
        if includes_ext_type {
            bytes.push(0);
        }
        bytes.extend(std::iter::repeat_n(1, len as usize));
        bytes
    }

    fn len16_payload(marker: Marker, len: u16, includes_ext_type: bool) -> Vec<u8> {
        let mut bytes = vec![marker.to_u8()];
        push_u16(&mut bytes, len);
        if includes_ext_type {
            bytes.push(0);
        }
        bytes.extend(std::iter::repeat_n(1, len as usize));
        bytes
    }

    fn len32_payload(marker: Marker, len: u32, includes_ext_type: bool) -> Vec<u8> {
        let mut bytes = vec![marker.to_u8()];
        push_u32(&mut bytes, len);
        if includes_ext_type {
            bytes.push(0);
        }
        bytes.extend(std::iter::repeat_n(1, len as usize));
        bytes
    }

    #[test]
    fn redraw_frame_probe_rejects_non_redraw_messages() {
        assert!(
            RedrawFrameInfo::probe(&encode_value("redraw".into()))
                .unwrap()
                .is_none()
        );
        assert!(
            RedrawFrameInfo::probe(&rpc_message(vec![]))
                .unwrap()
                .is_none()
        );
        assert!(
            RedrawFrameInfo::probe(&rpc_message(vec![2_u64.into()]))
                .unwrap()
                .is_none()
        );

        let request = rpc_message(vec![
            0_u64.into(),
            7_u64.into(),
            "redraw".into(),
            ValueRef::Array(vec![]),
        ]);
        assert!(RedrawFrameInfo::probe(&request).unwrap().is_none());

        let non_integer_message_type = rpc_message(vec![
            "notification".into(),
            "redraw".into(),
            ValueRef::Array(vec![]),
        ]);
        assert!(
            RedrawFrameInfo::probe(&non_integer_message_type)
                .unwrap()
                .is_none()
        );

        let response = rpc_message(vec![
            1_u64.into(),
            7_u64.into(),
            ValueRef::Nil,
            ValueRef::Boolean(true),
        ]);
        assert!(RedrawFrameInfo::probe(&response).unwrap().is_none());

        let non_redraw = rpc_message(vec![
            2_u64.into(),
            "not-redraw".into(),
            ValueRef::Array(vec![]),
        ]);
        assert!(RedrawFrameInfo::probe(&non_redraw).unwrap().is_none());

        let non_string_method =
            rpc_message(vec![2_u64.into(), 7_u64.into(), ValueRef::Array(vec![])]);
        assert!(
            RedrawFrameInfo::probe(&non_string_method)
                .unwrap()
                .is_none()
        );

        let method_only = rpc_message(vec![2_u64.into(), "redraw".into()]);
        assert!(RedrawFrameInfo::probe(&method_only).unwrap().is_none());

        let non_array_params = rpc_message(vec![2_u64.into(), "redraw".into(), "not-array".into()]);
        assert!(RedrawFrameInfo::probe(&non_array_params).unwrap().is_none());
    }

    #[test]
    fn redraw_frame_probe_reports_complete_incomplete_and_not_redraw() {
        let redraw = redraw_notification(vec![ValueRef::Array(vec!["flush".into()])]);
        let incomplete_redraw_prefix = [
            Marker::FixArray(3).to_u8(),
            2,
            Marker::FixStr(6).to_u8(),
            b'r',
            b'e',
            b'd',
            b'r',
            b'a',
            b'w',
        ];
        let request = rpc_message(vec![
            0_u64.into(),
            1_u64.into(),
            "method".into(),
            ValueRef::Array(vec![]),
        ]);

        let info = RedrawFrameInfo::probe(&redraw)
            .unwrap()
            .expect("redraw frame");
        assert_eq!(info.consumed(), redraw.len());
        let frame = RedrawFrame::from_slice(&redraw).unwrap();
        assert_eq!(frame.as_bytes(), redraw.as_slice());
        assert_incomplete(RedrawFrameInfo::probe(&[]));
        assert_incomplete(RedrawFrameInfo::probe(&incomplete_redraw_prefix));
        assert!(RedrawFrameInfo::probe(&request).unwrap().is_none());
    }

    #[test]
    fn redraw_frame_probe_counts_extra_outer_fields() {
        let redraw = rpc_message(vec![
            2_u64.into(),
            "redraw".into(),
            ValueRef::Array(vec![ValueRef::Array(vec!["flush".into()])]),
            "extra".into(),
        ]);
        let mut bytes = redraw.clone();
        bytes.extend_from_slice(&encode_value("tail".into()));

        let info = RedrawFrameInfo::probe(&bytes)
            .unwrap()
            .expect("redraw frame");
        assert_eq!(info.consumed(), redraw.len());
    }

    #[test]
    fn redraw_frame_notification_reads_params_from_probe_offset() {
        let redraw = redraw_notification(vec![
            ValueRef::Array(vec![
                "grid_resize".into(),
                ValueRef::Array(vec![1_u64.into(), 80_u64.into(), 24_u64.into()]),
            ]),
            ValueRef::Array(vec!["flush".into()]),
        ]);
        let frame = RedrawFrame::from_slice(&redraw).unwrap();
        let mut notification = frame.notification().unwrap();
        let mut seen = vec![];

        assert_eq!(notification.batch_count(), 2);
        notification
            .for_each_batch(|batch| {
                seen.push(batch.name.to_owned());

                if batch.name == "grid_resize" {
                    batch.args.read_array(|args| {
                        assert_eq!(args.read::<u64>()?, 1);
                        assert_eq!(args.read::<u64>()?, 80);
                        assert_eq!(args.read::<u64>()?, 24);
                        Ok(true)
                    })?;
                } else {
                    assert!(batch.args.is_empty());
                }

                Ok(true)
            })
            .unwrap();

        assert_eq!(seen, vec!["grid_resize", "flush"]);
    }

    #[test]
    fn redraw_frame_probe_reports_malformed_method_payloads() {
        let invalid_utf8_method = vec![
            Marker::FixArray(3).to_u8(),
            2,
            Marker::Str8.to_u8(),
            1,
            0xff,
            Marker::FixArray(0).to_u8(),
        ];
        assert_reader_error_kind(
            RedrawFrameInfo::probe(&invalid_utf8_method),
            ErrorKind::InvalidData,
        );

        let incomplete_method = vec![
            Marker::FixArray(3).to_u8(),
            2,
            Marker::Str8.to_u8(),
            2,
            b'a',
        ];
        assert_incomplete(RedrawFrameInfo::probe(&incomplete_method));

        let incomplete_first_item = vec![Marker::FixArray(3).to_u8(), Marker::U16.to_u8(), 1];
        assert_incomplete(RedrawFrameInfo::probe(&incomplete_first_item));
    }

    #[test]
    fn redraw_notification_reads_batches_and_args() {
        let bytes = redraw_notification(vec![
            ValueRef::Array(vec![
                "grid_line".into(),
                ValueRef::Array(vec![1_u64.into(), (-2_i64).into(), ValueRef::Boolean(true)]),
            ]),
            ValueRef::Array(vec!["flush".into()]),
        ]);
        let mut redraw = read_redraw_notification(&bytes);
        let mut seen = vec![];
        assert_eq!(redraw.batch_count(), 2);

        redraw
            .for_each_batch(|batch| {
                seen.push(batch.name.to_owned());

                if batch.name == "grid_line" {
                    assert_eq!(batch.args.remaining(), 1);
                    while !batch.args.is_empty() {
                        batch.args.read_array(|args| {
                            assert_eq!(args.read::<u64>()?, 1);
                            assert_eq!(args.read::<i64>()?, -2);
                            assert!(args.read::<bool>()?);
                            Ok(())
                        })?;
                    }
                } else {
                    assert!(batch.args.is_empty());
                }

                Ok(true)
            })
            .unwrap();

        assert_eq!(seen, vec!["grid_line", "flush"]);
    }

    #[test]
    fn redraw_notification_reports_invalid_batch_name() {
        let bytes = redraw_notification(vec![ValueRef::Array(vec![1_u64.into()])]);
        let mut redraw = read_redraw_notification(&bytes);

        assert_reader_error_kind(
            redraw.for_each_batch(|_| unreachable!()),
            ErrorKind::InvalidData,
        );
    }

    #[test]
    fn redraw_batch_skips_unread_args() {
        let bytes = redraw_notification(vec![
            ValueRef::Array(vec![
                "grid_line".into(),
                ValueRef::Array(vec![1_u64.into()]),
                ValueRef::Array(vec![2_u64.into()]),
            ]),
            ValueRef::Array(vec!["flush".into()]),
        ]);
        let mut redraw = read_redraw_notification(&bytes);
        let mut names = vec![];

        redraw
            .for_each_batch(|batch| {
                names.push(batch.name.to_owned());
                Ok(true)
            })
            .unwrap();

        assert_eq!(names, vec!["grid_line", "flush"]);
    }

    #[test]
    fn redraw_notification_stops_when_batch_callback_returns_false() {
        let bytes = redraw_notification(vec![
            ValueRef::Array(vec![
                "grid_line".into(),
                ValueRef::Array(vec![1_u64.into()]),
            ]),
            ValueRef::Array(vec!["flush".into()]),
        ]);
        let mut redraw = read_redraw_notification(&bytes);
        let mut names = vec![];

        redraw
            .for_each_batch(|batch| {
                names.push(batch.name.to_owned());
                Ok(false)
            })
            .unwrap();

        assert_eq!(names, vec!["grid_line"]);
        assert_eq!(redraw.batch_count(), 1);

        redraw
            .for_each_batch(|batch| {
                names.push(batch.name.to_owned());
                Ok(true)
            })
            .unwrap();

        assert_eq!(names, vec!["grid_line", "flush"]);
        assert_eq!(redraw.batch_count(), 0);
    }

    #[test]
    fn array_reader_reads_float_values() {
        let f32_value = 1.25_f32;
        let f64_value = -2.5_f64;
        let bytes = encode_value(ValueRef::Array(vec![f32_value.into(), f64_value.into()]));

        let mut reader = Bytes::new(&bytes);
        let mut array = reader.read_array_reader().unwrap();
        assert_eq!(array.read::<f32>().unwrap(), f32_value);
        assert_eq!(array.read::<f64>().unwrap(), f64_value);
        assert!(array.is_empty());
    }

    #[test]
    fn array_reader_new_reads_array_payload() {
        let bytes = encode_value(ValueRef::Array(vec![24_u64.into(), "text".into()]));
        let mut array = ArrayReader::new(&bytes).unwrap();

        assert_eq!(array.remaining(), 2);
        assert_eq!(array.read::<u64>().unwrap(), 24);
        assert_eq!(array.read::<&str>().unwrap(), "text");
        assert!(array.is_empty());
    }

    #[test]
    fn array_reader_reads_u32_values() {
        let bytes = encode_value(ValueRef::Array(vec![
            24_u32.into(),
            256_u32.into(),
            65_536_u32.into(),
            2_i8.into(),
        ]));

        let mut array = ArrayReader::new(&bytes).unwrap();

        assert_eq!(array.read::<u32>().unwrap(), 24);
        assert_eq!(array.read::<u32>().unwrap(), 256);
        assert_eq!(array.read::<u32>().unwrap(), 65_536);
        assert_eq!(array.read::<u32>().unwrap(), 2);
        assert!(array.is_empty());
        assert_incomplete(array.read::<u32>());
    }

    #[test]
    fn array_reader_reports_u32_errors() {
        let bytes = encode_value(ValueRef::Array(vec!["not-u32".into()]));
        let mut array = ArrayReader::new(&bytes).unwrap();

        assert_reader_error_kind(array.read::<u32>(), ErrorKind::InvalidData);

        let bytes = encode_value(ValueRef::Array(vec![(-1_i64).into()]));
        let mut array = ArrayReader::new(&bytes).unwrap();

        assert_reader_error_kind(array.read::<u32>(), ErrorKind::InvalidData);

        let bytes = encode_value(ValueRef::Array(vec![(u64::from(u32::MAX) + 1).into()]));
        let mut array = ArrayReader::new(&bytes).unwrap();

        assert_reader_error_kind(array.read::<u32>(), ErrorKind::InvalidData);

        let bytes = vec![Marker::FixArray(1).to_u8(), Marker::U16.to_u8()];
        let mut array = ArrayReader::new(&bytes).unwrap();

        assert_incomplete(array.read::<u32>());
    }

    #[test]
    fn array_reader_reads_usize_values() {
        let bytes = encode_value(ValueRef::Array(vec![
            24_usize.into(),
            256_usize.into(),
            65_536_usize.into(),
            2_i8.into(),
        ]));

        let mut array = ArrayReader::new(&bytes).unwrap();

        assert_eq!(array.read::<usize>().unwrap(), 24);
        assert_eq!(array.read::<usize>().unwrap(), 256);
        assert_eq!(array.read::<usize>().unwrap(), 65_536);
        assert_eq!(array.read::<usize>().unwrap(), 2);
        assert!(array.is_empty());
        assert_incomplete(array.read::<usize>());
    }

    #[test]
    fn array_reader_reports_usize_errors() {
        let bytes = encode_value(ValueRef::Array(vec!["not-usize".into()]));
        let mut array = ArrayReader::new(&bytes).unwrap();

        assert_reader_error_kind(array.read::<usize>(), ErrorKind::InvalidData);

        let bytes = encode_value(ValueRef::Array(vec![(-1_i64).into()]));
        let mut array = ArrayReader::new(&bytes).unwrap();

        assert_reader_error_kind(array.read::<usize>(), ErrorKind::InvalidData);

        let bytes = vec![Marker::FixArray(1).to_u8(), Marker::U16.to_u8()];
        let mut array = ArrayReader::new(&bytes).unwrap();

        assert_incomplete(array.read::<usize>());
    }

    #[test]
    fn array_reader_reads_u32_or_nil_values() {
        let bytes = encode_value(ValueRef::Array(vec![
            24_u32.into(),
            ValueRef::Nil,
            256_u32.into(),
            u32::MAX.into(),
            2_i8.into(),
        ]));

        let mut array = ArrayReader::new(&bytes).unwrap();

        assert_eq!(array.read::<Option<u32>>().unwrap(), Some(24));
        assert_eq!(array.read::<Option<u32>>().unwrap(), None);
        assert_eq!(array.read::<Option<u32>>().unwrap(), Some(256));
        assert_eq!(array.read::<Option<u32>>().unwrap(), Some(u32::MAX));
        assert_eq!(array.read::<Option<u32>>().unwrap(), Some(2));
        assert!(array.is_empty());
        assert_incomplete(array.read::<Option<u32>>());
    }

    #[test]
    fn array_reader_reports_u32_or_nil_errors() {
        let bytes = encode_value(ValueRef::Array(vec!["not-u32".into()]));
        let mut array = ArrayReader::new(&bytes).unwrap();

        assert_reader_error_kind(array.read::<Option<u32>>(), ErrorKind::InvalidData);

        let bytes = encode_value(ValueRef::Array(vec![(-1_i64).into()]));
        let mut array = ArrayReader::new(&bytes).unwrap();

        assert_reader_error_kind(array.read::<Option<u32>>(), ErrorKind::InvalidData);

        let bytes = encode_value(ValueRef::Array(vec![(u64::from(u32::MAX) + 1).into()]));
        let mut array = ArrayReader::new(&bytes).unwrap();

        assert_reader_error_kind(array.read::<Option<u32>>(), ErrorKind::InvalidData);

        let bytes = vec![Marker::FixArray(1).to_u8(), Marker::U16.to_u8()];
        let mut array = ArrayReader::new(&bytes).unwrap();

        assert_incomplete(array.read::<Option<u32>>());
    }

    #[test]
    fn array_reader_stringifies_scalar_values() {
        let bytes = encode_value(ValueRef::Array(vec![
            24_u64.into(),
            (-1_i64).into(),
            255_u64.into(),
            256_u64.into(),
            65_536_u64.into(),
            9_007_199_254_740_991_u64.into(),
            (-2_i64).into(),
            (-257_i64).into(),
            (-65_537_i64).into(),
            (-2_147_483_649_i64).into(),
            ValueRef::Boolean(true),
            ValueRef::Boolean(false),
            1.25_f32.into(),
            (-2.5_f64).into(),
            "text".into(),
            ValueRef::Nil,
            ValueRef::Array(vec![1_u64.into()]),
            ValueRef::Map(vec![(ValueRef::Nil, ValueRef::Boolean(true))]),
        ]));

        let mut reader = Bytes::new(&bytes);
        let mut array = reader.read_array_reader().unwrap();

        assert_eq!(array.read_as_string().unwrap().as_deref(), Some("24"));
        assert_eq!(array.read_as_string().unwrap().as_deref(), Some("-1"));
        assert_eq!(array.read_as_string().unwrap().as_deref(), Some("255"));
        assert_eq!(array.read_as_string().unwrap().as_deref(), Some("256"));
        assert_eq!(array.read_as_string().unwrap().as_deref(), Some("65536"));
        assert_eq!(
            array.read_as_string().unwrap().as_deref(),
            Some("9007199254740991")
        );
        assert_eq!(array.read_as_string().unwrap().as_deref(), Some("-2"));
        assert_eq!(array.read_as_string().unwrap().as_deref(), Some("-257"));
        assert_eq!(array.read_as_string().unwrap().as_deref(), Some("-65537"));
        assert_eq!(
            array.read_as_string().unwrap().as_deref(),
            Some("-2147483649")
        );
        assert_eq!(array.read_as_string().unwrap().as_deref(), Some("true"));
        assert_eq!(array.read_as_string().unwrap().as_deref(), Some("false"));
        assert_eq!(array.read_as_string().unwrap().as_deref(), Some("1.25"));
        assert_eq!(array.read_as_string().unwrap().as_deref(), Some("-2.5"));
        assert_eq!(array.read_as_string().unwrap().as_deref(), Some("text"));
        assert_eq!(array.read_as_string().unwrap(), None);
        assert_eq!(array.read_as_string().unwrap(), None);
        assert_eq!(array.read_as_string().unwrap(), None);
        assert!(array.is_empty());
        assert_incomplete(array.read_as_string());
    }

    #[test]
    fn array_reader_reads_value_refs() {
        let bytes = encode_value(ValueRef::Array(vec![
            "text".into(),
            ValueRef::Array(vec![1_u64.into(), ValueRef::Boolean(true)]),
            ValueRef::Map(vec![("key".into(), "value".into())]),
        ]));
        let mut array = ArrayReader::new(&bytes).unwrap();

        assert_matches!(
            array.read::<ValueRef<'_>>().unwrap(),
            ValueRef::String(value) if value.as_str() == Some("text")
        );
        assert_matches!(
            array.read::<ValueRef<'_>>().unwrap(),
            ValueRef::Array(values) if values.len() == 2
        );
        assert_matches!(
            array.read::<ValueRef<'_>>().unwrap(),
            ValueRef::Map(entries) if entries.len() == 1
        );
        assert_incomplete(array.read::<ValueRef<'_>>());
    }

    #[test]
    fn array_reader_reads_map_values() {
        let bytes = encode_value(ValueRef::Array(vec![ValueRef::Map(vec![(
            "k".into(),
            1_u64.into(),
        )])]));
        let mut array = ArrayReader::new(&bytes).unwrap();

        assert_eq!(array.read_map(|map| Ok(map.remaining())).unwrap(), 1);
        assert!(array.is_empty());
    }

    #[test]
    fn map_reader_reads_string_key_value_ref_pairs() {
        let bytes = encode_value(ValueRef::Map(vec![
            ("k1".into(), 1_u64.into()),
            ("k2".into(), ValueRef::Boolean(true)),
            ("k3".into(), "v".into()),
        ]));
        let mut reader = Bytes::new(&bytes);
        let mut map = reader.read_map_reader().unwrap();

        let (key, value): (&str, ValueRef<'_>) = map.read().unwrap();
        assert_eq!(key, "k1");
        assert_eq!(value.as_u64(), Some(1));

        let (key, value): (&str, ValueRef<'_>) = map.read().unwrap();
        assert_eq!(key, "k2");
        assert_matches!(value, ValueRef::Boolean(true));

        let (key, value): (&str, ValueRef<'_>) = map.read().unwrap();
        assert_eq!(key, "k3");
        assert_matches!(value, ValueRef::String(value) if value.as_str() == Some("v"));

        assert!(map.is_empty());
        assert_reader_error_kind(map.read::<&str, ValueRef<'_>>(), ErrorKind::UnexpectedEof);
    }

    #[test]
    fn map_reader_read_pair_reports_non_string_key() {
        let bytes = encode_value(ValueRef::Map(vec![(1_u64.into(), ValueRef::Boolean(true))]));
        let mut reader = Bytes::new(&bytes);
        let mut map = reader.read_map_reader().unwrap();

        assert_reader_error_kind(map.read::<&str, ValueRef<'_>>(), ErrorKind::InvalidData);
    }

    #[test]
    fn array_reader_reads_values_and_reports_boundaries() {
        let bytes = encode_value(ValueRef::Array(vec!["value".into()]));
        let mut reader = Bytes::new(&bytes);
        let mut array = reader.read_array_reader().unwrap();

        assert_eq!(array.read::<&str>().unwrap(), "value");
        assert_reader_error_kind(array.read::<&str>(), ErrorKind::UnexpectedEof);
    }

    #[test]
    fn array_reader_reports_type_errors() {
        let bytes = encode_value(ValueRef::Array(vec!["not-u64".into()]));
        let mut array = ArrayReader::new(&bytes).unwrap();
        assert_reader_error_kind(array.read::<u64>(), ErrorKind::InvalidData);

        let bytes = encode_value(ValueRef::Array(vec!["not-i64".into()]));
        let mut array = ArrayReader::new(&bytes).unwrap();
        assert_reader_error_kind(array.read::<i64>(), ErrorKind::InvalidData);

        let bytes = encode_value(ValueRef::Array(vec![1_u64.into()]));
        let mut array = ArrayReader::new(&bytes).unwrap();
        assert_reader_error_kind(array.read::<bool>(), ErrorKind::InvalidData);

        let bytes = encode_value(ValueRef::Array(vec![1_u64.into()]));
        let mut array = ArrayReader::new(&bytes).unwrap();
        assert_reader_error_kind(array.read_array(|_| Ok(())), ErrorKind::InvalidData);

        let bytes = encode_value(ValueRef::Array(vec![1_u64.into()]));
        let mut array = ArrayReader::new(&bytes).unwrap();
        assert_reader_error_kind(array.read_map(|_| Ok(())), ErrorKind::InvalidData);
    }

    #[test]
    fn map_reader_skips_entries() {
        let bytes = encode_value(ValueRef::Map(vec![
            ("k1".into(), 1_u64.into()),
            ("k2".into(), 2_u64.into()),
        ]));
        let mut reader = Bytes::new(&bytes);
        let mut map = reader.read_map_reader().unwrap();

        assert_eq!(map.remaining(), 2);
        assert!(!map.is_empty());

        map.skip_next().unwrap();
        assert_eq!(map.remaining(), 1);

        let (key, value): (&str, ValueRef<'_>) = map.read().unwrap();
        assert_eq!(key, "k2");
        assert_eq!(value.as_u64(), Some(2));

        assert!(map.is_empty());
        assert_reader_error_kind(map.skip_next(), ErrorKind::UnexpectedEof);
    }

    #[test]
    fn map_reader_skips_remaining_entries() {
        let bytes = encode_value(ValueRef::Map(vec![("k".into(), "v".into())]));
        let mut reader = Bytes::new(&bytes);
        let mut map = reader.read_map_reader().unwrap();

        map.skip_remaining().unwrap();
        assert!(map.is_empty());
    }

    #[test]
    fn bytes_reader_skips_all_payload_marker_families() {
        skip_ok(vec![Marker::FixPos(1).to_u8()]);
        skip_ok(vec![Marker::FixNeg(-1).to_u8()]);
        skip_ok(vec![Marker::Null.to_u8()]);
        skip_ok(vec![Marker::False.to_u8()]);
        skip_ok(vec![Marker::True.to_u8()]);
        skip_ok(vec![Marker::FixStr(1).to_u8(), b'a']);
        skip_ok(vec![Marker::FixArray(1).to_u8(), Marker::Null.to_u8()]);
        skip_ok(vec![
            Marker::FixMap(1).to_u8(),
            Marker::Null.to_u8(),
            Marker::True.to_u8(),
        ]);

        skip_ok(len8_payload(Marker::Bin8, 1, false));
        skip_ok(len16_payload(Marker::Bin16, 1, false));
        skip_ok(len32_payload(Marker::Bin32, 1, false));
        skip_ok(len8_payload(Marker::Ext8, 1, true));
        skip_ok(len16_payload(Marker::Ext16, 1, true));
        skip_ok(len32_payload(Marker::Ext32, 1, true));
        skip_ok(vec![Marker::F32.to_u8(), 0, 0, 0, 0]);
        skip_ok(vec![Marker::F64.to_u8(), 0, 0, 0, 0, 0, 0, 0, 0]);
        skip_ok(vec![Marker::U8.to_u8(), 1]);
        skip_ok(vec![Marker::I8.to_u8(), 1]);
        skip_ok(vec![Marker::U16.to_u8(), 0, 1]);
        skip_ok(vec![Marker::I16.to_u8(), 0, 1]);
        skip_ok(vec![Marker::U32.to_u8(), 0, 0, 0, 1]);
        skip_ok(vec![Marker::I32.to_u8(), 0, 0, 0, 1]);
        skip_ok(vec![Marker::U64.to_u8(), 0, 0, 0, 0, 0, 0, 0, 1]);
        skip_ok(vec![Marker::I64.to_u8(), 0, 0, 0, 0, 0, 0, 0, 1]);
        skip_ok(fixed_payload(Marker::FixExt1, 1));
        skip_ok(fixed_payload(Marker::FixExt2, 2));
        skip_ok(fixed_payload(Marker::FixExt4, 4));
        skip_ok(fixed_payload(Marker::FixExt8, 8));
        skip_ok(fixed_payload(Marker::FixExt16, 16));
        skip_ok(len8_payload(Marker::Str8, 1, false));
        skip_ok(len16_payload(Marker::Str16, 1, false));
        skip_ok(len32_payload(Marker::Str32, 1, false));

        let mut array16 = vec![Marker::Array16.to_u8()];
        push_u16(&mut array16, 1);
        array16.push(Marker::Null.to_u8());
        skip_ok(array16);

        let mut array32 = vec![Marker::Array32.to_u8()];
        push_u32(&mut array32, 1);
        array32.push(Marker::Null.to_u8());
        skip_ok(array32);

        let mut map16 = vec![Marker::Map16.to_u8()];
        push_u16(&mut map16, 1);
        map16.push(Marker::Null.to_u8());
        map16.push(Marker::True.to_u8());
        skip_ok(map16);

        let mut map32 = vec![Marker::Map32.to_u8()];
        push_u32(&mut map32, 1);
        map32.push(Marker::Null.to_u8());
        map32.push(Marker::True.to_u8());
        skip_ok(map32);
    }

    #[test]
    fn bytes_reader_reports_skip_errors() {
        let reserved = [Marker::Reserved.to_u8()];
        let mut reader = Bytes::new(&reserved);
        assert_reader_error_kind(reader.skip_value(), ErrorKind::InvalidData);

        let mut reader = Bytes::new(&[]);
        assert_reader_error_kind(reader.skip_value(), ErrorKind::UnexpectedEof);

        let incomplete_bin = [Marker::Bin8.to_u8()];
        let mut reader = Bytes::new(&incomplete_bin);
        assert_reader_error_kind(reader.skip_value(), ErrorKind::UnexpectedEof);

        let incomplete_fixstr = [Marker::FixStr(2).to_u8(), b'a'];
        let mut reader = Bytes::new(&incomplete_fixstr);
        assert_reader_error_kind(reader.skip_value(), ErrorKind::UnexpectedEof);
    }

    #[test]
    fn bytes_reader_reports_integer_range_errors() {
        let bytes = [
            Marker::U64.to_u8(),
            0xff,
            0xff,
            0xff,
            0xff,
            0xff,
            0xff,
            0xff,
            0xff,
        ];
        let mut reader = Bytes::new(&bytes);

        assert_reader_error_kind(read::<i64>(&mut reader), ErrorKind::InvalidData);
    }

    #[test]
    fn bytes_reader_reports_truncated_reads() {
        let mut reader = Bytes::new(&[]);
        assert_reader_error_kind(read::<bool>(&mut reader), ErrorKind::UnexpectedEof);

        let mut reader = Bytes::new(&[]);
        assert_reader_error_kind(read::<i64>(&mut reader), ErrorKind::UnexpectedEof);

        let truncated_u64 = [Marker::U64.to_u8(), 0];
        let mut reader = Bytes::new(&truncated_u64);
        assert_reader_error_kind(read::<i64>(&mut reader), ErrorKind::UnexpectedEof);

        let truncated_array_len = [Marker::Array16.to_u8()];
        let mut reader = Bytes::new(&truncated_array_len);
        assert_reader_error_kind(
            reader.read_array_reader().map(|_| ()),
            ErrorKind::UnexpectedEof,
        );

        let truncated_map_len = [Marker::Map16.to_u8()];
        let mut reader = Bytes::new(&truncated_map_len);
        assert_reader_error_kind(
            reader.read_map_reader().map(|_| ()),
            ErrorKind::UnexpectedEof,
        );

        let truncated_value = [Marker::Str8.to_u8(), 2, b'a'];
        let mut reader = Bytes::new(&truncated_value);
        assert_reader_error_kind(read::<ValueRef<'_>>(&mut reader), ErrorKind::UnexpectedEof);

        let mut reader = Bytes::new(&[]);
        assert_reader_error_kind(read::<&str>(&mut reader), ErrorKind::UnexpectedEof);
    }
}