navy-nvim-rs 0.0.23

use std::{
    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, ResponseReceiveError},
    rpc::{
        decode::Decoder,
        encode::{EncodeArgs, Encoder},
        handler::Handler,
        payload::{NotificationPayload, Payload, RequestPayload, ResponsePayload},
    },
};

type RequestCallback = oneshot::Sender<ResponsePayload>;

#[derive(Default)]
struct PendingRequests {
    // `Vec` is okay here because its size is 1 in most cases.
    callbacks: SyncMutex<Vec<(u32, RequestCallback)>>,
}

impl PendingRequests {
    #[inline]
    fn clear(&self) {
        self.callbacks.lock().clear();
    }

    #[inline]
    fn add(&self, id: u32, callback: RequestCallback) {
        self.callbacks.lock().push((id, callback));
    }

    #[inline]
    fn take_callback(&self, msgid: u32) -> io::Result<RequestCallback> {
        let mut callbacks = self.callbacks.lock();
        if let Some(pos) = callbacks.iter().position(|(id, _)| *id == msgid) {
            Ok(callbacks.remove(pos).1)
        } else {
            Err(Error::other(MessageIdNotFound::new(msgid)))
        }
    }
}

enum HandlerMessage {
    Notification(NotificationPayload),
    Request(RequestPayload),
}

/// An active Neovim session.
pub(crate) struct NeovimInner<W> {
    encoder: Mutex<Encoder<W>>,
    pending: PendingRequests,
    next_msgid: AtomicU32,
}

pub struct Neovim<W> {
    pub(crate) inner: Arc<NeovimInner<W>>,
}

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

impl<W> Neovim<W>
where
    W: AsyncWrite + Send + Unpin + 'static,
{
    pub fn new<R, H>(
        reader: R,
        writer: W,
        handler: H,
    ) -> (Self, impl Future<Output = io::Result<()>>)
    where
        R: AsyncRead + Send + Unpin + 'static,
        H: Handler,
    {
        let neovim = Neovim {
            inner: Arc::new(NeovimInner {
                encoder: Mutex::new(Encoder::new(writer)),
                pending: PendingRequests::default(),
                next_msgid: AtomicU32::new(0),
            }),
        };

        let (handler_tx, handler_rx) = unbounded_channel();
        let io = {
            let neovim = neovim.clone();
            async move {
                let result = tokio::try_join!(
                    neovim.io_loop(reader, handler_tx),
                    neovim.handler_loop(handler, handler_rx),
                )
                .map(|_| ());
                neovim.inner.pending.clear();
                result
            }
        };

        (neovim, io)
    }

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

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

        match receiver.await {
            Ok(response) => Ok(response),
            Err(err) => Err(Error::other(ResponseReceiveError::new(method, err))),
        }
    }

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

    async fn handler_loop<H: Handler>(
        &self,
        handler: H,
        mut receiver: UnboundedReceiver<HandlerMessage>,
    ) -> 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
                 */
                return Ok(());
            };

            match msg {
                HandlerMessage::Notification(payload) => handler.handle_notify(payload)?,
                HandlerMessage::Request(payload) => {
                    let msgid = payload.message_id();
                    let response = handler.handle_request(payload)?;
                    let mut encoder = self.inner.encoder.lock().await;
                    match response {
                        Ok(result) => encoder.encode_result_response(msgid, &result).await?,
                        Err(error) => encoder.encode_error_response(msgid, &error).await?,
                    }
                }
            }
        }
    }

    async fn io_loop<R>(
        &self,
        mut reader: R,
        sender: UnboundedSender<HandlerMessage>,
    ) -> io::Result<()>
    where
        R: AsyncRead + Send + Unpin + 'static,
    {
        let mut decoder = Decoder::default();
        loop {
            match decoder.decode_payload()? {
                Some(Payload::Notification(payload)) => sender
                    .send(HandlerMessage::Notification(payload))
                    .map_err(|_| Error::other("Could not send notification because RPC handler channel has been closed"))?,
                Some(Payload::Request(payload)) => sender
                    .send(HandlerMessage::Request(payload))
                    .map_err(|_| Error::other("Could not send request because RPC handler channel has been closed"))?,
                Some(Payload::Response(payload)) => {
                    let response_sender = self.inner.pending.take_callback(payload.message_id())?;
                    // The request future may have been cancelled after sending the request.
                    let _ = response_sender.send(payload);
                }
                None => {
                    if !decoder.read_next_chunk(&mut reader).await? {
                        return Ok(());
                    }
                }
            }
        }
    }
}

impl Neovim<WriteHalf<TcpStream>> {
    /// Connect to a Neovim instance via TCP.
    pub async fn connect<A, H>(
        addr: A,
        handler: H,
    ) -> io::Result<(Self, JoinHandle<io::Result<()>>)>
    where
        A: ToSocketAddrs,
        H: Handler,
    {
        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))
    }
}

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

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

        Ok((neovim, io_handle, child))
    }
}

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

    use rmpv::{Value, ValueRef, encode::write_value};
    use tokio::{
        io::{AsyncReadExt, AsyncWriteExt, DuplexStream},
        time::timeout,
    };

    use super::*;
    use crate::rpc::handler::{Dummy, ResponseValue};
    use crate::{UiAttachOptions, error::NeovimError};

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

    #[derive(Clone)]
    struct FailingHandler;

    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 {
        fn handle_notify(&self, payload: NotificationPayload) -> io::Result<()> {
            match payload.method()? {
                "redraw" => {
                    let mut batches = payload.params();
                    while !batches.is_empty() {
                        batches.read_array(|batch| {
                            assert_eq!(batch.read::<&str>()?, "flush");
                            assert!(batch.is_empty());
                            Ok(())
                        })?;
                    }
                    self.redraw_count.fetch_add(1, Ordering::Relaxed);
                }
                "ignored" => {
                    UNKNOWN_NOTIFICATION_COUNT.fetch_add(1, Ordering::Relaxed);
                }
                method => panic!("unexpected notification: {method}"),
            }
            Ok(())
        }

        fn handle_request(&self, payload: RequestPayload) -> io::Result<ResponseValue<'_>> {
            assert_eq!(payload.method()?, "ignored");
            UNKNOWN_REQUEST_MSGID.store(payload.message_id(), Ordering::Relaxed);
            UNKNOWN_REQUEST_COUNT.fetch_add(1, Ordering::Relaxed);
            Ok(Ok(ValueRef::from("handled")))
        }
    }

    impl Handler for ChildProcessHandler {
        fn handle_notify(&self, payload: NotificationPayload) -> io::Result<()> {
            if payload.method()? != "redraw" {
                return Ok(());
            }

            let mut batches = payload.params();
            let _ = self.redraw_tx.send(RedrawSignal::Any);

            while !batches.is_empty() {
                batches.read_array(|batch| {
                    if batch.read::<&str>()? == "grid_resize" {
                        while !batch.is_empty() {
                            batch.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(())
                })?;
            }
            Ok(())
        }

        fn handle_request(&self, _payload: RequestPayload) -> io::Result<ResponseValue<'_>> {
            unreachable!()
        }
    }

    impl Handler for FailingHandler {
        fn handle_notify(&self, _payload: NotificationPayload) -> io::Result<()> {
            Err(ErrorKind::InvalidData.into())
        }

        fn handle_request(&self, _payload: RequestPayload) -> io::Result<ResponseValue<'_>> {
            unreachable!()
        }
    }

    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 {
        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: impl Into<Value>) -> Vec<u8> {
        let mut bytes = Vec::new();
        write_value(&mut bytes, &value.into()).unwrap();
        bytes
    }

    fn request_bytes(msgid: u32, method: &str, params: impl Into<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.into()).unwrap();
        bytes
    }

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

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

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

    fn response_bytes(msgid: u32, result: impl Into<Value>, error: impl Into<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.into()).unwrap();
        write_value(&mut bytes, &result.into()).unwrap();
        bytes
    }

    fn connected_neovim<H: Handler>(
        handler: H,
    ) -> (
        Neovim<WriteHalf<DuplexStream>>,
        DuplexStream,
        JoinHandle<io::Result<()>>,
    ) {
        let (client, remote) = tokio::io::duplex(1024);
        let (reader, writer) = split(client);
        let (neovim, io) = Neovim::new(reader, writer, handler);
        (neovim, remote, tokio::spawn(io))
    }

    async fn expect_rpc_message(remote: &mut DuplexStream, expected: &[u8]) {
        let mut actual = vec![0; expected.len()];
        remote.read_exact(&mut actual).await.unwrap();
        assert_eq!(actual, expected);
    }

    async fn respond_to_rpc(remote: &mut DuplexStream, expected: &[u8], response: &[u8]) {
        expect_rpc_message(remote, expected).await;
        remote.write_all(response).await.unwrap();
    }

    async fn close_connection(remote: DuplexStream, io_handle: JoinHandle<io::Result<()>>) {
        drop(remote);
        io_handle.await.unwrap().unwrap();
    }

    fn test_neovim() -> Neovim<Vec<u8>> {
        Neovim {
            inner: Arc::new(NeovimInner {
                encoder: Mutex::new(Encoder::new(Vec::new())),
                pending: PendingRequests::default(),
                next_msgid: 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);
        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 = 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 u32);

        let status = match 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}");
        timeout(Duration::from_secs(5), io_handle)
            .await
            .expect("timed out waiting for io loop")
            .unwrap()
            .unwrap();
    }

    #[tokio::test]
    async fn spawn_reports_channel_close_to_request_and_io_loop() {
        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 = 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 = 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::Other, "unexpected error: {err}");
        assert!(err.get_ref().unwrap().is::<ResponseReceiveError>());

        timeout(Duration::from_secs(5), io_handle)
            .await
            .expect("timed out waiting for io loop")
            .unwrap()
            .unwrap();

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

    #[tokio::test]
    async fn connect_connects_to_nvim_over_tcp() {
        let listener = std::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap();
        let addr = listener.local_addr().unwrap();
        drop(listener);

        let path = nvim_test_bin();
        let listen = addr.to_string();
        let mut cmd = Command::new(path);
        cmd.args(["-u", "NONE", "--headless", "--listen", &listen])
            .kill_on_drop(true)
            .stderr(Stdio::piped());
        let mut child = cmd.spawn().unwrap();

        let (nvim, io_handle) = timeout(Duration::from_secs(5), async {
            loop {
                match Neovim::connect(addr, Dummy).await {
                    Ok(connection) => break connection,
                    Err(_) => tokio::time::sleep(Duration::from_millis(10)).await,
                }
            }
        })
        .await
        .expect("timed out connecting to nvim over TCP");

        let servername = timeout(
            Duration::from_secs(5),
            nvim.request("nvim_get_vvar", &ValueRef::from("servername")),
        )
        .await
        .expect("timed out waiting for nvim_get_vvar response")
        .unwrap();
        assert_eq!(servername.decode_result::<&str>().unwrap(), listen);

        let quit_input = "<Cmd>qall!<CR>";
        let written = 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 u32);

        let status = match 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}");
        timeout(Duration::from_secs(5), io_handle)
            .await
            .expect("timed out waiting for io loop")
            .unwrap()
            .unwrap();
    }

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

        let (sender, receiver) = unbounded_channel();
        let redraw_count = Arc::new(AtomicUsize::new(0));
        let handler = CountingHandler {
            redraw_count: redraw_count.clone(),
        };

        let mut bytes = bytes::BytesMut::from(ignored_notification_bytes().as_slice());
        let Payload::Notification(notification) = Payload::decode(&mut bytes).unwrap() else {
            panic!("expected notification");
        };
        sender
            .send(HandlerMessage::Notification(notification))
            .unwrap();

        let mut bytes = bytes::BytesMut::from(ignored_request_bytes().as_slice());
        let Payload::Request(request) = Payload::decode(&mut bytes).unwrap() else {
            panic!("expected request");
        };
        sender.send(HandlerMessage::Request(request)).unwrap();

        let mut bytes = bytes::BytesMut::from(redraw_bytes().as_slice());
        let Payload::Notification(redraw) = Payload::decode(&mut bytes).unwrap() else {
            panic!("expected redraw notification");
        };
        sender.send(HandlerMessage::Notification(redraw)).unwrap();
        drop(sender);

        let nvim = test_neovim();
        let inner = nvim.inner.clone();
        nvim.handler_loop(handler, receiver).await.unwrap();

        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);
        assert_eq!(redraw_count.load(Ordering::Relaxed), 1);
        let response = response_bytes(99, "handled", Value::Nil);
        assert_eq!(inner.encoder.lock().await.writer(), &response);
    }

    #[tokio::test]
    async fn io_loop_reports_closed_handler_channel_as_internal_error() {
        let (mut remote, reader) = tokio::io::duplex(1024);
        let (sender, receiver) = unbounded_channel();
        drop(receiver);
        let io_handle = tokio::spawn(async move { test_neovim().io_loop(reader, sender).await });

        remote
            .write_all(&ignored_notification_bytes())
            .await
            .unwrap();

        let error = io_handle.await.unwrap().unwrap_err();
        assert_eq!(error.kind(), ErrorKind::Other);
        assert_eq!(
            error.to_string(),
            "Could not send notification because RPC handler channel has been closed"
        );
    }

    #[tokio::test]
    async fn notify_writes_value_ref_notification() {
        let (nvim, mut remote, io_handle) = connected_neovim(Dummy);
        let arg = ValueRef::Boolean(true);
        let expected = encoded_value(vec![
            Value::from(2),
            Value::from("nvim_ui_set_focus"),
            Value::from(vec![Value::from(true)]),
        ]);

        let (result, ()) = tokio::join!(
            nvim.notify("nvim_ui_set_focus", &arg),
            expect_rpc_message(&mut remote, &expected),
        );

        result.unwrap();
        close_connection(remote, io_handle).await;
    }

    #[tokio::test]
    async fn notify_input_writes_notification() {
        let (nvim, mut remote, io_handle) = connected_neovim(Dummy);
        let expected = encoded_value(vec![
            Value::from(2),
            Value::from("nvim_input"),
            Value::from(vec![Value::from("<C-D>")]),
        ]);

        let (result, ()) = tokio::join!(
            nvim.notify_input("<C-D>"),
            expect_rpc_message(&mut remote, &expected),
        );

        result.unwrap();
        close_connection(remote, io_handle).await;
    }

    #[tokio::test]
    async fn ui_try_resize_writes_notification() {
        let (nvim, mut remote, io_handle) = connected_neovim(Dummy);
        let expected = encoded_value(vec![
            Value::from(2),
            Value::from("nvim_ui_try_resize"),
            Value::from(vec![Value::from(120), Value::from(40)]),
        ]);

        let (result, ()) = tokio::join!(
            nvim.ui_try_resize(120, 40),
            expect_rpc_message(&mut remote, &expected),
        );

        result.unwrap();
        close_connection(remote, io_handle).await;
    }

    #[tokio::test]
    async fn ui_set_focus_writes_notification() {
        let (nvim, mut remote, io_handle) = connected_neovim(Dummy);
        let expected = encoded_value(vec![
            Value::from(2),
            Value::from("nvim_ui_set_focus"),
            Value::from(vec![Value::from(false)]),
        ]);

        let (result, ()) = tokio::join!(
            nvim.ui_set_focus(false),
            expect_rpc_message(&mut remote, &expected),
        );

        result.unwrap();
        close_connection(remote, io_handle).await;
    }

    #[tokio::test]
    async fn request_input_writes_request_and_returns_written_byte_count() {
        let (nvim, mut remote, io_handle) = connected_neovim(Dummy);
        let request = request_bytes(0, "nvim_input", vec![Value::from("<Cmd>qall<CR>")]);
        let response = response_bytes(0, 14, Value::Nil);

        let (result, ()) = tokio::join!(
            nvim.request_input("<Cmd>qall<CR>"),
            respond_to_rpc(&mut remote, &request, &response),
        );

        assert_eq!(result.unwrap(), 14);
        close_connection(remote, io_handle).await;
    }

    #[tokio::test]
    async fn pending_request_closes_when_io_loop_fails() {
        let (nvim, mut remote, io_handle) = connected_neovim(Dummy);
        let request = request_bytes(0, "nvim_input", vec![Value::from("<C-D>")]);
        let invalid_response = encoded_value(vec![Value::from(3)]);

        let (result, ()) = tokio::join!(
            nvim.request_input("<C-D>"),
            respond_to_rpc(&mut remote, &request, &invalid_response),
        );

        let err = result.unwrap_err();
        assert_eq!(err.kind(), ErrorKind::Other);
        assert!(err.get_ref().unwrap().is::<ResponseReceiveError>());

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

    #[tokio::test]
    async fn pending_request_closes_when_io_loop_reaches_eof() {
        let (nvim, mut remote, io_handle) = connected_neovim(Dummy);
        let request = request_bytes(0, "nvim_input", vec![Value::from("<C-D>")]);

        let (result, ()) = tokio::join!(nvim.request_input("<C-D>"), async move {
            expect_rpc_message(&mut remote, &request).await;
        });

        io_handle.await.unwrap().unwrap();

        let err = result.unwrap_err();
        assert_eq!(err.kind(), ErrorKind::Other);
        assert!(err.get_ref().unwrap().is::<ResponseReceiveError>());
    }

    #[tokio::test]
    async fn pending_request_closes_when_handler_fails() {
        let (nvim, mut remote, io_handle) = connected_neovim(FailingHandler);
        let request = request_bytes(0, "nvim_input", vec![Value::from("<C-D>")]);
        let notification = ignored_notification_bytes();

        let (result, ()) = tokio::join!(
            nvim.request_input("<C-D>"),
            respond_to_rpc(&mut remote, &request, &notification),
        );

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

        let err = result.unwrap_err();
        assert_eq!(err.kind(), ErrorKind::Other);
        assert!(err.get_ref().unwrap().is::<ResponseReceiveError>());
    }

    #[tokio::test]
    async fn ui_attach_writes_request_and_waits_for_success_response() {
        let (nvim, mut remote, io_handle) = connected_neovim(Dummy);
        let request = request_bytes(
            0,
            "nvim_ui_attach",
            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)),
                ]),
            ],
        );
        let response = response_bytes(0, Value::Nil, Value::Nil);
        let mut options = UiAttachOptions::default();
        options.set_rgb(true);
        options.set_linegrid_external(true);
        options.set_hlstate_external(true);

        let (result, ()) = tokio::join!(
            nvim.ui_attach(120, 40, options),
            respond_to_rpc(&mut remote, &request, &response),
        );

        result.unwrap();
        close_connection(remote, io_handle).await;
    }

    #[tokio::test]
    async fn cmd_writes_nvim_cmd_request_and_returns_output() {
        let (nvim, mut remote, io_handle) = connected_neovim(Dummy);
        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)),
                ]),
            ),
        ]);
        let request = request_bytes(
            0,
            "nvim_cmd",
            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![]),
            ],
        );
        let response = response_bytes(0, "opened", Value::Nil);

        let (result, ()) = tokio::join!(
            nvim.cmd(command, ValueRef::Map(vec![])),
            respond_to_rpc(&mut remote, &request, &response),
        );

        assert_eq!(result.unwrap(), "opened");
        close_connection(remote, io_handle).await;
    }

    #[tokio::test]
    async fn cmd_returns_neovim_error_response() {
        let (nvim, mut remote, io_handle) = connected_neovim(Dummy);
        let request = request_bytes(
            0,
            "nvim_cmd",
            vec![
                Value::Map(vec![(Value::from("cmd"), Value::from("edit"))]),
                Value::Map(vec![]),
            ],
        );
        let response = response_bytes(0, Value::Nil, vec![Value::from(1), Value::from("failed")]);

        let (result, ()) = tokio::join!(
            nvim.cmd(
                ValueRef::Map(vec![(ValueRef::from("cmd"), ValueRef::from("edit"))]),
                ValueRef::Map(vec![]),
            ),
            respond_to_rpc(&mut remote, &request, &response),
        );

        let err = result.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"));
        close_connection(remote, io_handle).await;
    }

    #[test]
    fn pending_requests_takes_callback_by_message_id() {
        let pending = PendingRequests::default();
        let (callback1, _receiver1) = oneshot::channel();
        let (callback2, _receiver2) = oneshot::channel();
        let (callback3, _receiver3) = oneshot::channel();
        pending.add(1, callback1);
        pending.add(2, callback2);
        pending.add(3, callback3);

        pending.take_callback(2).unwrap();
        let ids = pending
            .callbacks
            .lock()
            .iter()
            .map(|(id, _)| *id)
            .collect::<Vec<_>>();
        assert_eq!(ids, [1, 3]);
    }

    #[test]
    fn pending_requests_reports_unknown_message_id() {
        let err = PendingRequests::default().take_callback(17).unwrap_err();
        let err = err
            .get_ref()
            .and_then(|err| err.downcast_ref::<MessageIdNotFound>())
            .expect("expected MessageIdNotFound");
        assert_eq!(err.msgid, 17);
    }

    #[tokio::test]
    async fn pending_requests_clear_closes_all_callbacks() {
        let pending = PendingRequests::default();
        let (callback1, receiver1) = oneshot::channel();
        let (callback2, receiver2) = oneshot::channel();
        pending.add(1, callback1);
        pending.add(2, callback2);

        pending.clear();

        assert!(receiver1.await.is_err());
        assert!(receiver2.await.is_err());
        assert!(pending.callbacks.lock().is_empty());
    }
}