feat: implemented consumer thread, all tests pass

src/core/Consumer.cxx

@@ -0,0 +1,104 @@
+// ConsumerThread.cxx
+// SPDX-License-Identifier: GPL-3.0-or-later
+// Author: Unai Blazquez <unaibg2000@gmail.com>
+
+#include "Consumer.hpp"
+
+#include <sys/socket.h>
+#include <sys/un.h>
+#include <unistd.h>
+
+#include <cstring>
+#include <iostream>
+#include <stdexcept>
+
+ConsumerThread::ConsumerThread(const std::string& socket_path, QObject* parent)
+    : QObject(parent), m_socket_path(socket_path)
+{
+}
+
+ConsumerThread::~ConsumerThread() { stop(); }
+
+void ConsumerThread::start()
+{
+  // Remove stale socket from previous runs
+  unlink(m_socket_path.c_str());
+
+  // Create, bind and listen BEFORE spawning the thread so the socket
+  // is guaranteed ready when start() returns — no race with the producer.
+  m_server_fd = socket(AF_UNIX, SOCK_STREAM, 0);
+  if (m_server_fd < 0)
+  {
+    throw std::runtime_error("ConsumerThread: socket() failed");
+  }
+
+  struct sockaddr_un addr = {};
+  addr.sun_family = AF_UNIX;
+  std::strncpy(addr.sun_path, m_socket_path.c_str(), sizeof(addr.sun_path) - 1);
+
+  if (bind(m_server_fd, reinterpret_cast<sockaddr*>(&addr), sizeof(addr)) < 0)
+  {
+    close(m_server_fd);
+    m_server_fd = -1;
+    throw std::runtime_error("ConsumerThread: bind() failed");
+  }
+
+  if (listen(m_server_fd, 5) < 0)
+  {
+    close(m_server_fd);
+    m_server_fd = -1;
+    throw std::runtime_error("ConsumerThread: listen() failed");
+  }
+
+  m_running.store(true);
+  m_thread = std::thread(&ConsumerThread::run_loop, this);
+}
+
+void ConsumerThread::stop()
+{
+  if (!m_running.exchange(false))
+  {
+    return;  // already stopped or never started
+  }
+
+  // Shutdown the server fd to unblock the blocking accept() call
+  if (m_server_fd >= 0)
+  {
+    shutdown(m_server_fd, SHUT_RDWR);
+    close(m_server_fd);
+    m_server_fd = -1;
+  }
+
+  if (m_thread.joinable())
+  {
+    m_thread.join();
+  }
+
+  unlink(m_socket_path.c_str());
+}
+
+void ConsumerThread::run_loop()
+{
+  while (m_running.load())
+  {
+    int client_fd = accept(m_server_fd, nullptr, nullptr);
+    if (client_fd < 0)
+    {
+      // accept() failed — most likely stop() closed the fd
+      break;
+    }
+
+    int value = 0;
+    ssize_t n = recv(client_fd, &value, sizeof(value), MSG_WAITALL);
+    close(client_fd);
+
+    if (n == static_cast<ssize_t>(sizeof(value)))
+    {
+      // 1) Print to console (spec requirement)
+      std::cout << "ConsumerThread received: " << value << std::endl;
+
+      // 2) Emit Qt signal (spec requirement)
+      emit valueReceived(value);
+    }
+  }
+}