Bridge unix sockets and named pipes

src/bridge/handler.rs

@@ -7,7 +7,7 @@ use crate::bridge::clipboard::{get_clipboard_contents, set_clipboard_contents};
 #[cfg(windows)]
 use crate::bridge::ui_commands::{ParallelCommand, UiCommand};
 use crate::{
-    bridge::{events::parse_redraw_event, TxWrapper},
+    bridge::{events::parse_redraw_event, NeovimWriter},
     editor::EditorCommand,
     error_handling::ResultPanicExplanation,
     event_aggregator::EVENT_AGGREGATOR,
@@ -26,13 +26,13 @@ impl NeovimHandler {
 
 #[async_trait]
 impl Handler for NeovimHandler {
-    type Writer = TxWrapper;
+    type Writer = NeovimWriter;
 
     async fn handle_request(
         &self,
         event_name: String,
         arguments: Vec<Value>,
-        neovim: Neovim<TxWrapper>,
+        neovim: Neovim<Self::Writer>,
     ) -> Result<Value, Value> {
         trace!("Neovim request: {:?}", &event_name);
 
@@ -61,7 +61,7 @@ impl Handler for NeovimHandler {
         &self,
         event_name: String,
         arguments: Vec<Value>,
-        _neovim: Neovim<TxWrapper>,
+        _neovim: Neovim<Self::Writer>,
     ) {
         trace!("Neovim notification: {:?}", &event_name);
 

src/bridge/mod.rs

@@ -1,10 +1,9 @@
 mod clipboard;
 mod command;
-pub mod create;
 mod events;
 mod handler;
+pub mod session;
 mod setup;
-mod tx_wrapper;
 mod ui_commands;
 
 use std::{process::exit, sync::Arc, thread};
@@ -20,20 +19,16 @@ use crate::{
 pub use command::create_nvim_command;
 pub use events::*;
 use handler::NeovimHandler;
+pub use session::NeovimWriter;
+use session::{NeovimInstance, NeovimSession};
 use setup::setup_neovide_specific_state;
-pub use tx_wrapper::{TxWrapper, WrapTx};
 pub use ui_commands::{start_ui_command_handler, ParallelCommand, SerialCommand, UiCommand};
 
-enum ConnectionMode {
-    Child,
-    RemoteTcp(String),
-}
-
-fn connection_mode() -> ConnectionMode {
-    if let Some(arg) = SETTINGS.get::<CmdLineSettings>().remote_tcp {
-        ConnectionMode::RemoteTcp(arg)
+fn neovim_instance() -> NeovimInstance {
+    if let Some(address) = SETTINGS.get::<CmdLineSettings>().server {
+        NeovimInstance::Server { address }
     } else {
-        ConnectionMode::Child
+        NeovimInstance::Embedded(create_nvim_command())
     }
 }
 
@@ -46,11 +41,11 @@ pub fn start_bridge() {
 #[tokio::main]
 async fn start_neovim_runtime() {
     let handler = NeovimHandler::new();
-    let (nvim, io_handler) = match connection_mode() {
-        ConnectionMode::Child => create::new_child_cmd(&mut create_nvim_command(), handler).await,
-        ConnectionMode::RemoteTcp(address) => create::new_tcp(address, handler).await,
-    }
-    .unwrap_or_explained_panic("Could not locate or start neovim process");
+    let session = NeovimSession::new(neovim_instance(), handler)
+        .await
+        .unwrap_or_explained_panic("Could not locate or start neovim process");
+
+    let nvim = Arc::new(session.neovim);
 
     // Check the neovim version to ensure its high enough
     match nvim.command_output("echo has('nvim-0.4')").await.as_deref() {
@@ -63,11 +58,8 @@ async fn start_neovim_runtime() {
 
     let settings = SETTINGS.get::<CmdLineSettings>();
 
-    let mut is_remote = settings.wsl;
-    if let ConnectionMode::RemoteTcp(_) = connection_mode() {
-        is_remote = true;
-    }
-    setup_neovide_specific_state(&nvim, is_remote).await;
+    let should_handle_clipboard = settings.wsl || settings.server.is_some();
+    setup_neovide_specific_state(&nvim, should_handle_clipboard).await;
 
     let geometry = settings.geometry;
     let mut options = UiAttachOptions::new();
@@ -82,13 +74,11 @@ async fn start_neovim_runtime() {
 
     info!("Neovim process attached");
 
-    let nvim = Arc::new(nvim);
-
     start_ui_command_handler(nvim.clone());
     SETTINGS.read_initial_values(&nvim).await;
     SETTINGS.setup_changed_listeners(&nvim).await;
 
-    match io_handler.await {
+    match session.io_handle.await {
         Err(join_error) => error!("Error joining IO loop: '{}'", join_error),
         Ok(Err(error)) => {
             if !error.is_channel_closed() {

src/bridge/session.rs

@@ -0,0 +1,117 @@
+//! This module contains adaptations of the functions found in
+//! https://github.com/KillTheMule/nvim-rs/blob/master/src/create/tokio.rs
+
+use std::{
+    io::{Error, ErrorKind, Result},
+    process::Stdio,
+};
+
+use nvim_rs::{error::LoopError, neovim::Neovim, Handler};
+use tokio::{
+    io::{split, AsyncRead, AsyncWrite},
+    net::TcpStream,
+    process::Command,
+    spawn,
+    task::JoinHandle,
+};
+use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};
+
+pub type NeovimWriter = Box<dyn futures::AsyncWrite + Send + Unpin + 'static>;
+
+type BoxedReader = Box<dyn AsyncRead + Send + Unpin + 'static>;
+type BoxedWriter = Box<dyn AsyncWrite + Send + Unpin + 'static>;
+
+pub struct NeovimSession {
+    pub neovim: Neovim<NeovimWriter>,
+    pub io_handle: JoinHandle<std::result::Result<(), Box<LoopError>>>,
+}
+
+impl NeovimSession {
+    pub async fn new(
+        instance: NeovimInstance,
+        handler: impl Handler<Writer = NeovimWriter>,
+    ) -> Result<Self> {
+        let (reader, writer) = instance.connect().await?;
+        let (neovim, io) =
+            Neovim::<NeovimWriter>::new(reader.compat(), Box::new(writer.compat_write()), handler);
+        let io_handle = spawn(io);
+
+        Ok(Self { neovim, io_handle })
+    }
+}
+
+/// An existing or future Neovim instance along with a means for establishing a connection.
+#[derive(Debug)]
+pub enum NeovimInstance {
+    /// A new embedded instance to be spawned by the given command.
+    Embedded(Command),
+
+    /// An existing instance listening on `address`.
+    ///
+    /// Interprets `address` in the same way as `:help --server`: If it contains a `:` it's
+    /// interpreted as a TCP/IPv4/IPv6 address. Otherwise it's interpreted as a named pipe or Unix
+    /// domain socket path. Spawns and connects to an embedded Neovim instance.
+    Server { address: String },
+}
+
+impl NeovimInstance {
+    async fn connect(self) -> Result<(BoxedReader, BoxedWriter)> {
+        match self {
+            NeovimInstance::Embedded(cmd) => Self::spawn_process(cmd).await,
+            NeovimInstance::Server { address } => Self::connect_to_server(address).await,
+        }
+    }
+
+    async fn spawn_process(mut cmd: Command) -> Result<(BoxedReader, BoxedWriter)> {
+        let mut child = cmd.stdin(Stdio::piped()).stdout(Stdio::piped()).spawn()?;
+        let reader = Box::new(
+            child
+                .stdout
+                .take()
+                .ok_or_else(|| Error::new(ErrorKind::Other, "Can't open stdout"))?,
+        );
+        let writer = Box::new(
+            child
+                .stdin
+                .take()
+                .ok_or_else(|| Error::new(ErrorKind::Other, "Can't open stdin"))?,
+        );
+
+        Ok((reader, writer))
+    }
+
+    async fn connect_to_server(address: String) -> Result<(BoxedReader, BoxedWriter)> {
+        if address.contains(':') {
+            Ok(Self::split(TcpStream::connect(address).await?))
+        } else {
+            #[cfg(unix)]
+            return Ok(Self::split(tokio::net::UnixStream::connect(address).await?));
+
+            #[cfg(windows)]
+            {
+                // Fixup the address if the pipe on windows does not start with \\.\pipe\.
+                let address = if address.starts_with("\\\\.\\pipe\\") {
+                    address
+                } else {
+                    format!("\\\\.\\pipe\\{}", address)
+                };
+                return Ok(Self::split(
+                    tokio::net::windows::named_pipe::ClientOptions::new().open(address)?,
+                ));
+            }
+
+            #[cfg(not(any(unix, windows)))]
+            Err(Error::new(
+                ErrorKind::Unsupported,
+                "Unix Domain Sockets and Named Pipes are not supported on this platform",
+            ))
+        }
+    }
+
+    fn split(
+        stream: impl AsyncRead + AsyncWrite + Send + Unpin + 'static,
+    ) -> (BoxedReader, BoxedWriter) {
+        let (reader, writer) = split(stream);
+        (Box::new(reader), Box::new(writer))
+    }
+}

src/bridge/setup.rs

@@ -2,7 +2,7 @@ use log::{info, warn};
 use nvim_rs::Neovim;
 use rmpv::Value;
 
-use crate::{bridge::TxWrapper, error_handling::ResultPanicExplanation};
+use crate::{bridge::NeovimWriter, error_handling::ResultPanicExplanation};
 
 const REGISTER_CLIPBOARD_PROVIDER_LUA: &str = r"
     local function set_clipboard(register)
@@ -30,7 +30,7 @@ const REGISTER_CLIPBOARD_PROVIDER_LUA: &str = r"
         cache_enabled = 0
     }";
 
-pub async fn setup_neovide_remote_clipboard(nvim: &Neovim<TxWrapper>, neovide_channel: u64) {
+pub async fn setup_neovide_remote_clipboard(nvim: &Neovim<NeovimWriter>, neovide_channel: u64) {
     // Users can opt-out with
     // vim: `let g:neovide_no_custom_clipboard = v:true`
     // lua: `vim.g.neovide_no_custom_clipboard = true`
@@ -52,7 +52,10 @@ pub async fn setup_neovide_remote_clipboard(nvim: &Neovim<TxWrapper>, neovide_ch
         .ok();
 }
 
-pub async fn setup_neovide_specific_state(nvim: &Neovim<TxWrapper>, is_remote: bool) {
+pub async fn setup_neovide_specific_state(
+    nvim: &Neovim<NeovimWriter>,
+    should_handle_clipboard: bool,
+) {
     // Set variable indicating to user config that neovide is being used.
     nvim.set_var("neovide", Value::Boolean(true))
         .await
@@ -122,7 +125,7 @@ pub async fn setup_neovide_specific_state(nvim: &Neovim<TxWrapper>, is_remote: b
         .await
         .ok();
 
-        if is_remote {
+        if should_handle_clipboard {
             setup_neovide_remote_clipboard(nvim, neovide_channel).await;
         }
     } else {