feat: Added DispatchCommand function

wazuh · Aug 16, 2024 · ddd4c8a · ddd4c8a
1 parent e2d0ae7
commit ddd4c8a
Show file tree

Hide file tree

Showing 2 changed files with 53 additions and 3 deletions.
diff --git a/src/agent/command_manager/include/command_manager.hpp b/src/agent/command_manager/include/command_manager.hpp
@@ -13,7 +13,8 @@ namespace command_manager
         ~CommandManager() {};
 
         template<typename T>
-        boost::asio::awaitable<void> ProcessCommandsFromQueue(const std::function<std::optional<T>()> GetCommand)
+        boost::asio::awaitable<void> ProcessCommandsFromQueue(const std::function<std::optional<T>()> GetCommand,
+                                                              const std::function<int(T&)> DispatchMessage)
         {
             using namespace std::chrono_literals;
             const auto executor = co_await boost::asio::this_coro::executor;
@@ -29,7 +30,8 @@ namespace command_manager
                     std::cout << "Queue is empty - WAITING" << std::endl;
                     continue;
                 }
-                // DispatchMessage();
+
+                DispatchMessage(cmd.value());
             }
         }
     };

diff --git a/src/agent/src/agent.cpp b/src/agent/src/agent.cpp
@@ -22,6 +22,17 @@ Agent::~Agent()
 
 void Agent::Run()
 {
+    // add some test command to the queue
+    // const nlohmann::json dataContent = {{"command", {{"name", "001"}, {"type", "stateless"}}},
+    //                                    {"origin", {{"serverName", "node01"}, {"moduleName", "upgradeModule"}}},
+    //                                    {"parameters", {{"error", 0}, {"data", "Command received"}}},
+    //                                    {"status", "Pending"}};
+    //
+
+    const nlohmann::json dataContent = {{"command", "upgradeModule"}, {"status", "Pending"}};
+    const Message messageToSend {MessageType::COMMAND, dataContent, "CommandManager"};
+    m_agentQueue.push(messageToSend);
+
     m_taskManager.EnqueueTask(m_communicator.WaitForTokenExpirationAndAuthenticate());
 
     m_taskManager.EnqueueTask(m_communicator.GetCommandsFromManager(
@@ -42,7 +53,44 @@ void Agent::Run()
             {
                 return std::nullopt;
             }
-            return m_agentQueue.getNext(MessageType::COMMAND);
+            Message m = m_agentQueue.getNext(MessageType::COMMAND);
+
+            std::cout << "COMMAND retrieved from Queue:" << std::endl;
+
+            // pop message from queue
+            m_agentQueue.pop(MessageType::COMMAND);
+
+            nlohmann::json jdata = m.data.at(0);
+            for (auto& [key, value] : jdata.items())
+            {
+                std::cout << key << ": " << value << std::endl;
+            }
+
+            std::cout << "Data inside data:" << std::endl;
+
+            nlohmann::json jdataData = m.data.at(0).at("data");
+            for (auto& [key, value] : jdataData.items())
+            {
+                std::cout << key << ": " << value << std::endl;
+            }
+
+            // change status and push again
+            // if (jdataData.at("status") == "Pending")
+            //{
+            //     std::cout << "Message status is Pending. Updating status and pushing back." << std::endl;
+            //     jdataData["status"] = "InProcess";
+            //     Message newMessage(MessageType::COMMAND, jdataData, "CommandManager");
+            //     // m_agentQueue.push(newMessage);
+            //     // return m;
+            // }
+
+            return m;
+        },
+        [this](Message& cmd) -> int
+        {
+            std::cout << "Dispatching command" << std::endl;
+            std::this_thread::sleep_for(std::chrono::milliseconds(100));
+            return 1;
         }));
 
     m_signalHandler.WaitForSignal();