#include "http_action.h" #include #include #include #include #include #include #include #include #include "utils/logger.h" #if __has_include() #include #define HAS_CURL 1 #else #define HAS_CURL 0 #endif namespace rk3588 { namespace { std::string ToLower(std::string s) { for (char& c : s) c = static_cast(std::tolower(static_cast(c))); return s; } std::string JsonEscape(std::string_view s) { std::string out; out.reserve(s.size() + 8); for (char c : s) { switch (c) { case '\\': out += "\\\\"; break; case '"': out += "\\\""; break; case '\n': out += "\\n"; break; case '\r': out += "\\r"; break; case '\t': out += "\\t"; break; default: if (static_cast(c) < 0x20) out += "?"; else out.push_back(c); } } return out; } #if HAS_CURL void EnsureCurlGlobalInitOnce() { static std::once_flag once; std::call_once(once, []() { curl_global_init(CURL_GLOBAL_DEFAULT); std::atexit([]() { curl_global_cleanup(); }); }); } #endif } // namespace HttpAction::~HttpAction() { Drain(); Stop(); } bool HttpAction::Init(const SimpleJson& config) { url_ = config.ValueOr("url", ""); timeout_ms_ = config.ValueOr("timeout_ms", 3000); include_media_url_ = config.ValueOr("include_media_url", true); method_ = config.ValueOr("method", "POST"); max_queue_size_ = static_cast(std::max(0, config.ValueOr("max_queue_size", static_cast(max_queue_size_)))); const std::string policy = config.ValueOr("queue_policy", "drop_oldest"); if (policy == "drop_newest") queue_policy_ = QueuePolicy::DropNewest; else queue_policy_ = QueuePolicy::DropOldest; if (url_.empty()) { LogError("[HttpAction] url is required"); return false; } #if HAS_CURL EnsureCurlGlobalInitOnce(); #endif running_.store(true); worker_ = std::thread(&HttpAction::WorkerLoop, this); LogInfo("[HttpAction] initialized, url=" + url_ + ", method=" + method_); return true; } void HttpAction::Execute(AlarmEvent& event, std::shared_ptr /*frame*/) { // Build JSON body std::ostringstream oss; oss << "{"; oss << "\"node_id\":\"" << JsonEscape(event.node_id) << "\","; oss << "\"rule_name\":\"" << JsonEscape(event.rule_name) << "\","; oss << "\"timestamp\":" << event.timestamp_ms << ","; oss << "\"frame_id\":" << event.frame_id << ","; if (include_media_url_) { if (!event.snapshot_url.empty()) { oss << "\"snapshot_url\":\"" << JsonEscape(event.snapshot_url) << "\","; } if (!event.clip_url.empty()) { oss << "\"clip_url\":\"" << JsonEscape(event.clip_url) << "\","; } } oss << "\"detections\":["; for (size_t i = 0; i < event.detections.size(); ++i) { const auto& det = event.detections[i]; if (i > 0) oss << ","; oss << "{\"cls_id\":" << det.cls_id << ",\"score\":" << det.score << ",\"bbox\":{\"x\":" << det.bbox.x << ",\"y\":" << det.bbox.y << ",\"w\":" << det.bbox.w << ",\"h\":" << det.bbox.h << "}" << ",\"track_id\":" << det.track_id << "}"; } oss << "]}"; std::string json_body = oss.str(); { std::lock_guard lock(queue_mutex_); if (max_queue_size_ > 0 && request_queue_.size() >= max_queue_size_) { ++dropped_total_; if (queue_policy_ == QueuePolicy::DropOldest) { request_queue_.pop(); request_queue_.push(json_body); } else { // DropNewest: drop this request } } else { request_queue_.push(json_body); } } queue_cv_.notify_one(); } void HttpAction::Drain() { if (!running_.load()) return; // Wait for queue to be fully drained (including in-flight request). std::unique_lock lock(queue_mutex_); queue_cv_.wait_for(lock, std::chrono::seconds(5), [this] { return request_queue_.empty() && in_flight_ == 0; }); } void HttpAction::Stop() { bool was_running = running_.exchange(false); (void)was_running; queue_cv_.notify_all(); if (worker_.joinable()) worker_.join(); } void HttpAction::WorkerLoop() { while (running_.load()) { std::string body; { std::unique_lock lock(queue_mutex_); queue_cv_.wait_for(lock, std::chrono::milliseconds(100), [this] { return !request_queue_.empty() || !running_.load(); }); if (!running_.load() && request_queue_.empty()) break; if (request_queue_.empty()) continue; body = std::move(request_queue_.front()); request_queue_.pop(); ++in_flight_; } if (!body.empty()) { SendRequest(body); } { std::lock_guard lock(queue_mutex_); if (in_flight_ > 0) --in_flight_; } queue_cv_.notify_all(); } } bool HttpAction::SendRequest(const std::string& json_body) { #if HAS_CURL CURL* curl = curl_easy_init(); if (!curl) { LogError("[HttpAction] curl_easy_init failed"); return false; } struct curl_slist* headers = nullptr; headers = curl_slist_append(headers, "Content-Type: application/json"); curl_easy_setopt(curl, CURLOPT_URL, url_.c_str()); curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headers); const std::string method_lc = ToLower(method_); if (method_lc == "post") { curl_easy_setopt(curl, CURLOPT_POST, 1L); curl_easy_setopt(curl, CURLOPT_POSTFIELDS, json_body.c_str()); } else if (method_lc == "put" || method_lc == "patch") { curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST, method_.c_str()); curl_easy_setopt(curl, CURLOPT_POSTFIELDS, json_body.c_str()); } else { LogWarn("[HttpAction] unsupported method='" + method_ + "', falling back to POST"); curl_easy_setopt(curl, CURLOPT_POST, 1L); curl_easy_setopt(curl, CURLOPT_POSTFIELDS, json_body.c_str()); } curl_easy_setopt(curl, CURLOPT_TIMEOUT_MS, timeout_ms_); curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT_MS, timeout_ms_ / 2); CURLcode res = curl_easy_perform(curl); if (res != CURLE_OK) { LogWarn(std::string("[HttpAction] request failed: ") + curl_easy_strerror(res)); } curl_slist_free_all(headers); curl_easy_cleanup(curl); return res == CURLE_OK; #else // Stub implementation for Windows without curl LogInfo("[HttpAction] (stub) method=" + method_ + " url=" + url_ + " body=" + json_body); return true; #endif } } // namespace rk3588