add topic monitor

Author: KingRichord

CMakeLists.txt

@@ -11,4 +11,5 @@ if (NOT CMAKE_BUILD_TYPE)
 endif ()
 
 add_subdirectory(infinite_sense_core)
+add_subdirectory(tools)
 add_subdirectory(example)

example/GigeCam/main.cpp

@@ -15,7 +15,6 @@ int main() {
   synchronizer.Start();
 
   // 4.接收数据
-  Synchronizer::PrintSummary();
   zmq::context_t context(1);
   zmq::socket_t subscriber(context, zmq::socket_type::sub);
   subscriber.connect("tcp://localhost:5555");

infinite_sense_core/include/infinite_sense.h

@@ -75,11 +75,6 @@ class Synchronizer {
    */
   static bool GetLastTriggerTime(TriggerDevice dev, uint64_t time);
 
-  /**
-   * @brief 打印当前系统的配置与状态摘要信息。
-   */
-  static void PrintSummary();
-
  private:
   /// 网络地址
   std::string net_ip_;

infinite_sense_core/include/messenger.h

@@ -27,52 +27,4 @@ class Messenger {
   zmq::socket_t publisher_{};
   std::string endpoint_{};
 };
-
-class TopicMonitor {
- public:
-  static TopicMonitor& GetInstance() {
-    static TopicMonitor instance;
-    return instance;
-  }
-  TopicMonitor(const TopicMonitor&) = delete;
-  TopicMonitor(TopicMonitor&&) = delete;
-  TopicMonitor& operator=(const TopicMonitor&) = delete;
-  TopicMonitor& operator=(TopicMonitor&&) = delete;
-  std::unordered_set<std::string> GetTopics() const;
-  friend std::ostream& operator<<(std::ostream& os, const TopicMonitor& monitor) {
-    std::lock_guard lock(monitor.topics_mutex_);
-    os << "\n--- Topic Monitor ---\n";
-    if (monitor.topic_frequencies_.empty()) {
-      os << "  No active topics\n";
-    } else {
-      os << "  Active Topics (" << monitor.topic_frequencies_.size() << "):\n";
-      std::vector<std::pair<std::string, size_t>> sorted_topics(monitor.topic_frequencies_.begin(),
-                                                                monitor.topic_frequencies_.end());
-      std::sort(sorted_topics.begin(), sorted_topics.end(),
-                [](const auto& a, const auto& b) { return b.second < a.second; });
-
-      for (const auto& [topic, count] : sorted_topics) {
-        os << "    • " << topic << " (num: " << count << ")\n";
-      }
-    }
-    return os << "---------------------\n";
-  }
-  // 启动监控线程
-  void Start();
-  // 停止监控
-  void Stop();
-
- private:
-  TopicMonitor();
-  ~TopicMonitor();
-  // 监控线程主循环
-  void MonitorLoop();
-  std::unordered_map<std::string, size_t> topic_frequencies_;  // 新增：topic频率统计
-  zmq::context_t context_;
-  zmq::socket_t subscriber_;
-  std::unordered_set<std::string> topics_;
-  mutable std::mutex topics_mutex_;
-  std::thread monitor_thread_;
-  std::atomic<bool> should_run_;
-};
 }  // namespace infinite_sense

infinite_sense_core/src/infinite_sense.cpp

@@ -57,10 +57,4 @@ void Synchronizer::Stop() const {
   LOG(INFO) << "Synchronizer stopped";
 }
 
-void Synchronizer::PrintSummary() {
-  TopicMonitor::GetInstance().Start();
-  std::this_thread::sleep_for(std::chrono::milliseconds{1001});
-  TopicMonitor::GetInstance().Stop();
-  LOG(INFO) << TopicMonitor::GetInstance();
-}
 }  // namespace infinite_sense

infinite_sense_core/src/messenger.cpp

@@ -47,67 +47,4 @@ void Messenger::PubStruct(const std::string &topic, const void *data, const size
 }
 std::string Messenger::GetPubEndpoint() const { return endpoint_; }
 
-
-std::unordered_set<std::string> TopicMonitor::GetTopics() const {
-  std::lock_guard lock(topics_mutex_);
-  return topics_;
-}
-void TopicMonitor::Start() {
-  if (monitor_thread_.joinable()) {
-    return;
-  }
-  should_run_.store(true);
-  monitor_thread_ = std::thread(&TopicMonitor::MonitorLoop, this);
-}
-
-void TopicMonitor::Stop() {
-  if (!should_run_.load()) {
-    return;
-  }
-  should_run_.store(false);
-  if (monitor_thread_.joinable()) {
-    monitor_thread_.join();
-  }
-}
-TopicMonitor::TopicMonitor() : context_(1), subscriber_(context_, ZMQ_SUB), should_run_(false) {
-  try {
-    subscriber_.connect(Messenger::GetInstance().GetPubEndpoint());
-    subscriber_.set(zmq::sockopt::subscribe, "");
-  } catch (const zmq::error_t &e) {
-    LOG(ERROR) << "[TopicMonitor] Initialization failed: " << e.what();
-    throw;
-  }
-}
-TopicMonitor::~TopicMonitor() {
-  Stop();
-  try {
-    subscriber_.close();
-    context_.close();
-  } catch (const zmq::error_t &e) {
-    LOG(ERROR) << "[TopicMonitor] Cleanup error: " << e.what();
-  }
-}
-void TopicMonitor::MonitorLoop() {
-  zmq::message_t msg;
-
-  while (should_run_.load()) {
-    try {
-      if (subscriber_.recv(msg, zmq::recv_flags::dontwait)) {
-        {
-          std::string topic(static_cast<char *>(msg.data()), msg.size());
-          std::lock_guard lock(topics_mutex_);
-          topic_frequencies_[topic]++;
-        }
-        if (subscriber_.get(zmq::sockopt::rcvmore)) {
-          zmq::message_t dummy;
-          subscriber_.recv(dummy);
-        }
-      }
-    } catch (const zmq::error_t &e) {
-      if (e.num() != ETERM) {
-      }
-    }
-  }
-}
-
 }  // namespace infinite_sense

tools/CMakeLists.txt

@@ -0,0 +1,4 @@
+cmake_minimum_required(VERSION 3.10)
+project(example)
+
+add_subdirectory(topic_monitor)

tools/topic_monitor/CMakeLists.txt

@@ -0,0 +1,19 @@
+cmake_minimum_required(VERSION 3.16)
+
+project(monitor VERSION 1.0)
+
+set(CMAKE_CXX_STANDARD 17)
+if (NOT CMAKE_BUILD_TYPE)
+    set(CMAKE_BUILD_TYPE Release)
+    set(CMAKE_CXX_FLAGS "${CMAKE_C_FLAGS} -O3  -Wall")
+    set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -O3  -Wall")
+endif ()
+
+add_executable(${PROJECT_NAME}
+        main.cpp
+        monitor.cpp
+        monitor.h
+)
+target_link_libraries(${PROJECT_NAME} PRIVATE
+        infinite_sense_core
+)

tools/topic_monitor/main.cpp

@@ -0,0 +1,10 @@
+#include "infinite_sense.h"
+#include "monitor.h"
+using namespace infinite_sense;
+int main() {
+  TopicMonitor::GetInstance().Start();
+  std::this_thread::sleep_for(std::chrono::milliseconds{1001});
+  TopicMonitor::GetInstance().Stop();
+  LOG(INFO) << TopicMonitor::GetInstance();
+  return 0;
+}

tools/topic_monitor/monitor.cpp

@@ -0,0 +1,63 @@
+#include "monitor.h"
+#include "infinite_sense.h"
+std::unordered_set<std::string> TopicMonitor::GetTopics() const {
+  std::lock_guard lock(topics_mutex_);
+  return topics_;
+}
+void TopicMonitor::Start() {
+  if (monitor_thread_.joinable()) {
+    return;
+  }
+  should_run_.store(true);
+  monitor_thread_ = std::thread(&TopicMonitor::MonitorLoop, this);
+}
+
+void TopicMonitor::Stop() {
+  if (!should_run_.load()) {
+    return;
+  }
+  should_run_.store(false);
+  if (monitor_thread_.joinable()) {
+    monitor_thread_.join();
+  }
+}
+TopicMonitor::TopicMonitor() : context_(1), subscriber_(context_, ZMQ_SUB), should_run_(false) {
+  try {
+    subscriber_.connect(infinite_sense::Messenger::GetInstance().GetPubEndpoint());
+    subscriber_.set(zmq::sockopt::subscribe, "");
+  } catch (const zmq::error_t &e) {
+    LOG(ERROR) << "[TopicMonitor] Initialization failed: " << e.what();
+    throw;
+  }
+}
+TopicMonitor::~TopicMonitor() {
+  Stop();
+  try {
+    subscriber_.close();
+    context_.close();
+  } catch (const zmq::error_t &e) {
+    LOG(ERROR) << "[TopicMonitor] Cleanup error: " << e.what();
+  }
+}
+void TopicMonitor::MonitorLoop() {
+  zmq::message_t msg;
+
+  while (should_run_.load()) {
+    try {
+      if (subscriber_.recv(msg, zmq::recv_flags::dontwait)) {
+        {
+          std::string topic(static_cast<char *>(msg.data()), msg.size());
+          std::lock_guard lock(topics_mutex_);
+          topic_frequencies_[topic]++;
+        }
+        if (subscriber_.get(zmq::sockopt::rcvmore)) {
+          zmq::message_t dummy;
+          subscriber_.recv(dummy);
+        }
+      }
+    } catch (const zmq::error_t &e) {
+      if (e.num() != ETERM) {
+      }
+    }
+  }
+}