// main.cpp #include "network/tcp_server.h" #include "mqtt/mqtt_client.h" #include "mqtt/mqtt_router.h" #include "systemMonitor/system_monitor.h" #include "spdlog/spdlog.h" #include "deviceManager/device_manager.h" #include "dataCache/data_cache.h" #include "dataCache/cache_uploader.h" #include "web/web_server.h" #include #include #include #include #include // 用于 ASIO 服务的全局 io_context boost::asio::io_context g_io_context; /** * @brief 处理终止信号 (SIGINT, SIGTERM). */ // void signalHandler(int signum) { // spdlog::warn("Interrupt signal ({}) received. Shutting down.", signum); // g_io_context.stop(); // } /** * @brief 周期性轮询系统状态并发布到 MQTT */ void poll_system_metrics( boost::asio::steady_timer& timer, SystemMonitor::SystemMonitor& monitor, MqttClient& mqtt_client ) { if (g_io_context.stopped()) return; auto cpu_util = monitor.getCpuUtilization(); auto mem_info = monitor.getMemoryInfo(); double mem_total_gb = mem_info.total_kb / 1024.0 / 1024.0; std::string topic = "proxy/system_status"; std::string payload = "{\"cpu_usage\":" + std::to_string(cpu_util.totalUsagePercentage) + ",\"mem_total_gb\":" + std::to_string(mem_total_gb) + "}"; mqtt_client.publish(topic, payload); spdlog::debug("System metrics published."); timer.expires_at(timer.expiry() + std::chrono::seconds(15)); timer.async_wait(std::bind(poll_system_metrics, std::ref(timer), std::ref(monitor), std::ref(mqtt_client))); } int main(int argc, char* argv[]) { try { spdlog::set_level(spdlog::level::debug); spdlog::info("Edge Proxy starting up..."); } catch (const spdlog::spdlog_ex& ex) { std::cerr << "Log initialization failed: " << ex.what() << std::endl; return 1; } // signal(SIGINT, signalHandler); // signal(SIGTERM, signalHandler); try { MqttClient mqtt_client("tcp://mqtt-broker:1883", "edge-proxy-main-client"); MqttRouter mqtt_router(mqtt_client); std::vector listen_ports = { 8888 }; TCPServer tcp_server(g_io_context, listen_ports, mqtt_client); SystemMonitor::SystemMonitor monitor; DataCache data_cache; if (!data_cache.open("edge_data_cache.db")) { spdlog::critical("Failed to initialize data cache. Exiting."); return 1; } CacheUploader cache_uploader(g_io_context, mqtt_client, data_cache); mqtt_client.set_connected_handler([&](const std::string& cause){ spdlog::info("MQTT client connected: {}", cause); cache_uploader.start_upload(); }); mqtt_client.connect(); mqtt_router.start(); monitor.getCpuUtilization(); boost::asio::steady_timer system_monitor_timer(g_io_context, std::chrono::seconds(15)); system_monitor_timer.async_wait(std::bind(poll_system_metrics, std::ref(system_monitor_timer), std::ref(monitor), std::ref(mqtt_client))); // --- 创建包含缓存逻辑的统一数据上报回调函数 >>> auto report_to_mqtt = [&](const UnifiedData& data) { if (mqtt_client.is_connected()) { // 网络正常,直接上报 std::string topic = "devices/" + data.device_id + "/data"; g_io_context.post([&, topic, payload = data.data_json]() { mqtt_client.publish(topic, payload, 1, false); }); } else { // 网络断开,写入缓存 spdlog::warn("MQTT disconnected. Caching data for device '{}'.", data.device_id); data_cache.add(data); } }; // 实例化设备管理器并从文件加载所有设备 (使用新的回调) DeviceManager device_manager(g_io_context, report_to_mqtt); device_manager.load_and_start("../config/devices.json"); WebServer web_server(monitor, 8080); web_server.start(); // --- 2. <<< MODIFIED: 设置 Boost.Asio 风格的信号处理器 >>> --- boost::asio::signal_set signals(g_io_context, SIGINT, SIGTERM); signals.async_wait([&](const boost::system::error_code& error, int signal_number) { spdlog::warn("Interrupt signal ({}) received. Shutting down.", signal_number); // a. 首先停止所有数据采集线程 spdlog::info("[Shutdown] A. Stopping device manager services..."); device_manager.stop_all(); // b. 停止Web服务器线程 spdlog::info("[Shutdown] B. Stopping web server..."); web_server.stop(); // c. 断开MQTT连接 (这将释放它对io_context的'劫持') spdlog::info("[Shutdown] C. Disconnecting from MQTT broker..."); mqtt_client.disconnect(); // d. 最后,安全地停止io_context spdlog::info("[Shutdown] D. Stopping main event loop..."); g_io_context.stop(); }); spdlog::info("All services are running. Press Ctrl+C to exit."); g_io_context.run(); } catch (const std::exception& e) { spdlog::critical("An unhandled exception occurred: {}", e.what()); return 1; } spdlog::info("Server has been shut down gracefully. Exiting."); return 0; }