feat(ipc): refactor IPC on Windows

espanso-ipc/src/lib.rs

@@ -18,9 +18,8 @@
  */
 
 use anyhow::Result;
-use crossbeam::channel::{unbounded, Receiver};
 use serde::{de::DeserializeOwned, Serialize};
-use std::path::Path;
+use std::{path::Path};
 use thiserror::Error;
 
 #[cfg(target_os = "windows")]
@@ -29,13 +28,22 @@ pub mod windows;
 #[cfg(not(target_os = "windows"))]
 pub mod unix;
 
+pub type EventHandler<Event> = Box<dyn Fn(Event) -> EventHandlerResponse<Event>>;
+
+pub enum EventHandlerResponse<Event> {
+  NoResponse,
+  Response(Event),
+  Error(anyhow::Error),
+  Exit,
+}
+
 pub trait IPCServer<Event> {
-  fn run(&self) -> Result<()>;
+  fn run(self, handler: EventHandler<Event>) -> Result<()>;
-  fn accept_one(&self) -> Result<()>;
 }
 
 pub trait IPCClient<Event> {
-  fn send(&self, event: Event) -> Result<()>;
+  fn send_sync(&mut self, event: Event) -> Result<Event>;
+  fn send_async(&mut self, event: Event) -> Result<()>;
 }
 
 #[cfg(not(target_os = "windows"))]
@@ -55,17 +63,16 @@ pub fn client<Event: Serialize>(id: &str, parent_dir: &Path) -> Result<impl IPCC
 }
 
 #[cfg(target_os = "windows")]
-pub fn server<Event: Send + Sync + DeserializeOwned>(
+pub fn server<Event: Send + Sync + DeserializeOwned + Serialize>(
   id: &str,
   _: &Path,
-) -> Result<(impl IPCServer<Event>, Receiver<Event>)> {
+) -> Result<impl IPCServer<Event>> {
-  let (sender, receiver) = unbounded();
+  let server = windows::WinIPCServer::new(id)?;
-  let server = windows::WinIPCServer::new(id, sender)?;
+  Ok(server)
-  Ok((server, receiver))
 }
 
 #[cfg(target_os = "windows")]
-pub fn client<Event: Serialize>(id: &str, _: &Path) -> Result<impl IPCClient<Event>> {
+pub fn client<Event: Serialize + DeserializeOwned>(id: &str, _: &Path) -> Result<impl IPCClient<Event>> {
   let client = windows::WinIPCClient::new(id)?;
   Ok(client)
 }
@@ -73,45 +80,184 @@ pub fn client<Event: Serialize>(id: &str, _: &Path) -> Result<impl IPCClient<Eve
 #[derive(Error, Debug)]
 pub enum IPCServerError {
   #[error("stream ended")]
-  StreamEnded(#[from] std::io::Error),
+  StreamEnded,
 
-  #[error("send failed")]
+  #[error("handler reported error `{0}`")]
-  SendFailed(),
+  HandlerError(#[from] anyhow::Error),
+}
+
+#[derive(Error, Debug)]
+pub enum IPCClientError {
+  #[error("empty response")]
+  EmptyResponse,
+
+  #[error("malformed response received `{0}`")]
+  MalformedResponse(#[from] anyhow::Error),
+
+  #[error("message response timed out")]
+  Timeout,
 }
 
 #[cfg(test)]
 mod tests {
+  use std::{
+    sync::{mpsc::channel},
+  };
+
   use super::*;
   use serde::{Deserialize, Serialize};
 
   #[derive(Serialize, Deserialize)]
   enum Event {
-    Bar,
+    Async,
-    Foo(String),
+    Sync(String),
+    SyncResult(String),
+    ExitRequest,
   }
 
   #[test]
-  fn ipc_works_correctly() {
+  fn ipc_async_message() {
-    let (server, receiver) = server::<Event>("testespansoipc", &std::env::temp_dir()).unwrap();
+    let server = server::<Event>("testespansoipcasync", &std::env::temp_dir()).unwrap();
-    let server_handle = std::thread::spawn(move || {
+    
-      server.accept_one().unwrap();
+    let client_handle = std::thread::spawn(move || {
+      let mut client = client::<Event>("testespansoipcasync", &std::env::temp_dir()).unwrap();
+
+      client.send_async(Event::Async).unwrap();
+      client.send_async(Event::ExitRequest).unwrap();
     });
 
-    // TODO: avoid delay and change the IPC code so that we can wait for the IPC
+    server
-    std::thread::sleep(std::time::Duration::from_millis(300));
+      .run(Box::new(move |event| match event {
+        Event::ExitRequest => EventHandlerResponse::Exit,
+        evt => {
+          assert!(matches!(evt, Event::Async));
+          EventHandlerResponse::NoResponse
+        }
+      }))
+      .unwrap();
 
-    let client = client::<Event>("testespansoipc", &std::env::temp_dir()).unwrap();
+    client_handle.join().unwrap();
-    client.send(Event::Foo("hello".to_string())).unwrap();
-
-    let event = receiver.recv().unwrap();
-    assert!(matches!(event, Event::Foo(x) if x == "hello"));
-
-    server_handle.join().unwrap();
   }
 
   #[test]
-  fn ipc_client_fails_to_send() {
+  fn ipc_sync_message() {
-    let client = client::<Event>("testespansoipc", &std::env::temp_dir()).unwrap();
+    let server = server::<Event>("testespansoipcsync", &std::env::temp_dir()).unwrap();
-    assert!(client.send(Event::Foo("hello".to_string())).is_err());
+    
+    let client_handle = std::thread::spawn(move || {
+      let mut client = client::<Event>("testespansoipcsync", &std::env::temp_dir()).unwrap();
+
+      let response = client.send_sync(Event::Sync("test".to_owned())).unwrap();
+      client.send_async(Event::ExitRequest).unwrap();
+
+      assert!(matches!(response, Event::SyncResult(s) if s == "test"));
+    });
+
+    server
+      .run(Box::new(move |event| match event {
+        Event::ExitRequest => EventHandlerResponse::Exit,
+        Event::Sync(s) => EventHandlerResponse::Response(Event::SyncResult(s)),
+        _ => EventHandlerResponse::NoResponse,
+      }))
+      .unwrap();
+
+    client_handle.join().unwrap();
+  }
+
+  #[test]
+  fn ipc_multiple_sync_with_delay_message() {
+    let server = server::<Event>("testespansoipcmultiplesync", &std::env::temp_dir()).unwrap();
+    
+    let client_handle = std::thread::spawn(move || {
+      let mut client = client::<Event>("testespansoipcmultiplesync", &std::env::temp_dir()).unwrap();
+
+      let response = client.send_sync(Event::Sync("test".to_owned())).unwrap();
+
+      std::thread::sleep(std::time::Duration::from_millis(500));
+
+      let response2 = client.send_sync(Event::Sync("test2".to_owned())).unwrap();
+      client.send_async(Event::ExitRequest).unwrap();
+
+      assert!(matches!(response, Event::SyncResult(s) if s == "test"));
+      assert!(matches!(response2, Event::SyncResult(s) if s == "test2"));
+    });
+
+    server
+      .run(Box::new(move |event| match event {
+        Event::ExitRequest => EventHandlerResponse::Exit,
+        Event::Sync(s) => EventHandlerResponse::Response(Event::SyncResult(s)),
+        _ => EventHandlerResponse::NoResponse,
+      }))
+      .unwrap();
+
+    client_handle.join().unwrap();
+  }
+
+  #[test]
+  fn ipc_multiple_clients() {
+    let server = server::<Event>("testespansoipcmultiple", &std::env::temp_dir()).unwrap();
+
+    let (tx, rx) = channel();
+    
+    let client_handle = std::thread::spawn(move || {
+      let mut client = client::<Event>("testespansoipcmultiple", &std::env::temp_dir()).unwrap();
+
+      let response = client.send_sync(Event::Sync("client1".to_owned())).unwrap();
+
+      tx.send(()).unwrap();
+
+      assert!(matches!(response, Event::SyncResult(s) if s == "client1"));
+    });
+
+    let client_handle2 = std::thread::spawn(move || {
+      let mut client = client::<Event>("testespansoipcmultiple", &std::env::temp_dir()).unwrap();
+
+      let response = client.send_sync(Event::Sync("client2".to_owned())).unwrap();
+
+      // Wait for the other client before terminating
+      rx.recv().unwrap();
+
+      client.send_async(Event::ExitRequest).unwrap();
+
+      assert!(matches!(response, Event::SyncResult(s) if s == "client2"));
+    });
+
+    server
+      .run(Box::new(move |event| match event {
+        Event::ExitRequest => EventHandlerResponse::Exit,
+        Event::Sync(s) => EventHandlerResponse::Response(Event::SyncResult(s)),
+        _ => EventHandlerResponse::NoResponse,
+      }))
+      .unwrap();
+
+    client_handle.join().unwrap();
+    client_handle2.join().unwrap();
+  }
+
+   #[test]
+  fn ipc_sync_big_payload_message() {
+    let server = server::<Event>("testespansoipcsyncbig", &std::env::temp_dir()).unwrap();
+    
+    let client_handle = std::thread::spawn(move || {
+      let mut client = client::<Event>("testespansoipcsyncbig", &std::env::temp_dir()).unwrap();
+
+      let mut payload = String::new();
+      for _ in 0..10000 {
+        payload.push_str("log string repeated");
+      }
+      let response = client.send_sync(Event::Sync(payload.clone())).unwrap();
+      client.send_async(Event::ExitRequest).unwrap();
+
+      assert!(matches!(response, Event::SyncResult(s) if s == payload));
+    });
+
+    server
+      .run(Box::new(move |event| match event {
+        Event::ExitRequest => EventHandlerResponse::Exit,
+        Event::Sync(s) => EventHandlerResponse::Response(Event::SyncResult(s)),
+        _ => EventHandlerResponse::NoResponse,
+      }))
+      .unwrap();
+
+    client_handle.join().unwrap();
   }
 }

espanso-ipc/src/windows.rs

@@ -18,95 +18,154 @@
  */
 
 use anyhow::Result;
-use crossbeam::channel::Sender;
 use log::{error, info};
-use named_pipe::{PipeClient, PipeOptions};
+use named_pipe::{ConnectingServer, PipeClient, PipeOptions};
 use serde::{de::DeserializeOwned, Serialize};
-use std::io::{BufReader, Read, Write};
+use std::{io::{Write}};
 
-use crate::{IPCClient, IPCServer, IPCServerError};
+use crate::{
+  EventHandler, EventHandlerResponse, IPCClient, IPCClientError, IPCServer,
+};
 
-const CLIENT_TIMEOUT: u32 = 2000;
+const DEFAULT_CLIENT_TIMEOUT: u32 = 2000;
 
-pub struct WinIPCServer<Event> {
+pub struct WinIPCServer {
-  options: PipeOptions,
+  server: Option<ConnectingServer>,
-  sender: Sender<Event>,
 }
 
-impl<Event> WinIPCServer<Event> {
+impl WinIPCServer {
-  pub fn new(id: &str, sender: Sender<Event>) -> Result<Self> {
+  pub fn new(id: &str) -> Result<Self> {
     let pipe_name = format!("\\\\.\\pipe\\{}", id);
 
     let options = PipeOptions::new(&pipe_name);
+    let server = Some(options.single()?);
 
     info!("binded to named pipe: {}", pipe_name);
 
-    Ok(Self { options, sender })
+    Ok(Self { server })
   }
 }
 
-impl<Event: Send + Sync + DeserializeOwned> IPCServer<Event> for WinIPCServer<Event> {
+impl<Event: Send + Sync + DeserializeOwned + Serialize> IPCServer<Event> for WinIPCServer {
-  fn run(&self) -> anyhow::Result<()> {
+  fn run(mut self, handler: EventHandler<Event>) -> anyhow::Result<()> {
+    let server = self
+      .server
+      .take()
+      .expect("unable to extract IPC server handle");
+    let mut stream = server.wait()?;
+
     loop {
-      self.accept_one()?;
+      // Read multiple commands from the client
-    }
+      loop {
-  }
+        match read_line(&mut stream) {
-
+          Ok(Some(line)) => {
-  fn accept_one(&self) -> Result<()> {
+            let event: Result<Event, serde_json::Error> = serde_json::from_str(&line);
-    let server = self.options.single()?;
-    let connection = server.wait();
-
-    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);
             match event {
-              Ok(event) => {
+              Ok(event) => match handler(event) {
-                if self.sender.send(event).is_err() {
+                EventHandlerResponse::Response(response) => {
-                  return Err(IPCServerError::SendFailed().into());
+                  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(())
+      stream = stream.disconnect()?.wait()?;
+    }
+  }
+}
+
+// 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 {
-  pipe_name: String,
+  stream: PipeClient,
 }
 
 impl WinIPCClient {
   pub fn new(id: &str) -> Result<Self> {
     let pipe_name = format!("\\\\.\\pipe\\{}", id);
-    Ok(Self { pipe_name })
+
+    let stream = PipeClient::connect_ms(&pipe_name, DEFAULT_CLIENT_TIMEOUT)?;
+    Ok(Self { stream })
   }
 }
 
-impl<Event: Serialize> IPCClient<Event> for WinIPCClient {
+impl<Event: Serialize + DeserializeOwned> IPCClient<Event> for WinIPCClient {
-  fn send(&self, event: Event) -> Result<()> {
+  fn send_sync(&mut self, event: Event) -> Result<Event> {
-    let mut stream = PipeClient::connect_ms(&self.pipe_name, CLIENT_TIMEOUT)?;
+    {
+      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)?;
+    // Read the response
-    stream.write_all(json_event.as_bytes())?;
+    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(())
   }