use std::io::{self, ErrorKind};
use bytes::{Buf, BytesMut};
use rmp::decode::Bytes as RmpBytes;
use rmpv::Value;
use super::decode::{ArrayReader, BytesReaderExt, Decode};
const MSG_TYPE_REQUEST: u8 = 0;
const MSG_TYPE_RESPONSE: u8 = 1;
const MSG_TYPE_NOTIFICATION: u8 = 2;
#[derive(Debug)]
pub struct NotificationPayload {
bytes: bytes::Bytes,
params_offset: usize,
params_len: u32,
}
impl NotificationPayload {
#[inline]
pub fn method(&self) -> io::Result<&str> {
let mut reader = RmpBytes::new(&self.bytes);
<&str>::decode(&mut reader)
}
#[inline]
pub fn params(&self) -> ArrayReader<'_> {
ArrayReader {
reader: RmpBytes::new(&self.bytes[self.params_offset..]),
remaining: self.params_len,
}
}
}
#[derive(Debug)]
pub struct RequestPayload {
bytes: bytes::Bytes,
msgid: u32,
params_offset: usize,
params_len: u32,
}
impl RequestPayload {
#[inline]
pub fn message_id(&self) -> u32 {
self.msgid
}
#[inline]
pub fn method(&self) -> io::Result<&str> {
let mut reader = RmpBytes::new(&self.bytes);
<&str>::decode(&mut reader)
}
#[inline]
pub fn params(&self) -> ArrayReader<'_> {
ArrayReader {
reader: RmpBytes::new(&self.bytes[self.params_offset..]),
remaining: self.params_len,
}
}
}
#[derive(Debug)]
pub struct ResponsePayload {
bytes: bytes::Bytes,
msgid: u32,
result_offset: usize,
}
impl ResponsePayload {
#[inline]
pub fn message_id(&self) -> u32 {
self.msgid
}
#[inline]
pub fn decode<'a, D: Decode<'a>>(&'a self) -> io::Result<Result<D, Value>> {
if self.has_error() {
Ok(Err(self.decode_error()?))
} else {
Ok(Ok(self.decode_result()?))
}
}
#[inline]
pub fn decode_result<'a, D: Decode<'a>>(&'a self) -> io::Result<D> {
let mut reader = RmpBytes::new(&self.bytes[self.result_offset..]);
D::decode(&mut reader)
}
#[inline]
pub fn decode_error<'a, D: Decode<'a>>(&'a self) -> io::Result<D> {
let mut reader = RmpBytes::new(&self.bytes[..self.result_offset]);
D::decode(&mut reader)
}
#[inline]
fn has_error(&self) -> bool {
self.bytes[0] != rmp::Marker::Null.to_u8()
}
}
#[derive(Debug)]
pub enum Payload {
Notification(NotificationPayload),
Request(RequestPayload),
Response(ResponsePayload),
}
impl Payload {
pub fn decode(buf: &mut BytesMut) -> io::Result<Self> {
let mut reader = RmpBytes::new(buf);
let len = reader.read_array_len()?;
if len == 0 {
return Err(ErrorKind::InvalidData.into());
}
match u8::decode(&mut reader)? {
MSG_TYPE_NOTIFICATION if len == 3 => {
let payload_offset = reader.position() as usize;
let payload_len = reader.remaining_slice().len();
reader.skip_value()?;
let params_len = reader.read_array_len()?;
let params_offset = payload_len - reader.remaining_slice().len();
reader.skip_values(params_len)?;
let frame_len = payload_len - reader.remaining_slice().len() + payload_offset;
let mut bytes = buf.split_to(frame_len).freeze();
bytes.advance(payload_offset);
Ok(Self::Notification(NotificationPayload {
bytes,
params_offset,
params_len,
}))
}
MSG_TYPE_REQUEST if len == 4 => {
let msgid = u32::decode(&mut reader)?;
let payload_offset = reader.position() as usize;
let payload_len = reader.remaining_slice().len();
reader.skip_value()?;
let params_len = reader.read_array_len()?;
let params_offset = payload_len - reader.remaining_slice().len();
reader.skip_values(params_len)?;
let frame_len = payload_offset + payload_len - reader.remaining_slice().len();
let mut bytes = buf.split_to(frame_len).freeze();
bytes.advance(payload_offset);
Ok(Self::Request(RequestPayload {
bytes,
msgid,
params_offset,
params_len,
}))
}
MSG_TYPE_RESPONSE if len == 4 => {
let msgid = u32::decode(&mut reader)?;
let payload_offset = reader.position() as usize;
let payload_len = reader.remaining_slice().len();
reader.skip_value()?;
let result_offset = payload_len - reader.remaining_slice().len();
reader.skip_value()?;
let frame_len = payload_offset + payload_len - reader.remaining_slice().len();
let mut bytes = buf.split_to(frame_len).freeze();
bytes.advance(payload_offset);
Ok(Self::Response(ResponsePayload {
bytes,
msgid,
result_offset,
}))
}
_ => Err(ErrorKind::InvalidData.into()),
}
}
}
#[cfg(test)]
mod tests {
use std::{fmt::Debug, io::ErrorKind};
use rmpv::{ValueRef, encode::write_value_ref};
use super::*;
fn encode(value: ValueRef<'_>) -> Vec<u8> {
let mut bytes = vec![];
write_value_ref(&mut bytes, &value).unwrap();
bytes
}
fn notification(method: ValueRef<'_>, params: ValueRef<'_>) -> Vec<u8> {
encode(ValueRef::Array(vec![2_u64.into(), method, params]))
}
fn request(msgid: ValueRef<'_>, method: ValueRef<'_>, params: ValueRef<'_>) -> Vec<u8> {
encode(ValueRef::Array(vec![0_u64.into(), msgid, method, params]))
}
fn response(msgid: ValueRef<'_>, error: ValueRef<'_>, result: ValueRef<'_>) -> Vec<u8> {
encode(ValueRef::Array(vec![1_u64.into(), msgid, error, result]))
}
#[track_caller]
fn decode_notification(bytes: Vec<u8>) -> NotificationPayload {
let mut bytes = BytesMut::from(bytes.as_slice());
match Payload::decode(&mut bytes).unwrap() {
Payload::Notification(payload) => {
assert!(bytes.is_empty());
payload
}
x => panic!("expected notification payload: {x:?}"),
}
}
#[track_caller]
fn decode_request(bytes: Vec<u8>) -> RequestPayload {
let mut bytes = BytesMut::from(bytes.as_slice());
match Payload::decode(&mut bytes).unwrap() {
Payload::Request(payload) => {
assert!(bytes.is_empty());
payload
}
x => panic!("expected request payload: {x:?}"),
}
}
#[track_caller]
fn decode_response(bytes: Vec<u8>) -> ResponsePayload {
let mut bytes = BytesMut::from(bytes.as_slice());
match Payload::decode(&mut bytes).unwrap() {
Payload::Response(payload) => {
assert!(bytes.is_empty());
payload
}
x => panic!("expected response payload: {x:?}"),
}
}
#[track_caller]
fn assert_error_kind<T: Debug>(result: io::Result<T>, kind: ErrorKind) {
let error = result.unwrap_err();
assert_eq!(error.kind(), kind);
}
#[test]
fn payload_decode_consumes_one_message_and_leaves_the_next_message() {
let first = notification("redraw".into(), ValueRef::Array(vec![]));
let second = request(
42_u64.into(),
"nvim_get_mode".into(),
ValueRef::Array(vec![]),
);
let mut input = BytesMut::from([first.as_slice(), second.as_slice()].concat().as_slice());
let Payload::Notification(payload) = Payload::decode(&mut input).unwrap() else {
panic!("expected notification payload");
};
assert_eq!(payload.method().unwrap(), "redraw");
assert_eq!(input.as_ref(), second);
let Payload::Request(payload) = Payload::decode(&mut input).unwrap() else {
panic!("expected request payload");
};
assert_eq!(payload.message_id(), 42);
assert!(input.is_empty());
}
#[test]
fn payload_decode_does_not_consume_incomplete_message() {
let complete = request(
7_u64.into(),
"nvim_input".into(),
ValueRef::Array(vec!["x".into()]),
);
let incomplete = &complete[..complete.len() - 1];
let mut input = BytesMut::from(incomplete);
assert_error_kind(Payload::decode(&mut input), ErrorKind::UnexpectedEof);
assert_eq!(input.as_ref(), incomplete);
}
#[test]
fn payload_decode_treats_empty_input_as_incomplete() {
let mut input = BytesMut::new();
assert_error_kind(Payload::decode(&mut input), ErrorKind::UnexpectedEof);
assert!(input.is_empty());
}
#[test]
fn payload_decode_rejects_non_array_message_without_consuming_input() {
let encoded = encode(ValueRef::from("not an RPC message"));
let mut input = BytesMut::from(encoded.as_slice());
assert_error_kind(Payload::decode(&mut input), ErrorKind::InvalidData);
assert_eq!(input.as_ref(), encoded);
}
#[test]
fn payload_decode_rejects_empty_envelope_without_consuming_input() {
let encoded = encode(ValueRef::Array(vec![]));
let mut input = BytesMut::from(encoded.as_slice());
assert_error_kind(Payload::decode(&mut input), ErrorKind::InvalidData);
assert_eq!(input.as_ref(), encoded);
}
#[test]
fn payload_decode_rejects_unknown_message_type() {
let encoded = encode(ValueRef::Array(vec![3_u64.into()]));
let mut input = BytesMut::from(encoded.as_slice());
assert_error_kind(Payload::decode(&mut input), ErrorKind::InvalidData);
}
#[test]
fn payload_decode_rejects_notification_with_wrong_envelope_length() {
let encoded = encode(ValueRef::Array(vec![
2_u64.into(),
"redraw".into(),
ValueRef::Array(vec![]),
ValueRef::Nil,
]));
let mut input = BytesMut::from(encoded.as_slice());
assert_error_kind(Payload::decode(&mut input), ErrorKind::InvalidData);
}
#[test]
fn payload_decode_rejects_request_with_wrong_envelope_length() {
let encoded = encode(ValueRef::Array(vec![
0_u64.into(),
1_u64.into(),
"nvim_input".into(),
]));
let mut input = BytesMut::from(encoded.as_slice());
assert_error_kind(Payload::decode(&mut input), ErrorKind::InvalidData);
}
#[test]
fn payload_decode_rejects_response_with_wrong_envelope_length() {
let encoded = encode(ValueRef::Array(vec![
1_u64.into(),
1_u64.into(),
ValueRef::Nil,
]));
let mut input = BytesMut::from(encoded.as_slice());
assert_error_kind(Payload::decode(&mut input), ErrorKind::InvalidData);
}
#[test]
fn payload_decode_rejects_notification_with_non_array_params() {
let encoded = notification("redraw".into(), ValueRef::Nil);
let mut input = BytesMut::from(encoded.as_slice());
assert_error_kind(Payload::decode(&mut input), ErrorKind::InvalidData);
}
#[test]
fn payload_decode_rejects_request_with_non_array_params() {
let encoded = request(1_u64.into(), "nvim_input".into(), ValueRef::Nil);
let mut input = BytesMut::from(encoded.as_slice());
assert_error_kind(Payload::decode(&mut input), ErrorKind::InvalidData);
}
#[test]
fn payload_decode_rejects_request_with_invalid_message_id() {
let encoded = request(
"not an integer".into(),
"nvim_input".into(),
ValueRef::Array(vec![]),
);
let mut input = BytesMut::from(encoded.as_slice());
assert_error_kind(Payload::decode(&mut input), ErrorKind::InvalidData);
}
#[test]
fn payload_decode_rejects_response_with_out_of_range_message_id() {
let encoded = response(
(u64::from(u32::MAX) + 1).into(),
ValueRef::Nil,
ValueRef::Nil,
);
let mut input = BytesMut::from(encoded.as_slice());
assert_error_kind(Payload::decode(&mut input), ErrorKind::InvalidData);
}
#[test]
fn notification_payload_exposes_method_and_params() {
let payload = decode_notification(notification(
"redraw".into(),
ValueRef::Array(vec![12_u64.into(), "text".into(), ValueRef::Boolean(true)]),
));
assert_eq!(payload.method().unwrap(), "redraw");
let mut params = payload.params();
assert_eq!(params.remaining(), 3);
assert_eq!(params.read::<u64>().unwrap(), 12);
assert_eq!(params.read::<&str>().unwrap(), "text");
assert!(params.read::<bool>().unwrap());
assert!(params.is_empty());
}
#[test]
fn notification_payload_preserves_offsets_for_long_method_and_array16_params() {
let method = "m".repeat(40);
let params = (0_u64..20).map(ValueRef::from).collect();
let payload = decode_notification(notification(
ValueRef::from(method.as_str()),
ValueRef::Array(params),
));
assert_eq!(payload.method().unwrap(), method);
let mut params = payload.params();
assert_eq!(params.remaining(), 20);
assert_eq!(params.read::<u64>().unwrap(), 0);
params.skip_remaining().unwrap();
assert!(params.is_empty());
}
#[test]
fn notification_payload_reports_non_string_method_when_method_is_read() {
let payload = decode_notification(notification(
123_u64.into(),
ValueRef::Array(vec![ValueRef::Boolean(true)]),
));
assert_error_kind(payload.method(), ErrorKind::InvalidData);
assert!(payload.params().read::<bool>().unwrap());
}
#[test]
fn request_payload_exposes_message_id_method_and_params() {
let payload = decode_request(request(
42_u64.into(),
"nvim_input".into(),
ValueRef::Array(vec!["abc".into(), ValueRef::Boolean(false)]),
));
assert_eq!(payload.message_id(), 42);
assert_eq!(payload.method().unwrap(), "nvim_input");
let mut params = payload.params();
assert_eq!(params.remaining(), 2);
assert_eq!(params.read::<&str>().unwrap(), "abc");
assert!(!params.read::<bool>().unwrap());
assert!(params.is_empty());
}
#[test]
fn request_payload_preserves_offsets_with_maximum_message_id() {
let method = "nvim_very_long_method_name_for_array_offsets";
let payload = decode_request(request(
u64::from(u32::MAX).into(),
method.into(),
ValueRef::Array(vec![1_u64.into(), 2_u64.into()]),
));
assert_eq!(payload.message_id(), u32::MAX);
assert_eq!(payload.method().unwrap(), method);
let mut params = payload.params();
assert_eq!(params.read::<u64>().unwrap(), 1);
assert_eq!(params.read::<u64>().unwrap(), 2);
}
#[test]
fn request_payload_reports_non_string_method_when_method_is_read() {
let payload = decode_request(request(
1_u64.into(),
ValueRef::Nil,
ValueRef::Array(vec![]),
));
assert_eq!(payload.message_id(), 1);
assert_error_kind(payload.method(), ErrorKind::InvalidData);
assert!(payload.params().is_empty());
}
#[test]
fn response_payload_decodes_success_response() {
let payload = decode_response(response(
11_u64.into(),
ValueRef::Nil,
ValueRef::Map(vec![("mode".into(), "normal".into())]),
));
assert_eq!(payload.message_id(), 11);
assert_eq!(
payload.decode::<Value>().unwrap(),
Ok(Value::Map(vec![("mode".into(), "normal".into())]))
);
let payload = decode_response(response(12_u64.into(), ValueRef::Nil, ValueRef::Nil));
assert_eq!(payload.message_id(), 12);
assert_eq!(payload.decode::<Value>().unwrap(), Ok(Value::Nil));
}
#[test]
fn response_payload_decodes_error_response() {
let error = ValueRef::Array(vec![1_u64.into(), "failed".into()]);
let payload = decode_response(response(12_u64.into(), error, ValueRef::Nil));
assert_eq!(payload.message_id(), 12);
assert_eq!(
payload.decode::<Value>().unwrap(),
Err(Value::Array(vec![1_u64.into(), "failed".into()]))
);
}
#[test]
fn response_payload_decodes_result_and_error_fields_independently() {
let payload = decode_response(response(13_u64.into(), ValueRef::Nil, ValueRef::from("ok")));
assert_eq!(payload.decode_result::<&str>().unwrap(), "ok");
assert_eq!(payload.decode_error::<Option<Value>>().unwrap(), None);
}
#[test]
fn response_payload_reports_result_type_mismatch() {
let payload = decode_response(response(
14_u64.into(),
ValueRef::Nil,
ValueRef::from("not an integer"),
));
assert_error_kind(payload.decode_result::<u32>(), ErrorKind::InvalidData);
}
#[test]
fn response_payload_preserves_result_offset_after_nested_error() {
let error = ValueRef::Map(vec![(
"exception".into(),
ValueRef::Array(vec!["type".into(), "message".into()]),
)]);
let result = ValueRef::Array(vec![1_u64.into(), 2_u64.into(), 3_u64.into()]);
let payload = decode_response(response(15_u64.into(), error, result));
let decoded_error = payload.decode_error::<Value>().unwrap();
assert!(decoded_error.is_map());
assert_eq!(
payload.decode_result::<Value>().unwrap(),
Value::Array(vec![1_u64.into(), 2_u64.into(), 3_u64.into()])
);
}
}