navy-nvim-rs 0.0.18

//! An active neovim session.
use std::{
    fmt,
    future::Future,
    io::{self, Error},
    process::Stdio,
    sync::{
        Arc,
        atomic::{AtomicU32, Ordering},
    },
};

use parking_lot::Mutex as SyncMutex;
use tokio::io::{AsyncRead, AsyncWrite, WriteHalf, split};
use tokio::net::{TcpStream, ToSocketAddrs};
use tokio::process::{Child, ChildStdin, Command};
use tokio::sync::{
    Mutex,
    mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel},
    oneshot,
};
use tokio::task::JoinHandle;

use crate::{
    error::{MessageIdNotFound, NeovimError, ResponseReceiveError, ResponseSendError},
    rpc::{
        decode::{DecodeState, IncomingMessage, RpcResponse},
        encode::{EncodeArgs, Encoder},
        handler::Handler,
        redraw::{RedrawFrame, RedrawFrameInfo},
    },
};
use rmpv::Value;

type ResponseResult = Result<Result<Value, Value>, io::Error>;
type Queue = SyncMutex<Vec<(u32, oneshot::Sender<ResponseResult>)>>;

enum HandlerMessage {
    Response(RpcResponse),
    RedrawPayload(RedrawFrame),
    UnknownRequest { msgid: u32 },
}

impl fmt::Debug for HandlerMessage {
    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Response(response) => fmt.debug_tuple("Response").field(response).finish(),
            Self::RedrawPayload(frame) => fmt
                .debug_struct("RedrawPayload")
                .field("len", &frame.consumed())
                .finish(),
            Self::UnknownRequest { msgid } => fmt
                .debug_struct("UnknownRequest")
                .field("msgid", msgid)
                .finish(),
        }
    }
}

/// An active Neovim session.
pub(crate) struct NeovimInner<W>
where
    W: AsyncWrite + Send + Unpin + 'static,
{
    encoder: Mutex<Encoder<W>>,
    queue: Queue,
    msgid_counter: AtomicU32,
}

pub struct Neovim<H: Handler> {
    pub(crate) inner: Arc<NeovimInner<H::Writer>>,
}

impl<H: Handler> Clone for Neovim<H> {
    fn clone(&self) -> Self {
        Neovim {
            inner: self.inner.clone(),
        }
    }
}

impl<H: Handler> PartialEq for Neovim<H> {
    fn eq(&self, other: &Self) -> bool {
        Arc::ptr_eq(&self.inner, &other.inner)
    }
}
impl<H: Handler> Eq for Neovim<H> {}

impl<H: Handler> Neovim<H> {
    pub fn new<R>(
        reader: R,
        writer: H::Writer,
        handler: H,
    ) -> (Self, impl Future<Output = io::Result<()>>)
    where
        R: AsyncRead + Send + Unpin + 'static,
    {
        let req = Neovim {
            inner: Arc::new(NeovimInner {
                encoder: Mutex::new(Encoder::new(writer)),
                queue: SyncMutex::new(Vec::new()),
                msgid_counter: AtomicU32::new(0),
            }),
        };

        let (sender, receiver) = unbounded_channel();
        let io_req = req.clone();
        let handler_req = req.clone();
        let fut = async move {
            tokio::try_join!(
                io_req.io_loop(reader, sender),
                handler_req.handler_loop(handler, receiver),
            )
            .map(|_| ())
        };

        (req, fut)
    }

    /// Connect to a Neovim instance via TCP.
    pub async fn connect<A>(addr: A, handler: H) -> io::Result<(Self, JoinHandle<io::Result<()>>)>
    where
        H: Handler<Writer = WriteHalf<TcpStream>>,
        A: ToSocketAddrs,
    {
        let stream = TcpStream::connect(addr).await?;
        let (reader, writer) = split(stream);
        let (neovim, io) = Self::new(reader, writer, handler);
        let io_handle = tokio::spawn(io);

        Ok((neovim, io_handle))
    }

    /// Spawn an embedded Neovim process and connect to it over stdio.
    ///
    /// stdin/stdout will be rewritten to `Stdio::piped()`.
    pub fn spawn(
        mut cmd: Command,
        handler: H,
    ) -> io::Result<(Self, JoinHandle<io::Result<()>>, Child)>
    where
        H: Handler<Writer = ChildStdin> + Send + 'static,
    {
        let mut child = cmd.stdin(Stdio::piped()).stdout(Stdio::piped()).spawn()?;
        let stdout = child
            .stdout
            .take()
            .ok_or_else(|| Error::other("Can't open stdout"))?;
        let stdin = child
            .stdin
            .take()
            .ok_or_else(|| Error::other("Can't open stdin"))?;

        let (neovim, io) = Self::new(stdout, stdin, handler);
        let io_handle = tokio::spawn(io);

        Ok((neovim, io_handle, child))
    }

    #[inline]
    pub async fn request<A: EncodeArgs>(&self, method: &'static str, args: A) -> io::Result<Value> {
        let msgid = self.inner.msgid_counter.fetch_add(1, Ordering::Relaxed);
        let (sender, receiver) = oneshot::channel();

        self.inner.queue.lock().push((msgid, sender));

        self.inner
            .encoder
            .lock()
            .await
            .encode_request(msgid, method, args)
            .await?;

        receive_response(receiver, method).await
    }

    #[inline]
    pub async fn notify<A: EncodeArgs>(&self, method: &'static str, args: A) -> io::Result<()> {
        self.inner
            .encoder
            .lock()
            .await
            .encode_notify(method, args)
            .await
    }

    fn send_error_to_callers(&self, queue: &Queue, err: &io::Error) {
        let mut queue = queue.lock();
        for (_, sender) in queue.drain(..) {
            let err = if let Some(raw_os_error) = err.raw_os_error() {
                io::Error::from_raw_os_error(raw_os_error)
            } else {
                err.kind().into()
            };
            let _ = sender.send(Err(err));
        }
    }

    async fn handler_loop(
        self,
        handler: H,
        mut receiver: UnboundedReceiver<RedrawFrame>,
    ) -> io::Result<()> {
        loop {
            let Some(msg) = receiver.recv().await else {
                /* If our receiver closes, that just means that io_handler started
                 * shutting down. This is normal, so shut down along with it and don't
                 * report an error
                 */
                break Ok(());
            };

            let redraw = msg.notification()?;
            handler.handle_redraw(redraw)?;
        }
    }

    async fn io_loop<R>(self, mut reader: R, sender: UnboundedSender<RedrawFrame>) -> io::Result<()>
    where
        R: AsyncRead + Send + Unpin + 'static,
    {
        let mut decoder = DecodeState::new();

        loop {
            let msg = match Self::decode_next(&mut decoder, &mut reader).await {
                Ok(msg) => msg,
                Err(err) => {
                    self.send_error_to_callers(&self.inner.queue, &err);
                    return Err(err);
                }
            };

            match msg {
                HandlerMessage::RedrawPayload(frame) => {
                    // Send redraw notifications to handler_loop().
                    sender.send(frame).unwrap();
                }
                HandlerMessage::UnknownRequest { msgid } => {
                    self.inner
                        .encoder
                        .lock()
                        .await
                        .encode_error_response(msgid, "Not implemented")
                        .await?;
                }
                HandlerMessage::Response(RpcResponse {
                    msgid,
                    result,
                    error,
                }) => {
                    let sender = find_sender(&self.inner.queue, msgid)?;
                    let response = if error == Value::Nil {
                        Ok(result)
                    } else {
                        Err(error)
                    };
                    if let Err(Ok(response)) = sender.send(Ok(response)) {
                        return Err(Error::other(ResponseSendError::new(msgid, response)));
                    }
                }
            }
        }
    }

    async fn decode_next<R>(decoder: &mut DecodeState, reader: &mut R) -> io::Result<HandlerMessage>
    where
        R: AsyncRead + Send + Unpin + 'static,
    {
        loop {
            if decoder.has_rest() {
                match RedrawFrameInfo::probe(decoder.rest()) {
                    Ok(Some(info)) => {
                        let bytes = decoder.take_rest(info.consumed());
                        let frame = info.frame(bytes);
                        return Ok(HandlerMessage::RedrawPayload(frame));
                    }
                    Ok(None) => {
                        if let Some(decoded) = decoder.try_decode_message()? {
                            let msg = match decoded.message {
                                IncomingMessage::Response(response) => {
                                    Some(HandlerMessage::Response(response))
                                }
                                IncomingMessage::Request { msgid, method } => {
                                    H::handle_unknown_request(msgid, method);
                                    Some(HandlerMessage::UnknownRequest { msgid })
                                }
                                IncomingMessage::Notification { method } => {
                                    H::handle_unknown_notify(method);
                                    None
                                }
                            };

                            decoder.consume(decoded.consumed);

                            if let Some(msg) = msg {
                                return Ok(msg);
                            }
                            continue;
                        }
                    }
                    Err(err) if err.kind() == io::ErrorKind::UnexpectedEof => {}
                    Err(err) => return Err(err),
                }
            }

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

#[inline]
async fn receive_response(
    receiver: oneshot::Receiver<ResponseResult>,
    method: &'static str,
) -> io::Result<Value> {
    match receiver.await {
        Ok(Ok(response)) => response.map_err(|value| Error::other(NeovimError::new(method, value))),
        Ok(Err(err)) => Err(err),
        Err(err) => Err(Error::other(ResponseReceiveError::new(method, err))),
    }
}

/* The idea to use Vec here instead of HashMap
 * is that Vec is faster on small queue sizes
 * in most cases Vec.len = 1 so we just take first item in iteration.
 */
#[inline]
fn find_sender(queue: &Queue, msgid: u32) -> io::Result<oneshot::Sender<ResponseResult>> {
    let mut queue = queue.lock();
    let Some(pos) = queue.iter().position(|req| req.0 == msgid) else {
        return Err(Error::other(MessageIdNotFound::new(msgid)));
    };
    let sender = queue.remove(pos).1;
    Ok(sender)
}

#[cfg(test)]
mod tests {
    use std::{
        env,
        io::{self, Cursor, ErrorKind},
        path::PathBuf,
        sync::{
            Arc,
            atomic::{AtomicU32, AtomicUsize, Ordering},
        },
        time::Duration,
    };

    use rmpv::{ValueRef, encode::write_value};
    use tokio::{io::AsyncWriteExt, task::yield_now, time};

    use super::*;
    use crate::UiAttachOptions;
    use crate::rpc::{handler::Dummy, redraw::RedrawNotification};

    #[derive(Clone)]
    struct CountingHandler {
        redraw_count: Arc<AtomicUsize>,
    }

    #[derive(Debug)]
    enum RedrawSignal {
        Any,
        GridResize { cols: u64, rows: u64 },
    }

    #[derive(Clone)]
    struct ChildProcessHandler {
        redraw_tx: UnboundedSender<RedrawSignal>,
    }

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

    impl Handler for CountingHandler {
        type Writer = Vec<u8>;

        fn handle_redraw(&self, mut redraw: RedrawNotification<'_>) -> io::Result<()> {
            redraw.for_each_batch(|batch| {
                assert_eq!(batch.name, "flush");
                assert!(batch.args.is_empty());
                Ok(true)
            })?;
            self.redraw_count.fetch_add(1, Ordering::Relaxed);
            Ok(())
        }

        fn handle_unknown_notify(name: &str) {
            assert_eq!(name, "ignored");
            UNKNOWN_NOTIFICATION_COUNT.fetch_add(1, Ordering::Relaxed);
        }

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

    impl Handler for ChildProcessHandler {
        type Writer = ChildStdin;

        fn handle_redraw(&self, mut redraw: RedrawNotification<'_>) -> io::Result<()> {
            let _ = self.redraw_tx.send(RedrawSignal::Any);

            redraw.for_each_batch(|batch| {
                if batch.name == "grid_resize" {
                    while !batch.args.is_empty() {
                        batch.args.read_array(|args| {
                            let _grid = args.read::<u64>()?;
                            let cols = args.read()?;
                            let rows = args.read()?;
                            let _ = self.redraw_tx.send(RedrawSignal::GridResize { cols, rows });
                            Ok(())
                        })?;
                    }
                }
                Ok(true)
            })
        }
    }

    fn nvim_test_bin() -> PathBuf {
        let default_nvim_bin = if cfg!(windows) { "nvim.exe" } else { "nvim" };
        match env::var_os("NVIMRS_TEST_BIN") {
            Some(path) => {
                let path = PathBuf::from(path);
                assert!(
                    path.exists(),
                    "nvim bin from NVIMRS_TEST_BIN does not exist: {}",
                    path.display()
                );
                path
            }
            None => PathBuf::from(default_nvim_bin),
        }
    }

    async fn recv_redraw(redraw_rx: &mut UnboundedReceiver<RedrawSignal>) -> RedrawSignal {
        time::timeout(Duration::from_secs(5), redraw_rx.recv())
            .await
            .expect("timed out waiting for redraw notification")
            .expect("redraw channel closed")
    }

    async fn wait_for_grid_resize(
        redraw_rx: &mut UnboundedReceiver<RedrawSignal>,
        expected_cols: u64,
        expected_rows: u64,
    ) {
        loop {
            match recv_redraw(redraw_rx).await {
                RedrawSignal::GridResize { cols, rows }
                    if cols == expected_cols && rows == expected_rows =>
                {
                    return;
                }
                _ => {}
            }
        }
    }

    fn encoded_value(value: Value) -> Vec<u8> {
        let mut bytes = Vec::new();
        write_value(&mut bytes, &value).unwrap();
        bytes
    }

    fn request_bytes(msgid: u32, method: &str, params: Value) -> Vec<u8> {
        let mut bytes = Vec::new();
        rmp::encode::write_array_len(&mut bytes, 4).unwrap();
        rmp::encode::write_uint8(&mut bytes, 0).unwrap();
        rmp::encode::write_u32(&mut bytes, msgid).unwrap();
        rmp::encode::write_str(&mut bytes, method).unwrap();
        write_value(&mut bytes, &params).unwrap();
        bytes
    }

    fn redraw_bytes() -> Vec<u8> {
        encoded_value(Value::from(vec![
            Value::from(2),
            Value::from("redraw"),
            Value::from(vec![Value::from(vec![Value::from("flush")])]),
        ]))
    }

    fn ignored_request_bytes() -> Vec<u8> {
        encoded_value(Value::from(vec![
            Value::from(0),
            Value::from(99),
            Value::from("ignored"),
            Value::from(Vec::<Value>::new()),
        ]))
    }

    fn ignored_notification_bytes() -> Vec<u8> {
        encoded_value(Value::from(vec![
            Value::from(2),
            Value::from("ignored"),
            Value::from(Vec::<Value>::new()),
        ]))
    }

    fn response_bytes(msgid: u32, result: Value, error: Value) -> Vec<u8> {
        let mut bytes = Vec::new();
        rmp::encode::write_array_len(&mut bytes, 4).unwrap();
        rmp::encode::write_uint8(&mut bytes, 1).unwrap();
        rmp::encode::write_u32(&mut bytes, msgid).unwrap();
        write_value(&mut bytes, &error).unwrap();
        write_value(&mut bytes, &result).unwrap();
        bytes
    }

    async fn wait_for_pending_request(queue: &Queue) {
        for _ in 0..100 {
            if !queue.lock().is_empty() {
                return;
            }
            yield_now().await;
        }
        panic!("request was not queued");
    }

    fn test_neovim<H>() -> Neovim<H>
    where
        H: Handler<Writer = Vec<u8>>,
    {
        Neovim {
            inner: Arc::new(NeovimInner {
                encoder: Mutex::new(Encoder::new(Vec::new())),
                queue: SyncMutex::new(Vec::new()),
                msgid_counter: AtomicU32::new(0),
            }),
        }
    }

    #[tokio::test]
    async fn spawn_starts_embedded_nvim_and_attaches_ui() {
        let path = nvim_test_bin();

        let mut cmd = Command::new(path);
        cmd.args(["--embed", "--headless", "--clean", "-i", "NONE"])
            .stderr(Stdio::piped());

        let (redraw_tx, mut redraw_rx) = unbounded_channel();
        let handler = ChildProcessHandler { redraw_tx };
        let (nvim, io_handle, mut child) = Neovim::spawn(cmd, handler).unwrap();

        let mut options = UiAttachOptions::default();
        options.set_rgb(true);
        options.set_linegrid_external(true);
        options.set_hlstate_external(true);
        time::timeout(Duration::from_secs(5), nvim.ui_attach(80, 24, options))
            .await
            .expect("timed out waiting for nvim_ui_attach response")
            .unwrap();
        let _ = recv_redraw(&mut redraw_rx).await;

        nvim.ui_try_resize(100, 30).await.unwrap();
        wait_for_grid_resize(&mut redraw_rx, 100, 30).await;

        let quit_input = "<Cmd>qall!<CR>";
        let written = time::timeout(Duration::from_secs(5), nvim.request_input(quit_input))
            .await
            .expect("timed out waiting for nvim_input response")
            .unwrap();
        assert_eq!(written, quit_input.len() as i64);

        let status = match time::timeout(Duration::from_secs(5), child.wait()).await {
            Ok(status) => status.unwrap(),
            Err(err) => {
                child.kill().await.unwrap();
                panic!("timed out waiting for nvim process to exit: {err}");
            }
        };
        assert!(status.success(), "nvim exited with status: {status}");
        io_handle.abort();
    }

    #[tokio::test]
    async fn spawn_reports_channel_close_as_call_error() {
        let path = nvim_test_bin();

        let mut cmd = Command::new(path);
        cmd.args(["--embed", "--headless", "--clean", "-i", "NONE"])
            .stderr(Stdio::piped());

        let (redraw_tx, _redraw_rx) = unbounded_channel();
        let handler = ChildProcessHandler { redraw_tx };
        let (nvim, io_handle, mut child) = Neovim::spawn(cmd, handler).unwrap();

        let api_info = time::timeout(Duration::from_secs(5), nvim.get_api_info())
            .await
            .expect("timed out waiting for nvim_get_api_info response")
            .unwrap();
        let channel = api_info[0].as_i64().unwrap();
        let close_arg = format!("chanclose({channel})");
        let close_cmd = ValueRef::Map(vec![
            (ValueRef::from("cmd"), ValueRef::from("call")),
            (
                ValueRef::from("args"),
                ValueRef::Array(vec![ValueRef::from(close_arg.as_str())]),
            ),
        ]);

        let err = time::timeout(
            Duration::from_secs(5),
            nvim.cmd(close_cmd, ValueRef::Map(vec![])),
        )
        .await
        .expect("timed out waiting for chanclose response")
        .expect_err("chanclose should close the RPC channel");
        assert_eq!(
            err.kind(),
            ErrorKind::UnexpectedEof,
            "unexpected error: {err}"
        );

        if time::timeout(Duration::from_secs(5), child.wait())
            .await
            .is_err()
        {
            child.kill().await.unwrap();
        }
        io_handle.abort();
    }

    #[tokio::test]
    async fn decode_next_reports_incoming_requests_and_ignores_notifications() {
        UNKNOWN_REQUEST_COUNT.store(0, Ordering::Relaxed);
        UNKNOWN_NOTIFICATION_COUNT.store(0, Ordering::Relaxed);
        UNKNOWN_REQUEST_MSGID.store(0, Ordering::Relaxed);

        let result = Value::from("ok");
        let response = response_bytes(7, result.clone(), Value::Nil);
        let mut bytes = ignored_request_bytes();
        bytes.extend_from_slice(&ignored_notification_bytes());
        bytes.extend_from_slice(&response);

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

        let msg = Neovim::<CountingHandler>::decode_next(&mut decoder, &mut reader)
            .await
            .unwrap();

        match msg {
            HandlerMessage::UnknownRequest { msgid } => {
                assert_eq!(msgid, 99);
            }
            msg => panic!("unexpected message: {msg:?}"),
        }

        assert_eq!(UNKNOWN_REQUEST_COUNT.load(Ordering::Relaxed), 1);
        assert_eq!(UNKNOWN_NOTIFICATION_COUNT.load(Ordering::Relaxed), 0);
        assert_eq!(UNKNOWN_REQUEST_MSGID.load(Ordering::Relaxed), 99);

        let msg = Neovim::<CountingHandler>::decode_next(&mut decoder, &mut reader)
            .await
            .unwrap();

        match msg {
            HandlerMessage::Response(RpcResponse {
                msgid,
                result: decoded_result,
                error,
            }) => {
                assert_eq!(msgid, 7);
                assert_eq!(decoded_result, result);
                assert_eq!(error, Value::Nil);
            }
            msg => panic!("unexpected message: {msg:?}"),
        }

        assert_eq!(UNKNOWN_REQUEST_COUNT.load(Ordering::Relaxed), 1);
        assert_eq!(UNKNOWN_NOTIFICATION_COUNT.load(Ordering::Relaxed), 1);
        assert_eq!(UNKNOWN_REQUEST_MSGID.load(Ordering::Relaxed), 99);
    }

    #[tokio::test]
    async fn io_loop_replies_to_incoming_unknown_request_with_default_error() {
        let nvim = test_neovim::<Dummy<Vec<u8>>>();
        let inner = nvim.inner.clone();
        let (sender, _receiver) = unbounded_channel();

        let err = nvim
            .io_loop(Cursor::new(ignored_request_bytes()), sender)
            .await
            .unwrap_err();

        assert_eq!(err.kind(), ErrorKind::UnexpectedEof);

        let writer = inner.encoder.lock().await;
        assert_eq!(
            writer.writer(),
            &response_bytes(99, Value::Nil, Value::from("Not implemented"))
        );
    }

    #[tokio::test]
    async fn decode_next_waits_for_complete_redraw_prefix() {
        let redraw = redraw_bytes();
        let prefix = [0x93, b'\x02', 0xa6, b'r', b'e', b'd', b'r', b'a', b'w'];

        assert!(redraw.starts_with(&prefix));

        let mut decoder = DecodeState::new();
        let mut prefix_reader = Cursor::new(prefix.to_vec());
        decoder.read_next_chunk(&mut prefix_reader).await.unwrap();
        let mut reader = Cursor::new(redraw[prefix.len()..].to_vec());
        let msg = Neovim::<CountingHandler>::decode_next(&mut decoder, &mut reader)
            .await
            .unwrap();

        match msg {
            HandlerMessage::RedrawPayload(frame) => {
                assert_eq!(frame.as_bytes(), redraw.as_slice());
            }
            msg => panic!("unexpected message: {msg:?}"),
        }
    }

    #[tokio::test]
    async fn handler_loop_handles_redraw_message() {
        let (sender, receiver) = unbounded_channel();
        let redraw_count = Arc::new(AtomicUsize::new(0));
        let handler = CountingHandler {
            redraw_count: redraw_count.clone(),
        };

        let frame = RedrawFrame::from_slice(&redraw_bytes()).unwrap();
        sender.send(frame).unwrap();
        drop(sender);

        test_neovim::<CountingHandler>()
            .handler_loop(handler, receiver)
            .await
            .unwrap();

        assert_eq!(redraw_count.load(Ordering::Relaxed), 1);
    }

    #[tokio::test]
    async fn notify_writes_value_ref_notification_without_queueing_request() {
        let nvim = test_neovim::<CountingHandler>();
        let arg = ValueRef::Boolean(true);

        nvim.notify("nvim_ui_set_focus", &arg).await.unwrap();

        assert!(nvim.inner.queue.lock().is_empty());

        let writer = nvim.inner.encoder.lock().await;
        assert_eq!(
            writer.writer(),
            &encoded_value(Value::from(vec![
                Value::from(2),
                Value::from("nvim_ui_set_focus"),
                Value::from(vec![Value::from(true)]),
            ]))
        );
    }

    #[tokio::test]
    async fn notify_input_writes_notification_without_queueing_request() {
        let nvim = test_neovim::<CountingHandler>();

        nvim.notify_input("<C-D>").await.unwrap();

        assert!(nvim.inner.queue.lock().is_empty());

        let writer = nvim.inner.encoder.lock().await;
        assert_eq!(
            writer.writer(),
            &encoded_value(Value::from(vec![
                Value::from(2),
                Value::from("nvim_input"),
                Value::from(vec![Value::from("<C-D>")]),
            ]))
        );
    }

    #[tokio::test]
    async fn ui_try_resize_writes_notification_without_queueing_request() {
        let nvim = test_neovim::<CountingHandler>();

        nvim.ui_try_resize(120, 40).await.unwrap();

        assert!(nvim.inner.queue.lock().is_empty());

        let writer = nvim.inner.encoder.lock().await;
        assert_eq!(
            writer.writer(),
            &encoded_value(Value::from(vec![
                Value::from(2),
                Value::from("nvim_ui_try_resize"),
                Value::from(vec![Value::from(120), Value::from(40)]),
            ]))
        );
    }

    #[tokio::test]
    async fn ui_set_focus_writes_notification_without_queueing_request() {
        let nvim = test_neovim::<CountingHandler>();

        nvim.ui_set_focus(false).await.unwrap();

        assert!(nvim.inner.queue.lock().is_empty());

        let writer = nvim.inner.encoder.lock().await;
        assert_eq!(
            writer.writer(),
            &encoded_value(Value::from(vec![
                Value::from(2),
                Value::from("nvim_ui_set_focus"),
                Value::from(vec![Value::from(false)]),
            ]))
        );
    }

    #[tokio::test]
    async fn request_input_writes_request_and_returns_written_byte_count() {
        let (mut remote, reader) = tokio::io::duplex(1024);
        let (nvim, io) = Neovim::new(reader, Vec::new(), Dummy::<Vec<u8>>::new());
        let inner = nvim.inner.clone();
        let io_handle = tokio::spawn(io);

        let request_handle = tokio::spawn({
            let nvim = nvim.clone();
            async move { nvim.request_input("<Cmd>qall<CR>").await }
        });

        wait_for_pending_request(&inner.queue).await;
        remote
            .write_all(&response_bytes(0, Value::from(14), Value::Nil))
            .await
            .unwrap();

        assert_eq!(request_handle.await.unwrap().unwrap(), 14);
        io_handle.abort();

        let writer = inner.encoder.lock().await;
        assert_eq!(
            writer.writer(),
            &request_bytes(
                0,
                "nvim_input",
                Value::from(vec![Value::from("<Cmd>qall<CR>")])
            )
        );
    }

    #[tokio::test]
    async fn pending_request_receives_io_error_from_io_loop() {
        let (mut remote, reader) = tokio::io::duplex(1024);
        let (nvim, io) = Neovim::new(reader, Vec::new(), Dummy::<Vec<u8>>::new());
        let inner = nvim.inner.clone();
        let io_handle = tokio::spawn(io);

        let request_handle = tokio::spawn({
            let nvim = nvim.clone();
            async move { nvim.request_input("<C-D>").await }
        });

        wait_for_pending_request(&inner.queue).await;
        remote
            .write_all(&encoded_value(Value::from(vec![Value::from(3)])))
            .await
            .unwrap();

        let err = request_handle.await.unwrap().unwrap_err();
        assert_eq!(err.kind(), ErrorKind::InvalidData);

        let err = io_handle.await.unwrap().unwrap_err();
        assert_eq!(err.kind(), ErrorKind::InvalidData);

        assert!(inner.queue.lock().is_empty());
    }

    #[tokio::test]
    async fn ui_attach_writes_request_and_waits_for_success_response() {
        let (mut remote, reader) = tokio::io::duplex(1024);
        let (nvim, io) = Neovim::new(reader, Vec::new(), Dummy::<Vec<u8>>::new());
        let inner = nvim.inner.clone();
        let io_handle = tokio::spawn(io);

        let request_handle = tokio::spawn({
            let nvim = nvim.clone();
            async move {
                let mut options = UiAttachOptions::default();
                options.set_rgb(true);
                options.set_linegrid_external(true);
                options.set_hlstate_external(true);
                nvim.ui_attach(120, 40, options).await
            }
        });

        wait_for_pending_request(&inner.queue).await;
        remote
            .write_all(&response_bytes(0, Value::Nil, Value::Nil))
            .await
            .unwrap();

        request_handle.await.unwrap().unwrap();
        io_handle.abort();

        let writer = inner.encoder.lock().await;
        assert_eq!(
            writer.writer(),
            &request_bytes(
                0,
                "nvim_ui_attach",
                Value::from(vec![
                    Value::from(120),
                    Value::from(40),
                    Value::Map(vec![
                        (Value::from("rgb"), Value::from(true)),
                        (Value::from("ext_linegrid"), Value::from(true)),
                        (Value::from("ext_hlstate"), Value::from(true)),
                    ]),
                ])
            )
        );
    }

    #[tokio::test]
    async fn cmd_writes_nvim_cmd_request_and_returns_output() {
        let (mut remote, reader) = tokio::io::duplex(1024);
        let (nvim, io) = Neovim::new(reader, Vec::new(), Dummy::<Vec<u8>>::new());
        let inner = nvim.inner.clone();
        let io_handle = tokio::spawn(io);

        let request_handle = tokio::spawn({
            let nvim = nvim.clone();
            async move {
                let command = ValueRef::Map(vec![
                    (ValueRef::from("cmd"), ValueRef::from("edit")),
                    (
                        ValueRef::from("args"),
                        ValueRef::Array(vec![ValueRef::from("file.txt")]),
                    ),
                    (
                        ValueRef::from("magic"),
                        ValueRef::Map(vec![
                            (ValueRef::from("file"), ValueRef::Boolean(false)),
                            (ValueRef::from("bar"), ValueRef::Boolean(false)),
                        ]),
                    ),
                ]);
                nvim.cmd(command, ValueRef::Map(vec![])).await
            }
        });

        wait_for_pending_request(&inner.queue).await;
        remote
            .write_all(&response_bytes(0, Value::from("opened"), Value::Nil))
            .await
            .unwrap();

        assert_eq!(request_handle.await.unwrap().unwrap(), "opened");
        io_handle.abort();

        let writer = inner.encoder.lock().await;
        assert_eq!(
            writer.writer(),
            &request_bytes(
                0,
                "nvim_cmd",
                Value::from(vec![
                    Value::Map(vec![
                        (Value::from("cmd"), Value::from("edit")),
                        (
                            Value::from("args"),
                            Value::Array(vec![Value::from("file.txt")])
                        ),
                        (
                            Value::from("magic"),
                            Value::Map(vec![
                                (Value::from("file"), Value::from(false)),
                                (Value::from("bar"), Value::from(false)),
                            ])
                        ),
                    ]),
                    Value::Map(vec![]),
                ])
            )
        );
    }

    #[tokio::test]
    async fn cmd_returns_neovim_error_response() {
        let (mut remote, reader) = tokio::io::duplex(1024);
        let (nvim, io) = Neovim::new(reader, Vec::new(), Dummy::<Vec<u8>>::new());
        let inner = nvim.inner.clone();
        let io_handle = tokio::spawn(io);

        let request_handle = tokio::spawn({
            let nvim = nvim.clone();
            async move {
                nvim.cmd(
                    ValueRef::Map(vec![(ValueRef::from("cmd"), ValueRef::from("edit"))]),
                    ValueRef::Map(vec![]),
                )
                .await
            }
        });

        wait_for_pending_request(&inner.queue).await;
        remote
            .write_all(&response_bytes(
                0,
                Value::Nil,
                Value::Array(vec![Value::from(1), Value::from("failed")]),
            ))
            .await
            .unwrap();

        let err = request_handle.await.unwrap().unwrap_err();
        let err = err
            .get_ref()
            .and_then(|err| err.downcast_ref::<NeovimError>())
            .expect("expected NeovimError");
        assert_eq!(err.error_type, Some(1));
        assert_eq!(err.message, Value::from("failed"));
        io_handle.abort();
    }

    #[tokio::test]
    async fn test_find_sender() {
        let queue = SyncMutex::new(Vec::new());

        {
            let (sender, _receiver) = oneshot::channel();
            queue.lock().push((1, sender));
        }
        {
            let (sender, _receiver) = oneshot::channel();
            queue.lock().push((2, sender));
        }
        {
            let (sender, _receiver) = oneshot::channel();
            queue.lock().push((3, sender));
        }

        find_sender(&queue, 1).unwrap();
        assert_eq!(2, queue.lock().len());
        find_sender(&queue, 2).unwrap();
        assert_eq!(1, queue.lock().len());
        find_sender(&queue, 3).unwrap();
        assert!(queue.lock().is_empty());

        let err = find_sender(&queue, 17).unwrap_err();
        let err = err
            .get_ref()
            .and_then(|err| err.downcast_ref::<MessageIdNotFound>())
            .expect("expected MessageIdNotFound");
        assert_eq!(err.msgid, 17);
    }
}