added more tests for ipc. added handlers to ipc

sharing/linux/daemon/BUILD

@@ -14,6 +14,7 @@ cc_test(
 cc_library(
   name = "ipc_server",
   hdrs = ["ipc_server.h"],
+  srcs = ["ipc_server.cc"],
   deps = [
         "@com_google_absl//absl/synchronization",
   ],

sharing/linux/daemon/ipc_server.cc

@@ -158,3 +158,43 @@ void IPCServer::StartEventLoop() {
 
   Stop();
 }
+
+void IPCServer::DispatchOne(const std::string& line) {
+  auto space = line.find(' ');
+
+  std::string command =
+      space == std::string::npos ? line : line.substr(0, space);
+
+  std::string_view args;
+  if (space != std::string::npos) {
+    args = std::string_view(line).substr(space + 1);
+  }
+
+  Handler handler;
+
+  {
+    absl::MutexLock lock(lock_);
+
+    auto it = handlers_.find(command);
+    if (it == handlers_.end()) {
+      std::cout << "Unknown command: " << command << "\n";
+      return;
+    }
+
+    handler = it->second;
+  }
+
+  handler(args);
+}
+void IPCServer::DispatchLoop() {
+  while (running_.load()) {
+    std::string line = Read();
+
+    if (line.empty()) {
+      std::this_thread::sleep_for(std::chrono::milliseconds(10));
+      continue;
+    }
+
+    DispatchOne(line);
+  }
+}

sharing/linux/daemon/ipc_server.h

@@ -18,27 +18,29 @@ class IPCServer {
 
   ~IPCServer() { Stop(); }
   void Stop();
-  void InitialiseSock();
-  void Recv();
   void StartEventLoop();
-  std::string Read();
+  using Handler = std::function<void(std::string_view args)>;
+
+  void RegisterHandler(std::string command, Handler handler) {
+    absl::MutexLock lock(lock_);
+    handlers_[std::move(command)] = std::move(handler);
+  }
+
+  void DispatchLoop();
 
  private:
-  FRIEND_TEST(IPCServerEventLoopTest, ClientCanConnectSendAndServerCanRead);
-  FRIEND_TEST(IPCServerEventLoopTest, ServerCanReadMultipleCommands);
-  FRIEND_TEST(IPCServerEventLoopTest, PartialCommandIsBufferedUntilDelimiter);
-  FRIEND_TEST(IPCServerEventLoopTest, LargeCommandAcrossMultipleRecvCalls);
-  FRIEND_TEST(IPCServerEventLoopTest, MultipleCommandsSplitAcrossSends);
-  FRIEND_TEST(IPCServerEventLoopTest, ClientDisconnectDoesNotCrashServer);
-  FRIEND_TEST(IPCServerEventLoopTest, ClientCanReconnectAfterDisconnect);
-  FRIEND_TEST(IPCServerEventLoopTest, StopEndsEventLoopThread);
-  FRIEND_TEST(IPCServerEventLoopTest, StopWhileClientConnected);
-  FRIEND_TEST(IPCServerEventLoopTest, StaleSocketPathIsCleanedUp);
+  friend class IPCServerTest;
+
+  void DispatchOne(const std::string& line);
+  void InitialiseSock();
+  void Recv();
+  std::string Read();
 
   int sock_fd_ = -1;
   int client_fd_ = -1;
   sockaddr_un addr{};
   std::string read_buf;
   absl::Mutex lock_;
+  std::unordered_map<std::string, Handler> handlers_;
   std::atomic<bool> running_{false};
 };

sharing/linux/daemon/ipc_server_test.cc

@@ -1,25 +1,78 @@
 #include <gtest/gtest.h>
 
 #include <sys/socket.h>
+#include <sys/stat.h>
 #include <sys/un.h>
 #include <unistd.h>
 
+#include <atomic>
 #include <chrono>
 #include <cstring>
-#include <string>
 #include <fstream>
+#include <functional>
+#include <string>
+#include <string_view>
 #include <thread>
 
 #include "ipc_server.h"
 
-namespace {
+class IPCServerTest : public ::testing::Test {
+ protected:
+  static std::string Read(IPCServer& server) {
+    return server.Read();
+  }
 
+  static void DispatchOne(IPCServer& server, const std::string& line) {
+    server.DispatchOne(line);
+  }
+
+  static void SetReadBuffer(IPCServer& server, std::string value) {
+    absl::MutexLock lock(server.lock_);
+    server.read_buf = std::move(value);
+  }
+
+  static void SetRunning(IPCServer& server, bool running) {
+    server.running_.store(running);
+  }
+
+  static bool IsRunning(IPCServer& server) {
+    return server.running_.load();
+  }
+
+  static std::string WaitRead(IPCServer& server) {
+    for (int i = 0; i < 100; ++i) {
+      std::string cmd = Read(server);
+
+      if (!cmd.empty()) {
+        return cmd;
+      }
+
+      std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    }
+
+    return "";
+  }
+
+  static bool WaitUntilTrue(const std::function<bool()>& condition) {
+    for (int i = 0; i < 100; ++i) {
+      if (condition()) {
+        return true;
+      }
+
+      std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    }
+
+    return false;
+  }
+};
+
+namespace {
 
 int ConnectClientWithRetry() {
   int client_fd = socket(AF_UNIX, SOCK_STREAM, 0);
   EXPECT_GE(client_fd, 0);
 
-  sockaddr_un addr{};
+  sockaddr_un addr {};
   addr.sun_family = AF_UNIX;
   strncpy(addr.sun_path, SOCK_PATH.data(), sizeof(addr.sun_path) - 1);
 
@@ -57,21 +110,6 @@ void SendAll(int fd, std::string_view data) {
   }
 }
 
-std::string WaitRead(IPCServer& server) {
-  for (int i = 0; i < 100; ++i) {
-    std::string cmd = server.Read();
-
-    if (!cmd.empty()) {
-      return cmd;
-    }
-
-    std::this_thread::sleep_for(std::chrono::milliseconds(10));
-  }
-
-  return "";
-}
-
-
 void StopAndJoin(IPCServer& server, std::thread& server_thread) {
   server.Stop();
 
@@ -82,7 +120,9 @@ void StopAndJoin(IPCServer& server, std::thread& server_thread) {
   unlink(SOCK_PATH.data());
 }
 
-TEST(IPCServerEventLoopTest, ServerCanReadMultipleCommands) {
+}  // namespace
+
+TEST_F(IPCServerTest, ServerCanReadMultipleCommands) {
   IPCServer server;
 
   std::thread server_thread([&server]() {
@@ -102,7 +142,7 @@ TEST(IPCServerEventLoopTest, ServerCanReadMultipleCommands) {
   StopAndJoin(server, server_thread);
 }
 
-TEST(IPCServerEventLoopTest, PartialCommandIsBufferedUntilDelimiter) {
+TEST_F(IPCServerTest, PartialCommandIsBufferedUntilDelimiter) {
   IPCServer server;
 
   std::thread server_thread([&server]() {
@@ -116,7 +156,7 @@ TEST(IPCServerEventLoopTest, PartialCommandIsBufferedUntilDelimiter) {
 
   std::this_thread::sleep_for(std::chrono::milliseconds(50));
 
-  EXPECT_EQ(server.Read(), "");
+  EXPECT_EQ(Read(server), "");
 
   SendAll(client_fd, "G\n");
 
@@ -126,7 +166,7 @@ TEST(IPCServerEventLoopTest, PartialCommandIsBufferedUntilDelimiter) {
   StopAndJoin(server, server_thread);
 }
 
-TEST(IPCServerEventLoopTest, LargeCommandAcrossMultipleRecvCalls) {
+TEST_F(IPCServerTest, LargeCommandAcrossMultipleRecvCalls) {
   IPCServer server;
 
   std::thread server_thread([&server]() {
@@ -145,7 +185,7 @@ TEST(IPCServerEventLoopTest, LargeCommandAcrossMultipleRecvCalls) {
   StopAndJoin(server, server_thread);
 }
 
-TEST(IPCServerEventLoopTest, MultipleCommandsSplitAcrossSends) {
+TEST_F(IPCServerTest, MultipleCommandsSplitAcrossSends) {
   IPCServer server;
 
   std::thread server_thread([&server]() {
@@ -167,7 +207,7 @@ TEST(IPCServerEventLoopTest, MultipleCommandsSplitAcrossSends) {
   StopAndJoin(server, server_thread);
 }
 
-TEST(IPCServerEventLoopTest, ClientDisconnectDoesNotCrashServer) {
+TEST_F(IPCServerTest, ClientDisconnectDoesNotCrashServer) {
   IPCServer server;
 
   std::thread server_thread([&server]() {
@@ -181,13 +221,12 @@ TEST(IPCServerEventLoopTest, ClientDisconnectDoesNotCrashServer) {
 
   std::this_thread::sleep_for(std::chrono::milliseconds(100));
 
-  // Main expectation: server did not crash/hang.
   StopAndJoin(server, server_thread);
 
   SUCCEED();
 }
 
-TEST(IPCServerEventLoopTest, ClientCanReconnectAfterDisconnect) {
+TEST_F(IPCServerTest, ClientCanReconnectAfterDisconnect) {
   IPCServer server;
 
   std::thread server_thread([&server]() {
@@ -214,7 +253,7 @@ TEST(IPCServerEventLoopTest, ClientCanReconnectAfterDisconnect) {
   StopAndJoin(server, server_thread);
 }
 
-TEST(IPCServerEventLoopTest, StopEndsEventLoopThread) {
+TEST_F(IPCServerTest, StopEndsEventLoopThread) {
   IPCServer server;
 
   std::thread server_thread([&server]() {
@@ -228,7 +267,7 @@ TEST(IPCServerEventLoopTest, StopEndsEventLoopThread) {
   SUCCEED();
 }
 
-TEST(IPCServerEventLoopTest, StopWhileClientConnected) {
+TEST_F(IPCServerTest, StopWhileClientConnected) {
   IPCServer server;
 
   std::thread server_thread([&server]() {
@@ -248,15 +287,16 @@ TEST(IPCServerEventLoopTest, StopWhileClientConnected) {
   SUCCEED();
 }
 
-TEST(IPCServerEventLoopTest, StaleSocketPathIsCleanedUp) {
+TEST_F(IPCServerTest, StaleSocketPathIsCleanedUp) {
   unlink(SOCK_PATH.data());
 
   {
     std::ofstream stale_file{SOCK_PATH.data()};
+    ASSERT_TRUE(stale_file.is_open());
     stale_file << "stale";
   }
 
-  struct stat before{};
+  struct stat before {};
   ASSERT_EQ(stat(SOCK_PATH.data(), &before), 0);
   ASSERT_TRUE(S_ISREG(before.st_mode));
 
@@ -269,11 +309,140 @@ TEST(IPCServerEventLoopTest, StaleSocketPathIsCleanedUp) {
   int client_fd = ConnectClientWithRetry();
   ASSERT_GE(client_fd, 0);
 
-  struct stat after{};
+  struct stat after {};
   ASSERT_EQ(stat(SOCK_PATH.data(), &after), 0);
   EXPECT_TRUE(S_ISSOCK(after.st_mode));
 
   close(client_fd);
   StopAndJoin(server, server_thread);
 }
-}  // namespace
+
+TEST_F(IPCServerTest, DispatchOneCallsRegisteredHandler) {
+  IPCServer server;
+
+  bool called = false;
+
+  server.RegisterHandler("PING", [&](std::string_view args) {
+    called = true;
+  });
+
+  DispatchOne(server, "PING");
+
+  EXPECT_TRUE(called);
+}
+
+TEST_F(IPCServerTest, DispatchOnePassesArguments) {
+  IPCServer server;
+
+  std::string received_args;
+
+  server.RegisterHandler("ECHO", [&](std::string_view args) {
+    received_args = std::string(args);
+  });
+
+  DispatchOne(server, "ECHO hello world");
+
+  EXPECT_EQ(received_args, "hello world");
+}
+
+TEST_F(IPCServerTest, DispatchOneHandlesCommandWithoutArgs) {
+  IPCServer server;
+
+  std::string received_args = "not empty";
+
+  server.RegisterHandler("PING", [&](std::string_view args) {
+    received_args = std::string(args);
+  });
+
+  DispatchOne(server, "PING");
+
+  EXPECT_EQ(received_args, "");
+}
+
+TEST_F(IPCServerTest, DispatchOneIgnoresUnknownCommand) {
+  IPCServer server;
+
+  bool called = false;
+
+  server.RegisterHandler("PING", [&](std::string_view args) {
+    called = true;
+  });
+
+  DispatchOne(server, "UNKNOWN something");
+
+  EXPECT_FALSE(called);
+}
+
+TEST_F(IPCServerTest, DispatchLoopDispatchesBufferedCommand) {
+  IPCServer server;
+
+  std::atomic<bool> called{false};
+  std::string received_args;
+
+  server.RegisterHandler("ECHO", [&](std::string_view args) {
+    received_args = std::string(args);
+    called.store(true);
+  });
+
+  SetReadBuffer(server, "ECHO hello\n");
+  SetRunning(server, true);
+
+  std::thread dispatch_thread([&server]() {
+    server.DispatchLoop();
+  });
+
+  ASSERT_TRUE(WaitUntilTrue([&]() {
+    return called.load();
+  }));
+
+  SetRunning(server, false);
+
+  dispatch_thread.join();
+
+  EXPECT_EQ(received_args, "hello");
+}
+
+TEST_F(IPCServerTest, SocketCommandReachesRegisteredHandler) {
+  IPCServer server;
+
+  std::atomic<bool> called{false};
+  std::string received_args;
+
+  server.RegisterHandler("ECHO", [&](std::string_view args) {
+    received_args = std::string(args);
+    called.store(true);
+  });
+
+  std::thread server_thread([&server]() {
+    server.StartEventLoop();
+  });
+
+  int client_fd = ConnectClientWithRetry();
+  ASSERT_GE(client_fd, 0);
+
+  std::thread dispatch_thread([&server]() {
+    server.DispatchLoop();
+  });
+
+  SendAll(client_fd, "ECHO from socket\n");
+
+  ASSERT_TRUE(WaitUntilTrue([&]() {
+    return called.load();
+  }));
+
+  EXPECT_EQ(received_args, "from socket");
+
+  close(client_fd);
+
+  server.Stop();
+
+  if (server_thread.joinable()) {
+    server_thread.join();
+  }
+
+  if (dispatch_thread.joinable()) {
+    dispatch_thread.join();
+  }
+
+  unlink(SOCK_PATH.data());
+}