Merge pull request #1067 from JakeStanger/fix/ipc-limit

fix(ipc): message size limited to 1024 bytes

docs/Controlling Ironbar.md

@@ -36,12 +36,10 @@ The server listens on a Unix socket.
 The path is printed on startup, and can usually be found at `/run/user/$UID/ironbar-ipc.sock`.
 
 Commands and responses are sent as JSON objects. 
+The JSON should be minified and must NOT contain any `\n` characters.
 
 Commands will have a `command` key, and a `subcommand` key when part of a sub-command.
 
-The message buffer is currently limited to `1024` bytes. 
-Particularly large messages will be truncated or cause an error.
-
 The full spec can be found below.
 
 ## Libraries

src/ipc/client.rs

@@ -2,7 +2,7 @@ use super::Ipc;
 use crate::ipc::{Command, Response};
 use color_eyre::Result;
 use color_eyre::{Help, Report};
-use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
 use tokio::net::UnixStream;
 
 impl Ipc {
@@ -16,18 +16,20 @@ impl Ipc {
                 .suggestion("Is Ironbar running?")),
         }?;
 
-        let write_buffer = serde_json::to_vec(&command)?;
+        let mut write_buffer = serde_json::to_vec(&command)?;
 
         if debug {
             eprintln!("REQUEST JSON: {}", serde_json::to_string(&command)?);
         }
 
+        write_buffer.push(b'\n');
         stream.write_all(&write_buffer).await?;
 
-        let mut read_buffer = vec![0; 1024];
+        let mut read_buffer = String::new();
-        let bytes = stream.read(&mut read_buffer).await?;
+        let mut reader = BufReader::new(stream);
+        let bytes = reader.read_line(&mut read_buffer).await?;
 
-        let response = serde_json::from_slice(&read_buffer[..bytes])?;
+        let response = serde_json::from_str(&read_buffer[..bytes])?;
         Ok(response)
     }
 }

src/ipc/server/mod.rs

@@ -8,10 +8,10 @@ use std::rc::Rc;
 use color_eyre::{Report, Result};
 use gtk::Application;
 use gtk::prelude::*;
-use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
 use tokio::net::{UnixListener, UnixStream};
 use tokio::sync::mpsc::{self, Receiver, Sender};
-use tracing::{debug, error, info, warn};
+use tracing::{debug, error, info, trace, warn};
 
 use super::Ipc;
 use crate::channels::{AsyncSenderExt, MpscReceiverExt};
@@ -52,11 +52,13 @@ impl Ipc {
             loop {
                 match listener.accept().await {
                     Ok((stream, _addr)) => {
+                        debug!("handling incoming connection");
                         if let Err(err) =
                             Self::handle_connection(stream, &cmd_tx, &mut res_rx).await
                         {
                             error!("{err:?}");
                         }
+                        debug!("done");
                     }
                     Err(err) => {
                         error!("{err:?}");
@@ -80,10 +82,16 @@ impl Ipc {
         cmd_tx: &Sender<Command>,
         res_rx: &mut Receiver<Response>,
     ) -> Result<()> {
-        let (mut stream_read, mut stream_write) = stream.split();
+        trace!("awaiting readable state");
+        stream.readable().await?;
 
-        let mut read_buffer = vec![0; 1024];
+        let mut read_buffer = Vec::with_capacity(1024);
-        let bytes = stream_read.read(&mut read_buffer).await?;
+
+        let mut reader = BufReader::new(&mut stream);
+
+        trace!("reading bytes");
+        let bytes = reader.read_until(b'\n', &mut read_buffer).await?;
+        debug!("read {} bytes", bytes);
 
         // FIXME: Error on invalid command
         let command = serde_json::from_slice::<Command>(&read_buffer[..bytes])?;
@@ -95,10 +103,18 @@ impl Ipc {
             .recv()
             .await
             .unwrap_or(Response::Err { message: None });
-        let res = serde_json::to_vec(&res)?;
 
-        stream_write.write_all(&res).await?;
+        let mut res = serde_json::to_vec(&res)?;
-        stream_write.shutdown().await?;
+        res.push(b'\n');
+
+        trace!("awaiting writable state");
+        stream.writable().await?;
+
+        debug!("writing {} bytes", res.len());
+        stream.write_all(&res).await?;
+
+        trace!("bytes written, shutting down stream");
+        stream.shutdown().await?;
 
         Ok(())
     }