feat: subscribe topics fetched from coScout (#23)

* feat: subscribe topics fetched from coScout

* nit

* version

* fix: bugs in humble

Author: YunxianHua

CMakeLists.txt

@@ -25,7 +25,11 @@ if(NOT CMAKE_CXX_STANDARD)
 endif()
 
 if(NOT CMAKE_BUILD_TYPE)
-    set(CMAKE_BUILD_TYPE Debug)
+    set(CMAKE_BUILD_TYPE RelWithDebInfo)
+endif()
+
+if(CMAKE_BUILD_TYPE STREQUAL "Debug")
+    add_definitions(-DDEBUG_BUILD)
 endif()
 
 macro(enable_strict_compiler_warnings target)
@@ -67,6 +71,7 @@ add_library(colistener_base SHARED
         listener_base/src/actions/agi_action.cpp
         listener_base/src/persistence/database_manager.cpp
         listener_base/src/utils/logger.cpp
+        listener_base/src/utils/curl_client.cpp
 
         ${CMAKE_CURRENT_BINARY_DIR}/colistener_base/src/version.cpp
 )

README.md

@@ -101,6 +101,10 @@ Install using apt
 
 This project uses GitHub Actions to automatically build and publish Debian packages. When a new tag is pushed or a new release is created, the workflow automatically updates the version number and builds the corresponding Debian package.
 
+## Caution
+
+coListener will record all the messages it subscribes to and parses in its log file, in order to locate and troubleshoot the problem, so please don't use colistener to subscribe to topics with large amount of data (e.g. sensor_msgs/Image, etc.) to prevent colistener from taking up a lot of disk IO and storage space.
+
 ## Development Guide
 
 * Adding a new action type

README_CN.md

@@ -103,6 +103,10 @@ ROS 环境。
 此项目使用 GitHub Actions 自动构建和发布 Debian 包。当推送新的标签或创建新的发布时，工作流会自动更新版本号并构建对应的
 Debian 包。
 
+## 注意事项
+
+coListener 会将其订阅并解析的所有 message 记录在其 log 文件中,以便进行问题定位及排查,所以,请勿使用 colistener 订阅大数据量的 topic (例如 sensor_msgs/Image 等),以防 colistener 占用大量磁盘 IO 和存储空间
+
 ## 开发指南
 
 * 添加新的动作类型

listener_base/include/actions/common_action.hpp

@@ -16,7 +16,7 @@
 #define ACTIONS__COMMON_ACTION_HPP_
 
 #include "actions/action.hpp"
-#include <curl/curl.h>
+#include "utils/curl_client.hpp"
 #include <string>
 #include <vector>
 
@@ -29,10 +29,9 @@ class CommonAction final : public Action {
     bool execute(const std::vector<MessageCache>& messages) override;
 
 private:
-    CURL* curl_;
+    CurlClient curl_client_;
+    std::map<std::string, std::string> headers_;
     std::string endpoint_;
-    struct curl_slist* headers_;
-    FILE* dev_null_;
 };
 
 }  // namespace colistener

listener_base/include/colistener.hpp

@@ -26,7 +26,8 @@ extern const char GIT_HASH[];
 
 constexpr char DEFAULT_URL[] = "http://localhost";
 constexpr char DEFAULT_PORT[] = "22524";
-constexpr char DEFAULT_ROUTE[] = "/ruleEngine/messages";
+constexpr char SEND_MESSAGES[] = "/ruleEngine/messages";
+constexpr char ACTIVE_TOPICS[] = "/ruleEngine/activeTopics";
 
 enum class RosDataType
 {

listener_base/include/utils/curl_client.hpp

@@ -0,0 +1,73 @@
+// Copyright 2025 coScene
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef UTILS__CURL_CLIENT_HPP_
+#define UTILS__CURL_CLIENT_HPP_
+
+#include <string>
+#include <map>
+#include <curl/curl.h>
+#include <utils/json.hpp>
+
+namespace colistener {
+
+struct HttpResponse {
+    int status_code;
+    std::string body;
+    std::map<std::string, std::string> headers;
+    bool success;
+    std::string error_message;
+};
+
+class CurlClient {
+public:
+    CurlClient();
+    ~CurlClient();
+
+    CurlClient(const CurlClient&) = delete;
+    CurlClient& operator=(const CurlClient&) = delete;
+
+    HttpResponse get(const std::string& url, 
+                     const std::map<std::string, std::string>& headers = {});
+
+    HttpResponse post(const std::string& url, 
+                      const nlohmann::json& data,
+                      const std::map<std::string, std::string>& headers = {});
+
+    HttpResponse post(const std::string& url, 
+                      const std::string& data,
+                      const std::map<std::string, std::string>& headers = {});
+
+    void setTimeout(int timeout_seconds);
+
+    void setVerifySSL(bool verify);
+
+private:
+    CURL* curl_;
+    struct curl_slist* header_list_;
+    int timeout_seconds_;
+    bool verify_ssl_;
+
+    bool initCurl();
+
+    HttpResponse executeRequest();
+
+    void cleanup();
+
+    void cleanupHeaders();
+};
+
+}  // namespace colistener
+
+#endif  // UTILS__CURL_CLIENT_HPP_

listener_base/include/utils/set_utils.hpp

@@ -0,0 +1,48 @@
+// Copyright 2025 coScene
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef UTILS__VECTOR_UTILS_HPP_
+#define UTILS__VECTOR_UTILS_HPP_
+
+#include <vector>
+#include <unordered_set>
+#include <algorithm>
+
+namespace colistener {
+
+template<typename T>
+struct SetDiff {
+    std::vector<T> missing;
+    std::vector<T> added;
+    
+    bool isIdentical() const {
+        return missing.empty() && added.empty();
+    }
+};
+
+template<typename T>
+SetDiff<T> findSetsDifference(const std::set<T>& A, const std::set<T>& B) {
+    SetDiff<T> result;
+
+    std::set_difference(A.begin(), A.end(), B.begin(), B.end(),
+                        std::inserter(result.missing, result.missing.begin()));
+
+    std::set_difference(B.begin(), B.end(), A.begin(), A.end(),
+                        std::inserter(result.added, result.added.begin()));
+
+    return result;
+}
+}  // namespace colistener
+
+#endif  // UTILS__VECTOR_UTILS_HPP_

listener_base/src/actions/agi_action.cpp

@@ -33,7 +33,7 @@ AgiAction::AgiAction() : curl_(nullptr), headers_(nullptr), dev_null_(nullptr) {
     endpoint_ = std::string(DEFAULT_URL) +
         ":" +
         std::string(DEFAULT_PORT) +
-        std::string(DEFAULT_ROUTE);
+        std::string(SEND_MESSAGES);
 
     curl_easy_setopt(curl_, CURLOPT_WRITEDATA, dev_null_);
 }

listener_base/src/actions/common_action.cpp

@@ -18,38 +18,17 @@
 #include "utils/logger.hpp"
 
 namespace colistener {
-CommonAction::CommonAction() : curl_(nullptr), headers_(nullptr), dev_null_(nullptr) {
-    curl_ = curl_easy_init();
-    if (!curl_) {
-        throw std::runtime_error("Failed to initialize CURL");
-    }
-    
-    dev_null_ = fopen("/dev/null", "w");
-    if (!dev_null_) {
-        curl_easy_cleanup(curl_);
-        throw std::runtime_error("Failed to open /dev/null");
-    }
-    
-    headers_ = curl_slist_append(nullptr, "Content-Type: application/json");
+CommonAction::CommonAction() {
+    headers_["Content-Type"] = "application/json";
+    headers_["User-Agent"] = "coListener/1.0";
+
     endpoint_ = std::string(DEFAULT_URL) +
         ":" +
         std::string(DEFAULT_PORT) +
-        std::string(DEFAULT_ROUTE);
-    
-    curl_easy_setopt(curl_, CURLOPT_WRITEDATA, dev_null_);
+        std::string(SEND_MESSAGES);
 }
 
-CommonAction::~CommonAction() {
-    if (headers_) {
-        curl_slist_free_all(headers_);
-    }
-    if (curl_) {
-        curl_easy_cleanup(curl_);
-    }
-    if (dev_null_) {
-        fclose(dev_null_);
-    }
-}
+CommonAction::~CommonAction() = default;
 
 bool CommonAction::execute(const std::vector<MessageCache>& messages) {
     if (messages.empty()) return true;
@@ -82,31 +61,16 @@ bool CommonAction::execute(const std::vector<MessageCache>& messages) {
 
         // Log request details
         COLOG_DEBUG("Sending request to endpoint: %s", endpoint_.c_str());
-        COLOG_DEBUG("Request payload size: %zu bytes", json_str.length());
         // Log detailed request payload for debugging
-        COLOG_DEBUG("Request payload: %s", json_str.c_str());
+        COLOG_INFO("Request payload: %s", json_str.c_str());
 
-        curl_easy_setopt(curl_, CURLOPT_URL, endpoint_.c_str());
-        curl_easy_setopt(curl_, CURLOPT_HTTPHEADER, headers_);
-        curl_easy_setopt(curl_, CURLOPT_POSTFIELDS, json_str.c_str());
-        curl_easy_setopt(curl_, CURLOPT_POSTFIELDSIZE, json_str.length());
 
-        const CURLcode res = curl_easy_perform(curl_);
-        if (res != CURLE_OK) {
-            // Log CURL errors
-            COLOG_ERROR("CURL request failed: %s", curl_easy_strerror(res));
-            return false;
-        }
+        HttpResponse post_response = curl_client_.post(endpoint_, root, headers_);
 
-        int64_t http_code = 0;
-        curl_easy_getinfo(curl_, CURLINFO_RESPONSE_CODE, &http_code);
-        if (http_code >= 200 && http_code < 300) {
-            // Log successful response
-            COLOG_INFO("Request successful, HTTP code: %ld", http_code);
+        if (post_response.success) {
             return true;
         } else {
-            // Log failed response
-            COLOG_ERROR("Request failed with HTTP code: %ld", http_code);
+            COLOG_ERROR("POST request failed: %s", post_response.error_message.c_str());
             return false;
         }
     }