Use ipmpsc crate for IPC

2 files changed, 35 insertions, 27 deletions

diff --git a/src/common/mod.rs b/src/common/mod.rs
index 0bf235f1..34249f86 100644
--- a/src/common/mod.rs
+++ b/src/common/mod.rs
@@ -18,7 +18,7 @@ use std::{collections::HashMap, fs};
 use crate::panes::PaneId;
 use directories_next::ProjectDirs;
 use input::handler::InputMode;
-use interprocess::local_socket::LocalSocketStream;
+use ipmpsc::{Sender as IpcSender, SharedRingBuffer};
 use serde::{Deserialize, Serialize};
 use termion::input::TermRead;
 use wasm_vm::PluginEnv;
@@ -132,37 +132,32 @@ thread_local!(
     /// stack in the form of an [`ErrorContext`].
     static OPENCALLS: RefCell<ErrorContext> = RefCell::default()
 );
+#[derive(Clone)]
 pub struct IpcSenderWithContext {
     err_ctx: ErrorContext,
-    sender: BufWriter<LocalSocketStream>,
+    sender: IpcSender,
 }
 
 impl IpcSenderWithContext {
-    pub fn new() -> Self {
+    pub fn new(buffer: SharedRingBuffer) -> Self {
         Self {
             err_ctx: ErrorContext::new(),
-            sender: BufWriter::new(LocalSocketStream::connect(ZELLIJ_IPC_PIPE).unwrap()),
+            sender: IpcSender::new(buffer),
         }
     }
 
-    pub fn update(&mut self, ctx: ErrorContext) {
-        self.err_ctx = ctx;
+    // This is expensive. Use this only if a buffer is not available.
+    // Otherwise clone the buffer and use `new()`
+    pub fn to_server() -> Self {
+        Self::new(SharedRingBuffer::open(ZELLIJ_IPC_PIPE).unwrap())
     }
 
-    pub fn send(&mut self, msg: ServerInstruction) -> std::io::Result<()> {
-        let command = bincode::serialize(&(self.err_ctx, msg)).unwrap();
-        let x = self.sender.write_all(&command);
-        self.sender.flush();
-        x
+    pub fn update(&mut self, ctx: ErrorContext) {
+        self.err_ctx = ctx;
     }
-}
 
-impl std::clone::Clone for IpcSenderWithContext {
-    fn clone(&self) -> Self {
-        Self {
-            err_ctx: self.err_ctx,
-            sender: BufWriter::new(LocalSocketStream::connect(ZELLIJ_IPC_PIPE).unwrap()),
-        }
+    pub fn send<T: Serialize>(&mut self, msg: T) -> ipmpsc::Result<()> {
+        self.sender.send(&(self.err_ctx, msg))
     }
 }
 
@@ -215,7 +210,7 @@ pub fn start(mut os_input: Box<dyn OsApi>, opts: CliArgs, config: Config) {
     let send_app_instructions =
         SenderWithContext::new(SenderType::SyncSender(send_app_instructions));
 
-    let pty_thread = start_server(
+    let ipc_thread = start_server(
         os_input.clone(),
         opts.clone(),
         command_is_executing.clone(),
@@ -555,7 +550,7 @@ pub fn start(mut os_input: Box<dyn OsApi>, opts: CliArgs, config: Config) {
             }
         });
 
-    let mut send_server_instructions = IpcSenderWithContext::new();
+    let mut send_server_instructions = IpcSenderWithContext::to_server();
 
     #[warn(clippy::never_loop)]
     loop {
@@ -572,7 +567,7 @@ pub fn start(mut os_input: Box<dyn OsApi>, opts: CliArgs, config: Config) {
             }
             AppInstruction::Error(backtrace) => {
                 let _ = send_server_instructions.send(ServerInstruction::Quit);
-                //let _ = pty_thread.join();
+                //let _ = ipc_thread.join();
                 let _ = send_screen_instructions.send(ScreenInstruction::Quit);
                 let _ = screen_thread.join();
                 let _ = send_plugin_instructions.send(PluginInstruction::Quit);
@@ -599,7 +594,7 @@ pub fn start(mut os_input: Box<dyn OsApi>, opts: CliArgs, config: Config) {
     }
 
     let _ = send_server_instructions.send(ServerInstruction::Quit);
-    //let _ = pty_thread.join().unwrap();
+    //let _ = ipc_thread.join().unwrap();
     let _ = send_screen_instructions.send(ScreenInstruction::Quit);
     screen_thread.join().unwrap();
     let _ = send_plugin_instructions.send(PluginInstruction::Quit);
diff --git a/src/common/pty_bus.rs b/src/common/pty_bus.rs
index 4cbd5675..52144454 100644
--- a/src/common/pty_bus.rs
+++ b/src/common/pty_bus.rs
@@ -193,12 +193,16 @@ pub struct PtyBus {
     pub send_server_instructions: IpcSenderWithContext,
 }
 
-fn stream_terminal_bytes(pid: RawFd, os_input: Box<dyn OsApi>, debug: bool) -> JoinHandle<()> {
+fn stream_terminal_bytes(
+    pid: RawFd,
+    os_input: Box<dyn OsApi>,
+    mut send_server_instructions: IpcSenderWithContext,
+    debug: bool,
+) -> JoinHandle<()> {
     let mut err_ctx = OPENCALLS.with(|ctx| *ctx.borrow());
     task::spawn({
         async move {
             err_ctx.add_call(ContextType::AsyncTask);
-            let mut send_server_instructions = IpcSenderWithContext::new();
             send_server_instructions.update(err_ctx);
             let mut vte_parser = vte::Parser::new();
             let mut vte_event_sender = VteEventSender::new(pid, send_server_instructions.clone());
@@ -286,8 +290,12 @@ impl PtyBus {
     pub fn spawn_terminal(&mut self, file_to_open: Option<PathBuf>) -> RawFd {
         let (pid_primary, pid_secondary): (RawFd, RawFd) =
             self.os_input.spawn_terminal(file_to_open);
-        let task_handle =
-            stream_terminal_bytes(pid_primary, self.os_input.clone(), self.debug_to_file);
+        let task_handle = stream_terminal_bytes(
+            pid_primary,
+            self.os_input.clone(),
+            self.send_server_instructions.clone(),
+            self.debug_to_file,
+        );
         self.task_handles.insert(pid_primary, task_handle);
         self.id_to_child_pid.insert(pid_primary, pid_secondary);
         pid_primary
@@ -307,7 +315,12 @@ impl PtyBus {
             )))
             .unwrap();
         for id in new_pane_pids {
-            let task_handle = stream_terminal_bytes(id, self.os_input.clone(), self.debug_to_file);
+            let task_handle = stream_terminal_bytes(
+                id,
+                self.os_input.clone(),
+                self.send_server_instructions.clone(),
+                self.debug_to_file,
+            );
             self.task_handles.insert(id, task_handle);
         }
     }