Support receiving IPC messages in multiple parts

include/components/controller.hpp

@@ -57,7 +57,6 @@ class controller : public signal_receiver<SIGN_PRIORITY_CONTROLLER, signals::eve
   void signal_handler(int signum);
 
   void conn_cb(int status, int events);
-  void ipc_cb(string buf);
   void confwatch_handler(const char* fname, int events, int status);
   void notifier_handler();
   void screenshot_handler();

include/components/eventloop.hpp

@@ -79,11 +79,12 @@ struct FSEventHandle : public UVHandle<uv_fs_event_t, const char*, int, int> {
 };
 
 struct PipeHandle : public UVHandleGeneric<uv_pipe_t, uv_stream_t, ssize_t, const uv_buf_t*> {
-  PipeHandle(uv_loop_t* loop, std::function<void(const string)> fun);
+  PipeHandle(uv_loop_t* loop, std::function<void(const string)> fun, std::function<void(void)> eof_cb);
   void start(int fd);
   void read_cb(ssize_t nread, const uv_buf_t* buf);
 
   std::function<void(const string)> func;
+  std::function<void(void)> eof_cb;
   int fd;
 };
 
@@ -114,7 +115,7 @@ class eventloop {
   void signal_handler(int signum, std::function<void(int)> fun);
   void poll_handler(int events, int fd, std::function<void(int, int)> fun);
   void fs_event_handler(const string& path, std::function<void(const char*, int, int)> fun);
-  void pipe_handle(int fd, std::function<void(const string)> fun);
+  void pipe_handle(int fd, std::function<void(const string)> fun, std::function<void(void)> eof_cb);
   void timer_handle(uint64_t timeout, uint64_t repeat, std::function<void(void)> fun);
   AsyncHandle_t async_handle(std::function<void(void)> fun);
 

include/components/ipc.hpp

@@ -34,7 +34,8 @@ class ipc {
   explicit ipc(signal_emitter& emitter, const logger& logger);
   ~ipc();
 
-  void receive_message(string buf);
+  void receive_data(string buf);
+  void receive_eof();
   int get_file_descriptor() const;
 
  private:
@@ -42,7 +43,12 @@ class ipc {
   const logger& m_log;
 
   string m_path{};
-  unique_ptr<file_descriptor> m_fd;
+  int m_fd;
+
+  /**
+   * Buffer for the currently received IPC message.
+   */
+  string m_buffer{};
 };
 
 POLYBAR_NS_END

src/components/controller.cpp

@@ -195,11 +195,6 @@ void controller::conn_cb(int status, int) {
   }
 }
 
-void controller::ipc_cb(string buf) {
-  // TODO handle messages sent in multiple parts.
-  m_ipc->receive_message(buf);
-}
-
 void controller::signal_handler(int signum) {
   m_log.notice("Received signal(%d): %s", signum, strsignal(signum));
   stop(signum == SIGUSR1);
@@ -264,7 +259,9 @@ void controller::read_events(bool confwatch) {
     }
 
     if (m_ipc) {
-      eloop->pipe_handle(m_ipc->get_file_descriptor(), [this](const string payload) { ipc_cb(payload); });
+      eloop->pipe_handle(
+          m_ipc->get_file_descriptor(), [this](const string payload) { m_ipc->receive_data(payload); },
+          [this]() { m_ipc->receive_eof(); });
     }
 
     if (!m_snapshot_dst.empty()) {

src/components/eventloop.cpp

@@ -73,8 +73,8 @@ void FSEventHandle::start(const string& path) {
 // }}}
 
 // PipeHandle {{{
-PipeHandle::PipeHandle(uv_loop_t* loop, std::function<void(const string)> fun)
+PipeHandle::PipeHandle(uv_loop_t* loop, std::function<void(const string)> fun, std::function<void(void)> eof_cb)
-    : UVHandleGeneric([&](ssize_t nread, const uv_buf_t* buf) { read_cb(nread, buf); }), func(fun) {
+    : UVHandleGeneric([&](ssize_t nread, const uv_buf_t* buf) { read_cb(nread, buf); }), func(fun), eof_cb(eof_cb) {
   UV(uv_pipe_init, loop, handle, false);
 }
 
@@ -97,6 +97,7 @@ void PipeHandle::read_cb(ssize_t nread, const uv_buf_t* buf) {
       log.err("Read error: %s", uv_err_name(nread));
       uv_close((uv_handle_t*)handle, nullptr);
     } else {
+      eof_cb();
       // TODO this causes constant EOFs
       start(this->fd);
     }
@@ -191,8 +192,8 @@ void eventloop::fs_event_handler(const string& path, std::function<void(const ch
   m_fs_event_handles.back()->start(path);
 }
 
-void eventloop::pipe_handle(int fd, std::function<void(const string)> fun) {
+void eventloop::pipe_handle(int fd, std::function<void(const string)> fun, std::function<void(void)> eof_cb) {
-  m_pipe_handles.emplace_back(std::make_unique<PipeHandle>(get(), fun));
+  m_pipe_handles.emplace_back(std::make_unique<PipeHandle>(get(), fun, eof_cb));
   m_pipe_handles.back()->start(fd);
 }
 

src/components/ipc.cpp

@@ -33,8 +33,11 @@ ipc::ipc(signal_emitter& emitter, const logger& logger) : m_sig(emitter), m_log(
     throw system_error("Failed to create ipc channel");
   }
 
+  if ((m_fd = open(m_path.c_str(), O_RDONLY | O_NONBLOCK)) == -1) {
+    throw system_error("Failed to open pipe '" + m_path + "'");
+  }
+
   m_log.info("Created ipc channel at: %s", m_path);
-  m_fd = file_util::make_file_descriptor(m_path, O_RDONLY | O_NONBLOCK, false);
 }
 
 /**
@@ -48,12 +51,23 @@ ipc::~ipc() {
 }
 
 /**
- * Receive available ipc messages and delegate valid events
+ * Receive parts of an IPC message
  */
-void ipc::receive_message(string buf) {
+void ipc::receive_data(string buf) {
-  m_log.info("Receiving ipc message");
+  m_buffer += buf;
+}
 
-  string payload{string_util::trim(std::move(buf), '\n')};
+/**
+ * Called once the end of the message arrives.
+ */
+void ipc::receive_eof() {
+  if (m_buffer.empty()) {
+    return;
+  }
+
+  string payload{string_util::trim(std::move(m_buffer), '\n')};
+
+  m_buffer = std::string();
 
   if (payload.find(ipc_command_prefix) == 0) {
     m_sig.emit(signals::ipc::command{payload.substr(strlen(ipc_command_prefix))});
@@ -61,7 +75,7 @@ void ipc::receive_message(string buf) {
     m_sig.emit(signals::ipc::hook{payload.substr(strlen(ipc_hook_prefix))});
   } else if (payload.find(ipc_action_prefix) == 0) {
     m_sig.emit(signals::ipc::action{payload.substr(strlen(ipc_action_prefix))});
-  } else if (!payload.empty()) {
+  } else {
     m_log.warn("Received unknown ipc message: (payload=%s)", payload);
   }
 }
@@ -70,7 +84,7 @@ void ipc::receive_message(string buf) {
  * Get the file descriptor to the ipc channel
  */
 int ipc::get_file_descriptor() const {
-  return *m_fd;
+  return m_fd;
 }
 
 POLYBAR_NS_END