navy-nvim-rs 0.0.20

//! Msgpack readers and decoding msgpack-rpc responses from Neovim.
use std::io::{self, ErrorKind};

use bytes::{Buf as _, Bytes as OwnedBytes, BytesMut};
use rmp::{
    Marker,
    decode::{
        self, Bytes, DecodeStringError, NumValueReadError, RmpRead, ValueReadError,
        bytes::BytesReadError,
    },
};
use rmpv::{
    Value, ValueRef,
    decode::{read_value, read_value_ref},
};
use tokio::io::{AsyncRead, AsyncReadExt};

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)
    }
}

impl<'de> Decode<'de> for Value {
    #[inline]
    fn decode(buf: &mut Bytes<'de>) -> io::Result<Self> {
        let mut bytes = buf.remaining_slice();
        let value = read_value(&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]
pub(super) 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]
pub(super) 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 borrowed reader over msgpack array elements.
#[derive(Debug)]
pub struct ArrayReader<'de> {
    pub(super) reader: Bytes<'de>,
    pub(super) 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]
    pub(crate) fn require_len(&self, expected: u32) -> io::Result<()> {
        if self.remaining < expected {
            Err(ErrorKind::InvalidData.into())
        } else {
            Ok(())
        }
    }

    #[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]
    pub(crate) fn finish(mut self) -> io::Result<&'de [u8]> {
        self.skip_remaining()?;
        Ok(self.reader.remaining_slice())
    }

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

/// A borrowed reader over msgpack map entries.
#[derive(Debug)]
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())
        }
    }
}

pub(super) 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(())
    }
}

const DECODE_READ_BUFFER_SIZE: usize = 64 * 1024;
const MSG_TYPE_REQUEST: u8 = 0;
const MSG_TYPE_RESPONSE: u8 = 1;
const MSG_TYPE_NOTIFICATION: u8 = 2;

#[derive(Debug, PartialEq, Clone)]
pub struct RpcResponse {
    pub msgid: u32,
    pub error: Value,
    pub result: Value,
}

#[derive(Debug)]
pub enum IncomingMessage<'a> {
    Response(RpcResponse),
    Request { msgid: u32, method: &'a str },
    Notification { method: &'a str },
}

impl<'a> IncomingMessage<'a> {
    fn decode(reader: &mut &'a [u8]) -> io::Result<Self> {
        let mut envelope = ArrayReader::new(*reader)?;
        if envelope.is_empty() {
            return Err(ErrorKind::InvalidData.into());
        }

        let msgtyp = envelope.read::<u8>()?;

        match msgtyp {
            MSG_TYPE_REQUEST => {
                envelope.require_len(3)?;
                let msgid = envelope.read::<u32>()?;
                let method = envelope.read::<&str>()?;
                *reader = envelope.finish()?;
                Ok(Self::Request { msgid, method })
            }
            MSG_TYPE_NOTIFICATION => {
                envelope.require_len(2)?;
                let method = envelope.read::<&str>()?;
                *reader = envelope.finish()?;
                Ok(Self::Notification { method })
            }
            MSG_TYPE_RESPONSE => {
                envelope.require_len(3)?;
                let msgid = envelope.read::<u32>()?;
                let error = envelope.read::<Value>()?;
                let result = envelope.read::<Value>()?;
                *reader = envelope.finish()?;

                let response = RpcResponse {
                    msgid,
                    error,
                    result,
                };

                Ok(Self::Response(response))
            }
            _ => Err(ErrorKind::InvalidData.into()),
        }
    }
}

#[derive(Debug)]
pub struct DecodedMessage<'a> {
    pub message: IncomingMessage<'a>,
    pub consumed: usize,
}

/// State reused while decoding msgpack-rpc messages from a stream.
pub struct Decoder {
    rest: BytesMut,
    read_buf: Box<[u8; DECODE_READ_BUFFER_SIZE]>,
}

impl Default for Decoder {
    fn default() -> Self {
        Self::new()
    }
}

impl Decoder {
    #[must_use]
    #[inline]
    pub fn new() -> Self {
        Self {
            rest: BytesMut::new(),
            read_buf: Box::new([0; DECODE_READ_BUFFER_SIZE]),
        }
    }

    #[inline]
    pub fn has_rest(&self) -> bool {
        !self.rest.is_empty()
    }

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

    pub fn try_decode_message(&self) -> io::Result<Option<DecodedMessage<'_>>> {
        let mut input = &self.rest[..];
        let available_len = input.len();

        match IncomingMessage::decode(&mut input) {
            Ok(message) => Ok(Some(DecodedMessage {
                message,
                consumed: available_len - input.len(),
            })),
            Err(err) if err.kind() == ErrorKind::UnexpectedEof => Ok(None),
            Err(err) => Err(err),
        }
    }

    pub fn consume(&mut self, consumed: usize) {
        debug_assert!(consumed <= self.rest.len());
        if consumed == self.rest.len() {
            self.rest.clear();
        } else {
            self.rest.advance(consumed);
        }
    }

    #[inline]
    pub fn take_rest(&mut self, consumed: usize) -> OwnedBytes {
        self.rest.split_to(consumed).freeze()
    }

    pub async fn read_next_chunk<R>(&mut self, reader: &mut R) -> io::Result<()>
    where
        R: AsyncRead + Send + Unpin + 'static,
    {
        let n = reader.read(self.read_buf.as_mut()).await?;
        if n == 0 {
            Err(io::Error::from(ErrorKind::UnexpectedEof))
        } else {
            self.rest.extend_from_slice(&self.read_buf[..n]);
            Ok(())
        }
    }
}

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

    use super::*;
    use crate::rpc::{
        handler::{Dummy, Handler},
        redraw::{RedrawFrame, RedrawNotification},
    };
    use rmpv::{ValueRef, encode::write_value_ref};
    use std::{
        io::Cursor,
        sync::atomic::{AtomicU32, AtomicUsize, Ordering},
    };

    #[derive(Clone)]
    struct CountingHandler;

    static UNKNOWN_REQUEST_COUNT: AtomicUsize = AtomicUsize::new(0);
    static UNKNOWN_REQUEST_MSGID: AtomicU32 = AtomicU32::new(0);

    impl Handler for CountingHandler {
        fn handle_unknown_request(msgid: u32, name: &str) {
            assert_eq!(name, "ignored");
            UNKNOWN_REQUEST_MSGID.store(msgid, Ordering::Relaxed);
            UNKNOWN_REQUEST_COUNT.fetch_add(1, Ordering::Relaxed);
        }
    }

    fn request(msgid: u32, method: &'static str) -> Vec<u8> {
        encoded_value(ValueRef::Array(vec![
            ValueRef::from(MSG_TYPE_REQUEST),
            ValueRef::from(msgid),
            ValueRef::from(method),
            ValueRef::Array(vec![]),
        ]))
    }

    fn response(msgid: u32, result: ValueRef<'_>) -> Vec<u8> {
        encoded_value(ValueRef::Array(vec![
            ValueRef::from(MSG_TYPE_RESPONSE),
            ValueRef::from(msgid),
            ValueRef::Nil,
            result,
        ]))
    }

    fn rpc_response(msgid: u32, result: Value) -> RpcResponse {
        RpcResponse {
            msgid,
            error: Value::Nil,
            result,
        }
    }

    fn notification(method: &str) -> Vec<u8> {
        encoded_value(ValueRef::Array(vec![
            ValueRef::from(MSG_TYPE_NOTIFICATION),
            ValueRef::from(method),
            ValueRef::Array(vec![]),
        ]))
    }

    fn encoded_value(value: ValueRef<'_>) -> Vec<u8> {
        let mut bytes = Vec::new();
        write_value_ref(&mut bytes, &value).unwrap();
        bytes
    }

    #[track_caller]
    fn assert_incomplete<T: Debug>(result: io::Result<T>) {
        let err = result.unwrap_err();
        assert_eq!(err.kind(), ErrorKind::UnexpectedEof);
    }

    #[track_caller]
    fn assert_invalid_data<T: Debug>(result: io::Result<T>) {
        let err = result.unwrap_err();
        assert_eq!(err.kind(), ErrorKind::InvalidData);
    }

    #[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
    }

    fn redraw_frame(bytes: &[u8]) -> RedrawFrame {
        RedrawFrame::from_slice(bytes).unwrap()
    }

    fn decoder_from_bytes(rest: Vec<u8>) -> Decoder {
        Decoder {
            rest: BytesMut::from(&rest[..]),
            read_buf: Box::new([0; DECODE_READ_BUFFER_SIZE]),
        }
    }

    fn consume_next_response_from_state<H: Handler>(
        decoder: &mut Decoder,
    ) -> io::Result<Option<RpcResponse>> {
        loop {
            let (consumed, response) = {
                let Some(decoded) = decoder.try_decode_message()? else {
                    return Ok(None);
                };
                let consumed = decoded.consumed;

                let response = match decoded.message {
                    IncomingMessage::Response(response) => Some(response),
                    IncomingMessage::Request { msgid, method } => {
                        H::handle_unknown_request(msgid, method);
                        None
                    }
                    IncomingMessage::Notification { method } => {
                        H::handle_unknown_notify(method);
                        None
                    }
                };

                (consumed, response)
            };

            decoder.consume(consumed);

            if let Some(response) = response {
                return Ok(Some(response));
            }
            if !decoder.has_rest() {
                return Ok(None);
            }
        }
    }

    async fn decode_next_response_from_state(
        decoder: &mut Decoder,
        reader: &mut Cursor<Vec<u8>>,
    ) -> RpcResponse {
        loop {
            while decoder.has_rest() {
                if let Some(response) = consume_next_response_from_state::<Dummy>(decoder).unwrap()
                {
                    return response;
                }
            }

            decoder.read_next_chunk(reader).await.unwrap();
        }
    }

    #[tokio::test]
    async fn decoder_decodes_concatenated_responses() {
        let msg_1 = response(1, ValueRef::from("one"));
        let msg_2 = response(2, ValueRef::from("two"));
        let response_1 = rpc_response(1, Value::from("one"));
        let response_2 = rpc_response(2, Value::from("two"));

        let mut bytes = msg_1;
        bytes.extend_from_slice(&msg_2);

        let mut reader = Cursor::new(bytes);
        let mut decoder = Decoder::new();

        assert_eq!(
            decode_next_response_from_state(&mut decoder, &mut reader).await,
            response_1
        );
        assert_eq!(
            decode_next_response_from_state(&mut decoder, &mut reader).await,
            response_2
        );
    }

    #[tokio::test]
    async fn decoder_ignores_requests_and_notifications_until_response() {
        let expected = rpc_response(3, Value::from("done"));
        let mut bytes = request(1, "ignored");
        bytes.extend_from_slice(&notification("ignored"));
        bytes.extend_from_slice(&response(3, ValueRef::from("done")));

        let mut reader = Cursor::new(bytes);
        let mut decoder = Decoder::new();

        assert_eq!(
            decode_next_response_from_state(&mut decoder, &mut reader).await,
            expected
        );
    }

    #[test]
    fn decoder_calls_unknown_request_after_complete_payload() {
        UNKNOWN_REQUEST_COUNT.store(0, Ordering::Relaxed);
        UNKNOWN_REQUEST_MSGID.store(0, Ordering::Relaxed);

        let mut bytes = encoded_value(ValueRef::Array(vec![
            ValueRef::from(MSG_TYPE_REQUEST),
            ValueRef::from(9),
            ValueRef::from("ignored"),
            ValueRef::Array(vec![ValueRef::from("xy")]),
        ]));
        let last_byte = bytes.pop().unwrap();
        let mut decoder = decoder_from_bytes(bytes.clone());

        assert!(
            consume_next_response_from_state::<CountingHandler>(&mut decoder)
                .unwrap()
                .is_none()
        );
        assert_eq!(UNKNOWN_REQUEST_COUNT.load(Ordering::Relaxed), 0);
        assert_eq!(decoder.rest(), bytes.as_slice());

        bytes.push(last_byte);
        decoder.rest.extend_from_slice(&[last_byte]);

        assert!(
            consume_next_response_from_state::<CountingHandler>(&mut decoder)
                .unwrap()
                .is_none()
        );
        assert_eq!(UNKNOWN_REQUEST_COUNT.load(Ordering::Relaxed), 1);
        assert_eq!(UNKNOWN_REQUEST_MSGID.load(Ordering::Relaxed), 9);
        assert!(!decoder.has_rest());
    }

    #[test]
    fn decoder_rejects_short_request_without_consuming_next_frame() {
        let mut bytes = encoded_value(ValueRef::Array(vec![
            ValueRef::from(MSG_TYPE_REQUEST),
            ValueRef::from(9),
            ValueRef::from("ignored"),
        ]));
        bytes.extend_from_slice(&response(1, ValueRef::from("ok")));
        let decoder = decoder_from_bytes(bytes.clone());

        assert_invalid_data(decoder.try_decode_message());
        assert_eq!(decoder.rest(), bytes.as_slice());
    }

    #[test]
    fn decoder_rejects_short_notification_without_consuming_next_frame() {
        let mut bytes = encoded_value(ValueRef::Array(vec![
            ValueRef::from(MSG_TYPE_NOTIFICATION),
            ValueRef::from("ignored"),
        ]));
        bytes.extend_from_slice(&response(1, ValueRef::from("ok")));
        let decoder = decoder_from_bytes(bytes.clone());

        assert_invalid_data(decoder.try_decode_message());
        assert_eq!(decoder.rest(), bytes.as_slice());
    }

    #[test]
    fn decoder_skips_extra_response_fields_without_decoding_values() {
        let bytes = encoded_value(ValueRef::Array(vec![
            ValueRef::from(MSG_TYPE_RESPONSE),
            ValueRef::from(9),
            ValueRef::Nil,
            ValueRef::from("ok"),
            ValueRef::Array(vec![
                ValueRef::Map(vec![(
                    ValueRef::from("nested"),
                    ValueRef::Array(vec![
                        ValueRef::from(1),
                        ValueRef::Nil,
                        ValueRef::from("tail"),
                    ]),
                )]),
                ValueRef::Boolean(true),
            ]),
        ]));
        let mut decoder = decoder_from_bytes(bytes);

        let response = consume_next_response_from_state::<Dummy>(&mut decoder)
            .unwrap()
            .unwrap();

        assert_eq!(response, rpc_response(9, Value::from("ok")));
        assert!(!decoder.has_rest());
    }

    #[test]
    fn decoder_waits_when_extra_response_field_is_incomplete() {
        let mut bytes = encoded_value(ValueRef::Array(vec![
            ValueRef::from(MSG_TYPE_RESPONSE),
            ValueRef::from(9),
            ValueRef::Nil,
            ValueRef::from("ok"),
            ValueRef::from("xy"),
        ]));
        let last_byte = bytes.pop().unwrap();
        let mut decoder = decoder_from_bytes(bytes.clone());

        assert!(
            consume_next_response_from_state::<Dummy>(&mut decoder)
                .unwrap()
                .is_none()
        );
        assert_eq!(decoder.rest(), bytes.as_slice());

        bytes.push(last_byte);
        decoder.rest.extend_from_slice(&[last_byte]);
        let response = consume_next_response_from_state::<Dummy>(&mut decoder)
            .unwrap()
            .unwrap();

        assert_eq!(response, rpc_response(9, Value::from("ok")));
        assert!(!decoder.has_rest());
    }

    #[test]
    fn decoder_reads_redraw_frame_without_owned_message() {
        let redraw = encoded_value(ValueRef::Array(vec![
            ValueRef::from(MSG_TYPE_NOTIFICATION),
            ValueRef::from("redraw"),
            ValueRef::Array(vec![ValueRef::Array(vec![ValueRef::from("flush")])]),
        ]));
        let msg_bytes = request(1, "test_method");

        let mut rest = redraw;
        rest.extend_from_slice(&msg_bytes);
        let mut decoder = decoder_from_bytes(rest);

        let consumed = {
            let frame = redraw_frame(decoder.rest());
            let consumed = frame.consumed();
            let mut redraw: RedrawNotification<'_> = frame.notification().unwrap();

            assert_eq!(redraw.batch_count(), 1);
            redraw
                .for_each_batch(|batch| {
                    assert_eq!(batch.name, "flush");
                    assert!(batch.args.is_empty());
                    Ok(true)
                })
                .unwrap();

            consumed
        };

        decoder.consume(consumed);
        assert_eq!(decoder.rest(), msg_bytes.as_slice());
    }

    #[test]
    fn decoder_takes_rest_after_consuming_prefix() {
        let msg_1 = request(1, "test_method");
        let msg_2 = request(2, "test_method_2");

        let mut rest = msg_1.clone();
        rest.extend_from_slice(&msg_2);
        let mut decoder = decoder_from_bytes(rest);

        decoder.consume(msg_1.len());
        let bytes = decoder.take_rest(msg_2.len());

        assert_eq!(&bytes[..], msg_2.as_slice());
        assert!(!decoder.has_rest());
    }
    #[test]
    fn array_reader_reads_float_values() {
        let f32_value = 1.25_f32;
        let f64_value = -2.5_f64;
        let bytes = encoded_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 = encoded_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 = encoded_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 = encoded_value(ValueRef::Array(vec!["not-u32".into()]));
        let mut array = ArrayReader::new(&bytes).unwrap();

        assert_invalid_data(array.read::<u32>());

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

        assert_invalid_data(array.read::<u32>());

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

        assert_invalid_data(array.read::<u32>());

        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 = encoded_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 = encoded_value(ValueRef::Array(vec!["not-usize".into()]));
        let mut array = ArrayReader::new(&bytes).unwrap();

        assert_invalid_data(array.read::<usize>());

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

        assert_invalid_data(array.read::<usize>());

        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 = encoded_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 = encoded_value(ValueRef::Array(vec!["not-u32".into()]));
        let mut array = ArrayReader::new(&bytes).unwrap();

        assert_invalid_data(array.read::<Option<u32>>());

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

        assert_invalid_data(array.read::<Option<u32>>());

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

        assert_invalid_data(array.read::<Option<u32>>());

        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 = encoded_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 = encoded_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 = encoded_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 = encoded_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_incomplete(map.read::<&str, ValueRef<'_>>());
    }

    #[test]
    fn map_reader_read_pair_reports_non_string_key() {
        let bytes = encoded_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_invalid_data(map.read::<&str, ValueRef<'_>>());
    }

    #[test]
    fn array_reader_reads_values_and_reports_boundaries() {
        let bytes = encoded_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_incomplete(array.read::<&str>());
    }

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

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

        let bytes = encoded_value(ValueRef::Array(vec![1_u64.into()]));
        let mut array = ArrayReader::new(&bytes).unwrap();
        assert_invalid_data(array.read::<bool>());

        let bytes = encoded_value(ValueRef::Array(vec![1_u64.into()]));
        let mut array = ArrayReader::new(&bytes).unwrap();
        assert_invalid_data(array.read_array(|_| Ok(())));

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

    #[test]
    fn map_reader_skips_entries() {
        let bytes = encoded_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_incomplete(map.skip_next());
    }

    #[test]
    fn map_reader_skips_remaining_entries() {
        let bytes = encoded_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_invalid_data(reader.skip_value());

        let mut reader = Bytes::new(&[]);
        assert_incomplete(reader.skip_value());

        let incomplete_bin = [Marker::Bin8.to_u8()];
        let mut reader = Bytes::new(&incomplete_bin);
        assert_incomplete(reader.skip_value());

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

    #[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_invalid_data(i64::decode(&mut reader));
    }

    #[test]
    fn bytes_reader_reports_truncated_reads() {
        let mut reader = Bytes::new(&[]);
        assert_incomplete(bool::decode(&mut reader));

        let mut reader = Bytes::new(&[]);
        assert_incomplete(i64::decode(&mut reader));

        let truncated_u64 = [Marker::U64.to_u8(), 0];
        let mut reader = Bytes::new(&truncated_u64);
        assert_incomplete(i64::decode(&mut reader));

        let truncated_array_len = [Marker::Array16.to_u8()];
        let mut reader = Bytes::new(&truncated_array_len);
        assert_incomplete(reader.read_array_reader());

        let truncated_map_len = [Marker::Map16.to_u8()];
        let mut reader = Bytes::new(&truncated_map_len);
        assert_incomplete(reader.read_map_reader());

        let truncated_value = [Marker::Str8.to_u8(), 2, b'a'];
        let mut reader = Bytes::new(&truncated_value);
        assert_incomplete(ValueRef::decode(&mut reader));

        let mut reader = Bytes::new(&[]);
        assert_incomplete(<&str>::decode(&mut reader));
    }
}