diff --git a/sharing/linux/daemon/BUILD b/sharing/linux/daemon/BUILD index 7fe8b40c..5e9f98c7 100644 --- a/sharing/linux/daemon/BUILD +++ b/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", ], diff --git a/sharing/linux/daemon/ipc_server.cc b/sharing/linux/daemon/ipc_server.cc index 260ecd97..da9c777d 100644 --- a/sharing/linux/daemon/ipc_server.cc +++ b/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); + } +} diff --git a/sharing/linux/daemon/ipc_server.h b/sharing/linux/daemon/ipc_server.h index 2da6cfd9..1c954647 100644 --- a/sharing/linux/daemon/ipc_server.h +++ b/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 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 handlers_; std::atomic running_{false}; }; diff --git a/sharing/linux/daemon/ipc_server_test.cc b/sharing/linux/daemon/ipc_server_test.cc index cc9009e8..bc22792f 100644 --- a/sharing/linux/daemon/ipc_server_test.cc +++ b/sharing/linux/daemon/ipc_server_test.cc @@ -1,25 +1,78 @@ #include #include +#include #include #include +#include #include #include -#include #include +#include +#include +#include #include #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& 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 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 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()); +}