#include #include #include #include #include #include #include #include "config.h" #include "scheduler.h" #include "reporter.h" #include "http_server.h" using namespace jtjai_media; // 全局变量用于信号处理 static std::unique_ptr g_scheduler; static std::unique_ptr g_http_server; static std::atomic g_interrupted(false); // 信号处理函数 void signal_handler(int signal) { std::cout << "\n接收到信号 " << signal << ",正在停止程序..." << std::endl; g_interrupted.store(true); if (g_scheduler) { g_scheduler->stop_execution(); } if (g_http_server) { g_http_server->stop(); } } // 进度回调函数 void progress_callback(const StreamScheduler::SchedulerStats& stats) { std::cout << "\r进度: " << stats.completed_tasks << "/" << stats.total_tasks << " 完成, " << stats.failed_tasks << " 失败, " << "当前并发: " << stats.max_concurrent_used << " 流" << std::flush; } // 生成时间戳字符串 std::string generate_timestamp() { auto now = std::chrono::system_clock::now(); auto time_t = std::chrono::system_clock::to_time_t(now); auto tm = *std::localtime(&time_t); std::stringstream ss; ss << std::put_time(&tm, "%Y%m%d_%H%M%S"); return ss.str(); } // 报告生成回调函数 void report_callback(const ConfigManager& config_mgr, int cycle_number, const std::string& cycle_output_dir, const std::vector& client_stats, const StreamScheduler::SchedulerStats& scheduler_stats, bool was_timeout = false) { std::cout << "生成第 " << cycle_number << " 轮执行报告"; if (was_timeout) { std::cout << " (⚠️ 超时强制结束)"; } std::cout << "..." << std::endl; // 为每个周期使用独立的时间戳目录 ConfigManager cycle_config = config_mgr; auto cycle_global_config = cycle_config.get_global_config(); // 使用传入的时间戳目录 cycle_global_config.output_directory = cycle_output_dir; cycle_config.set_global_config(cycle_global_config); ResultReporter cycle_reporter(cycle_config); if (cycle_reporter.generate_report(client_stats, scheduler_stats)) { std::cout << "第 " << cycle_number << " 轮报告生成成功"; if (was_timeout) { std::cout << " (包含超时标记)"; } std::cout << std::endl; std::cout << "输出目录: " << cycle_output_dir << std::endl; } else { std::cerr << "第 " << cycle_number << " 轮报告生成失败" << std::endl; } } int main(int argc, char* argv[]) { std::cout << "========================================" << std::endl; std::cout << "RTSP视频流并发拉取系统" << std::endl; std::cout << "========================================" << std::endl; // 注册信号处理器 signal(SIGINT, signal_handler); signal(SIGTERM, signal_handler); // 解析命令行参数 std::string config_file = "/Users/wenhongquan/CLionProjects/jtjai_media/config.json"; if (argc > 1) { config_file = argv[1]; } std::cout << "使用配置文件: " << config_file << std::endl; try { // 加载配置 ConfigManager config_mgr; if (!config_mgr.load_from_file(config_file)) { std::cerr << "加载配置文件失败: " << config_file << std::endl; return 1; } std::cout << "配置加载成功" << std::endl; std::cout << config_mgr.to_string() << std::endl; const auto& global_config = config_mgr.get_global_config(); // 启动HTTP服务器 std::cout << "\n启动HTTP管理服务器..." << std::endl; g_http_server = std::make_unique(global_config.output_directory, 8080); if (!g_http_server->start()) { std::cerr << "启动HTTP服务器失败" << std::endl; return 1; } std::cout << "\n🌐 Web管理界面: http://localhost:8080" << std::endl; std::cout << "📁 文件管理: http://localhost:8080/manager" << std::endl; std::cout << "📊 API文档: http://localhost:8080/api" << std::endl; int current_cycle = 0; // 开始多轮询循环 - 使用定时器强制切换 auto start_time = std::chrono::system_clock::now(); while ((global_config.poll_cycles == -1 || current_cycle < global_config.poll_cycles) && !g_interrupted.load()) { current_cycle++; std::cout << "\n========== 轮询周期 " << current_cycle; if (global_config.poll_cycles != -1) { std::cout << "/" << global_config.poll_cycles; } std::cout << " ==========" << std::endl; // 计算本轮应该开始的时间 auto cycle_start_time = start_time + std::chrono::seconds((current_cycle - 1) * global_config.total_poll_duration_seconds); auto now = std::chrono::system_clock::now(); // 如果还没到时间,等待 if (cycle_start_time > now) { auto wait_duration = std::chrono::duration_cast(cycle_start_time - now).count(); std::cout << "等待 " << wait_duration << " 秒后开始本轮任务..." << std::endl; for (int i = 0; i < wait_duration && !g_interrupted.load(); ++i) { std::this_thread::sleep_for(std::chrono::seconds(1)); if ((i + 1) % 10 == 0) { std::cout << "\r剩余等待时间: " << (wait_duration - i - 1) << " 秒" << std::flush; } } std::cout << std::endl; } if (g_interrupted.load()) { break; } // 强制停止上一轮的调度器(如果还在运行) if (g_scheduler && g_scheduler->is_running()) { std::cout << "⚠️ 强制停止上一轮任务(超时)" << std::endl; g_scheduler->stop_execution(); g_scheduler->wait_for_completion(); } // 为当前周期创建时间戳目录 std::string timestamp = generate_timestamp(); std::string base_output_dir = global_config.output_directory; std::string cycle_output_dir = base_output_dir + "/" + timestamp; // 创建时间戳目录 std::filesystem::create_directories(cycle_output_dir); std::cout << "当前周期输出目录: " << cycle_output_dir << std::endl; std::cout << "⏱️ 本轮最大执行时间: " << global_config.total_poll_duration_seconds << " 秒" << std::endl; // 更新配置管理器的输出目录 ConfigManager cycle_config_mgr = config_mgr; auto cycle_global_config = cycle_config_mgr.get_global_config(); cycle_global_config.output_directory = cycle_output_dir; cycle_config_mgr.set_global_config(cycle_global_config); // 使用新的配置创建调度器 g_scheduler = std::make_unique(cycle_config_mgr); g_scheduler->set_progress_callback(progress_callback); bool timeout_reached = false; // 设置当前周期的报告回调 g_scheduler->set_report_callback([&config_mgr, current_cycle, cycle_output_dir, &timeout_reached]( const std::vector& client_stats, const StreamScheduler::SchedulerStats& scheduler_stats) { report_callback(config_mgr, current_cycle, cycle_output_dir, client_stats, scheduler_stats, timeout_reached); }); // 启动调度执行 std::cout << "开始执行RTSP流拉取任务..." << std::endl; auto task_start_time = std::chrono::system_clock::now(); if (!g_scheduler->start_execution()) { std::cerr << "启动调度器失败" << std::endl; return 1; } // 等待执行完成或超时 std::cout << "等待任务执行中(最大 " << global_config.total_poll_duration_seconds << " 秒)..." << std::endl; while (g_scheduler->is_running() && !g_interrupted.load()) { auto elapsed = std::chrono::duration_cast( std::chrono::system_clock::now() - task_start_time).count(); if (elapsed >= global_config.total_poll_duration_seconds) { std::cout << "\n⏰ 达到最大执行时间(" << global_config.total_poll_duration_seconds << "秒),强制结束本轮任务" << std::endl; timeout_reached = true; break; } // 显示剩余时间 int remaining = global_config.total_poll_duration_seconds - elapsed; if (remaining % 10 == 0 && remaining > 0) { std::cout << "\r剩余执行时间: " << remaining << " 秒" << std::flush; } std::this_thread::sleep_for(std::chrono::milliseconds(1000)); } // 强制停止任务并生成报告 if (timeout_reached || g_scheduler->is_running()) { std::cout << "\n强制停止调度器..." << std::endl; g_scheduler->stop_execution(); } std::cout << "等待所有任务线程结束..." << std::endl; g_scheduler->wait_for_completion(); // 检查是否需要继续下一轮 if (g_interrupted.load()) { std::cout << "接收到中断信号,退出轮询" << std::endl; break; } } std::cout << "\n程序执行完成" << std::endl; } catch (const std::exception& e) { std::cerr << "程序执行异常: " << e.what() << std::endl; return 1; } return 0; }