feat(ipc): implement new IPC semantics on Unix

espanso-ipc/src/lib.rs

@@ -28,6 +28,8 @@ pub mod windows;
 #[cfg(not(target_os = "windows"))]
 pub mod unix;
 
+mod util;
+
 pub type EventHandler<Event> = Box<dyn Fn(Event) -> EventHandlerResponse<Event>>;
 
 pub enum EventHandlerResponse<Event> {
@@ -47,17 +49,16 @@ pub trait IPCClient<Event> {
 }
 
 #[cfg(not(target_os = "windows"))]
-pub fn server<Event: Send + Sync + DeserializeOwned>(
+pub fn server<Event: Send + Sync + DeserializeOwned + Serialize>(
   id: &str,
   parent_dir: &Path,
-) -> Result<(impl IPCServer<Event>, Receiver<Event>)> {
-  let (sender, receiver) = unbounded();
-  let server = unix::UnixIPCServer::new(id, parent_dir, sender)?;
-  Ok((server, receiver))
+) -> Result<impl IPCServer<Event>> {
+  let server = unix::UnixIPCServer::new(id, parent_dir)?;
+  Ok(server)
 }
 
 #[cfg(not(target_os = "windows"))]
-pub fn client<Event: Serialize>(id: &str, parent_dir: &Path) -> Result<impl IPCClient<Event>> {
+pub fn client<Event: Serialize + DeserializeOwned>(id: &str, parent_dir: &Path) -> Result<impl IPCClient<Event>> {
   let client = unix::UnixIPCClient::new(id, parent_dir)?;
   Ok(client)
 }

espanso-ipc/src/unix.rs

@@ -17,25 +17,24 @@
  * along with espanso.  If not, see <https://www.gnu.org/licenses/>.
  */
 
+use crate::{EventHandlerResponse, IPCClientError, util::read_line};
 use anyhow::Result;
-use crossbeam::channel::Sender;
 use log::{error, info};
 use serde::{de::DeserializeOwned, Serialize};
 use std::{
-  io::{BufReader, Read, Write},
+  io::{Write},
   os::unix::net::{UnixListener, UnixStream},
-  path::{Path, PathBuf},
+  path::{Path},
 };
 
-use crate::{IPCClient, IPCServer, IPCServerError};
+use crate::{EventHandler, IPCClient, IPCServer};
 
-pub struct UnixIPCServer<Event> {
+pub struct UnixIPCServer {
   listener: UnixListener,
-  sender: Sender<Event>,
 }
 
-impl<Event> UnixIPCServer<Event> {
-  pub fn new(id: &str, parent_dir: &Path, sender: Sender<Event>) -> Result<Self> {
+impl UnixIPCServer {
+  pub fn new(id: &str, parent_dir: &Path) -> Result<Self> {
     let socket_path = parent_dir.join(format!("{}.sock", id));
 
     // Remove previous Unix socket
@@ -50,72 +49,97 @@ impl<Event> UnixIPCServer<Event> {
       socket_path.to_string_lossy()
     );
 
-    Ok(Self { listener, sender })
+    Ok(Self { listener })
   }
 }
 
-impl<Event: Send + Sync + DeserializeOwned> IPCServer<Event> for UnixIPCServer<Event> {
-  fn run(&self) -> anyhow::Result<()> {
+impl<Event: Send + Sync + DeserializeOwned + Serialize> IPCServer<Event> for UnixIPCServer {
+  fn run(self, handler: EventHandler<Event>) -> Result<()> {
     loop {
-      self.accept_one()?;
-    }
-  }
+      let (mut stream, _) = self.listener.accept()?;
 
-  fn accept_one(&self) -> Result<()> {
-    let connection = self.listener.accept();
-
-    match connection {
-      Ok((stream, _)) => {
-        let mut json_str = String::new();
-        let mut buf_reader = BufReader::new(stream);
-        let result = buf_reader.read_to_string(&mut json_str);
-
-        match result {
-          Ok(_) => {
-            let event: Result<Event, serde_json::Error> = serde_json::from_str(&json_str);
+      // Read multiple commands from the client
+      loop {
+        match read_line(&mut stream) {
+          Ok(Some(line)) => {
+            let event: Result<Event, serde_json::Error> = serde_json::from_str(&line);
             match event {
-              Ok(event) => {
-                if self.sender.send(event).is_err() {
-                  return Err(IPCServerError::SendFailed().into());
+              Ok(event) => match handler(event) {
+                EventHandlerResponse::Response(response) => {
+                  let mut json_event = serde_json::to_string(&response)?;
+                  json_event.push('\n');
+                  stream.write_all(json_event.as_bytes())?;
+                  stream.flush()?;
                 }
+                EventHandlerResponse::NoResponse => {
+                  // Async event, no need to reply
                 }
+                EventHandlerResponse::Error(err) => {
+                  error!("ipc handler reported an error: {}", err);
+                }
+                EventHandlerResponse::Exit => {
+                  return Ok(());
+                }
+              },
               Err(error) => {
                 error!("received malformed event from ipc stream: {}", error);
+                break;
               }
             }
           }
+          Ok(None) => {
+            // EOF reached
+            break;
+          }
           Err(error) => {
             error!("error reading ipc stream: {}", error);
+            break;
           }
         }
       }
-      Err(err) => {
-        return Err(IPCServerError::StreamEnded(err).into());
     }
-    };
-
-    Ok(())
   }
 }
 
 pub struct UnixIPCClient {
-  socket_path: PathBuf,
+  stream: UnixStream,
 }
 
 impl UnixIPCClient {
   pub fn new(id: &str, parent_dir: &Path) -> Result<Self> {
     let socket_path = parent_dir.join(format!("{}.sock", id));
+    let stream = UnixStream::connect(&socket_path)?;
 
-    Ok(Self { socket_path })
+    Ok(Self { stream })
   }
 }
 
-impl<Event: Serialize> IPCClient<Event> for UnixIPCClient {
-  fn send(&self, event: Event) -> Result<()> {
-    let mut stream = UnixStream::connect(&self.socket_path)?;
+impl<Event: Serialize + DeserializeOwned> IPCClient<Event> for UnixIPCClient {
+  fn send_sync(&mut self, event: Event) -> Result<Event> {
+    {
+      let mut json_event = serde_json::to_string(&event)?;
+      json_event.push('\n');
+      self.stream.write_all(json_event.as_bytes())?;
+      self.stream.flush()?;
+    }
 
-    let json_event = serde_json::to_string(&event)?;
-    stream.write_all(json_event.as_bytes())?;
+    // Read the response
+    if let Some(line) = read_line(&mut self.stream)? {
+      let event: Result<Event, serde_json::Error> = serde_json::from_str(&line);
+      match event {
+        Ok(response) => Ok(response),
+        Err(err) => Err(IPCClientError::MalformedResponse(err.into()).into()),
+      }
+    } else {
+      Err(IPCClientError::EmptyResponse.into())
+    }
+  }
+
+  fn send_async(&mut self, event: Event) -> Result<()> {
+    let mut json_event = serde_json::to_string(&event)?;
+    json_event.push('\n');
+    self.stream.write_all(json_event.as_bytes())?;
+    self.stream.flush()?;
 
     Ok(())
   }

espanso-ipc/src/util.rs

@@ -0,0 +1,47 @@
+/*
+ * This file is part of espanso.
+ *
+ * Copyright (C) 2019-2021 Federico Terzi
+ *
+ * espanso is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * espanso is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with espanso.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+use anyhow::Result;
+
+// Unbuffered version, necessary to concurrently write
+// to the buffer if necessary (when receiving sync messages)
+pub fn read_line<R: std::io::Read>(stream: R) -> Result<Option<String>> {
+  let mut buffer = Vec::new();
+
+  let mut is_eof = true;
+
+  for byte_res in stream.bytes() {
+    let byte = byte_res?;
+
+    if byte == 10 {
+      // Newline
+      break;
+    } else {
+      buffer.push(byte);
+    }
+
+    is_eof = false;
+  }
+
+  if is_eof {
+    Ok(None)
+  } else {
+    Ok(Some(String::from_utf8(buffer)?))
+  }
+}

espanso-ipc/src/windows.rs

@@ -22,6 +22,7 @@ use log::{error, info};
 use named_pipe::{ConnectingServer, PipeClient, PipeOptions};
 use serde::{de::DeserializeOwned, Serialize};
 use std::{io::{Write}};
+use crate::util::read_line;
 
 use crate::{
   EventHandler, EventHandlerResponse, IPCClient, IPCClientError, IPCServer,
@@ -100,33 +101,6 @@ impl<Event: Send + Sync + DeserializeOwned + Serialize> IPCServer<Event> for Win
   }
 }
 
-// Unbuffered version, necessary to concurrently write
-// to the buffer if necessary (when receiving sync messages)
-fn read_line<R: std::io::Read>(stream: R) -> Result<Option<String>> {
-  let mut buffer = Vec::new();
-
-  let mut is_eof = true;
-
-  for byte_res in stream.bytes() {
-    let byte = byte_res?;
-
-    if byte == 10 {
-      // Newline
-      break;
-    } else {
-      buffer.push(byte);
-    }
-
-    is_eof = false;
-  }
-
-  if is_eof {
-    Ok(None)
-  } else {
-    Ok(Some(String::from_utf8(buffer)?))
-  }
-}
-
 pub struct WinIPCClient {
   stream: PipeClient,
 }