You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

941 lines
40 KiB
C++

This file contains ambiguous Unicode characters!

This file contains ambiguous Unicode characters that may be confused with others in your current locale. If your use case is intentional and legitimate, you can safely ignore this warning. Use the Escape button to highlight these characters.

// tcp_server.cpp - TCP 服务端(新版统一扫描协议)
//
// 消息以 '>' 结尾(不是换行)。每条命令是一个 JSON:
// {"function":"count|check|config",
// "scan":"start|stop",
// "scanTime":N, // >0: 到时自动停止,无需再发 stop(单位:秒)
// "box":{"l":..,"w":..,"h":..,"maxPerLayer":..},
// "task_id":".."}
// 语义:
// - function=count/check + scan=start:启动统一后台扫描;若扫描已在运行,则
// 强制重启(旧扫描结果丢弃);scanTime>0 时到点自动停止并返回结果。
// - function=count/check + scan=stop:立即停止当前扫描并返回结果。
// - function=config:重新加载配置文件。
// count/check 共用同一个 ScanSession,同一时刻只有一个采集线程访问雷达。
#include "tcp_server.h"
#include "config.h"
#include "json_utils.h"
#include "lidar_utils.h"
#include "count.h"
#include "detect.h"
#include "io_utils.h"
#include <winsock2.h>
#include <ws2tcpip.h>
#include <atomic>
#include <chrono>
#include <cctype>
#include <condition_variable>
#include <cstdlib>
#include <cstring>
#include <iostream>
#include <iomanip>
#include <map>
#include <mutex>
#include <set>
#include <sstream>
#include <thread>
namespace
{
// ==================== 会话状态(每雷达一份,各雷达可并行扫描) ====================
struct SessionState
{
bool active = false; // 是否有扫描在运行
bool auto_stop = false; // 是否到时自动停止
std::string radar; // 所属雷达标识
std::string function; // "count" / "check"
CountParams box; // count 的 box 参数
std::string task_id;
SOCKET owner = INVALID_SOCKET; // 发起者(自动停止结果回发)
std::chrono::steady_clock::time_point deadline; // 自动停止时刻
};
static std::map<std::string, SessionState> g_states; // 雷达标识 -> 会话状态
static std::mutex g_life; // 会话生命周期锁
static std::mutex g_cv_mtx; // 自动停止等待锁
static std::condition_variable g_cv;
static std::atomic<bool> g_stop{false}; // 服务端退出标志(自动停止线程用)
// ==================== 连接管理 ====================
// 不设 recv 超时时,任何"连上但不发数据"的连接(端口探测/异常客户端/旧TCP客户端
// 连到HTTP端口)都会让处理线程永久阻塞在 recv 上;线程越堆越多,最终
// std::thread 构造失败抛 std::system_error,未捕获的异常会直接终止整个进程
// —— 表现为"先卡死、然后进程突然消失且无任何日志"。
static const int MAX_CONNECTIONS = 64; // 并发连接上限
static const int TCP_RECV_TIMEOUT_MS = 300000; // TCP 长连接空闲上限 5 分钟
static const int HTTP_RECV_TIMEOUT_MS = 5000; // HTTP 头部读取超时(超时按TCP协议兜底)
static const int SEND_TIMEOUT_MS = 5000; // 发送超时
static std::atomic<int> g_conn_count{0}; // 当前活跃连接数
// ==================== 工作线程生命周期 ====================
// auto_stop_loop / 客户端处理线程都是 detach 的。服务端主循环一旦返回,main 会
// 立刻 radar_mgr.shutdown() 释放 RadarNode 与 driver,而这些线程仍在访问它们
// —— 典型表现就是"进程突然消失、没有任何日志"。这里统计活跃工作线程,
// 退出前先置停止位、唤醒阻塞在 recv 的连接、再等待线程真正结束。
static std::atomic<int> g_thread_count{0}; // 活跃工作线程数
static std::mutex g_sock_mtx;
static std::set<SOCKET> g_active_socks; // 活跃客户端 socket
struct WorkerGuard
{
WorkerGuard() { ++g_thread_count; }
~WorkerGuard() { --g_thread_count; }
};
// 服务端收尾:通知退出 + 唤醒阻塞连接 + 等待线程结束(最多 5 秒,避免卡住退出流程)
static void stop_server_workers()
{
g_stop = true;
{
std::lock_guard<std::mutex> lk(g_sock_mtx);
for (SOCKET s : g_active_socks)
{
if (s == INVALID_SOCKET) continue;
#ifdef _WIN32
shutdown(s, SD_BOTH); // 让阻塞的 recv 立即返回
#else
shutdown(s, SHUT_RDWR);
#endif
}
}
for (int i = 0; i < 100 && g_thread_count.load() > 0; ++i)
std::this_thread::sleep_for(std::chrono::milliseconds(50));
if (g_thread_count.load() > 0)
std::cout << "[警告] 仍有 " << g_thread_count.load()
<< " 个工作线程未退出,退出过程可能不稳定" << std::endl;
else
std::cout << "[关闭] 服务端工作线程已全部退出" << std::endl;
}
static void set_sock_timeouts(SOCKET s, int recv_ms, int send_ms)
{
if (s == INVALID_SOCKET) return;
#ifdef _WIN32
DWORD r = (recv_ms > 0) ? (DWORD)recv_ms : 0;
DWORD w = (send_ms > 0) ? (DWORD)send_ms : 0;
setsockopt(s, SOL_SOCKET, SO_RCVTIMEO, (const char*)&r, sizeof(r));
setsockopt(s, SOL_SOCKET, SO_SNDTIMEO, (const char*)&w, sizeof(w));
#else
struct timeval rv; rv.tv_sec = recv_ms / 1000; rv.tv_usec = (recv_ms % 1000) * 1000;
struct timeval wv; wv.tv_sec = send_ms / 1000; wv.tv_usec = (send_ms % 1000) * 1000;
setsockopt(s, SOL_SOCKET, SO_RCVTIMEO, (const char*)&rv, sizeof(rv));
setsockopt(s, SOL_SOCKET, SO_SNDTIMEO, (const char*)&wv, sizeof(wv));
#endif
}
// 连接句柄守卫:任何退出路径(含异常)都保证关闭 socket 并归还连接配额
class ConnGuard
{
public:
explicit ConnGuard(SOCKET s) : s_(s)
{
std::lock_guard<std::mutex> lk(g_sock_mtx);
g_active_socks.insert(s_); // 登记,供退出时统一唤醒
}
~ConnGuard()
{
{
std::lock_guard<std::mutex> lk(g_sock_mtx);
g_active_socks.erase(s_);
}
if (s_ != INVALID_SOCKET) closesocket(s_);
--g_conn_count;
}
private:
SOCKET s_;
};
// ==================== 发送 ====================
// 响应统一以 '>' 结尾
static void send_text(SOCKET s, const std::string& msg)
{
std::string out = msg;
if (!out.empty() && out.back() != '>') out += '>';
send(s, out.c_str(), (int)out.size(), 0);
}
// 响应出口:resp==nullptr 直接发到 client(TCP 模式);否则收集到 resp(HTTP 模式)。
static void send_reply(SOCKET client, std::string* resp, const std::string& msg)
{
if (resp) *resp = msg;
else send_text(client, msg);
}
// ==================== 停止并取回结果 ====================
// id: 雷达标识;need_auto=true 表示"自动到时"触发,只允许在 auto_stop 仍武装时抢到;
// 否则(手动 stop)只要 active 即可。
static bool claim_stop_and_fetch(const std::string& id, bool need_auto,
SessionState* snap,
std::vector<CountPoint>* pts)
{
RadarNode* node = radar_mgr.find(id);
std::lock_guard<std::mutex> lk(g_life);
auto it = g_states.find(id);
if (it == g_states.end() || !it->second.active) return false;
if (need_auto && !it->second.auto_stop) return false; // 已被手动 stop 抢先
*snap = it->second;
it->second.active = false;
it->second.auto_stop = false;
g_cv.notify_all();
if (node)
{
node->session.stop(); // 等待采集线程退出(≤100ms轮询粒度)
*pts = node->session.snapshot();
node->session.reset(); // 清空点集,为下一次 start 准备
}
return true;
}
// 仅停止指定雷达的当前扫描,不返回结果(用于 start 时的强制重启)。
// 调用方需已持有 g_life。
static bool discard_current_locked(const std::string& id)
{
auto it = g_states.find(id);
if (it == g_states.end() || !it->second.active) return false;
it->second.active = false;
it->second.auto_stop = false;
g_cv.notify_all();
RadarNode* node = radar_mgr.find(id);
if (node)
{
node->session.stop();
node->session.reset();
}
return true;
}
// ==================== 结果计算与回包 ====================
static void finish_count(const SessionState& snap,
const std::vector<CountPoint>& pts,
SOCKET target, std::string* resp = nullptr)
{
RadarScope scope(snap.radar); // 绑定该雷达配置(分层/多边形/滤波)
bool valid_box = snap.box.l > 0 && snap.box.w > 0 &&
snap.box.h > 0 && snap.box.maxPerLayer > 0;
std::vector<CountPoint> work = pts;
std::string detail;
std::vector<uint8_t> top_img;
int total = 0, tw = 0, th = 0, tlv = 0;
double tarea = 0.0;
if (!pts.empty() && valid_box)
{
sparse_density_filter(work); // 点云级稀疏拖影过滤(默认开, 配置可调)
z_gap_interpolate(work); // z空隙插值:只在停止后执行一次
total = calc_item_count(work, snap.box, &detail,
&top_img, &tw, &th, &tlv, &tarea);
}
std::cout << "[TCP][" << snap.radar << "] count结束: task_id=" << snap.task_id
<< ", 点数=" << pts.size() << ", 结果=" << total << std::endl;
std::ostringstream r;
r << "{\"code\":200,\"message\":\"count done\",\"count\":" << total;
if (!snap.task_id.empty())
r << ",\"task_id\":\"" << snap.task_id << "\"";
r << ",\"radar\":\"" << snap.radar << "\"}";
send_reply(target, resp, r.str());
// 后台保存 PCD/图片/日志(不阻塞响应)
// 最后一个参数 pts 是"插值前"的原始点集,用于另存 count_data_before.pcd 做对比
if (!pts.empty() && valid_box)
save_count_files_async(snap.radar, work, top_img, tw, th, snap.box, detail,
total, tlv, tarea, &pts);
}
static void finish_check(const SessionState& snap,
const std::vector<CountPoint>& pts,
SOCKET target, std::string* resp = nullptr)
{
RadarScope scope(snap.radar); // 绑定该雷达配置(截面参数/多边形)
// 输出目录 data/日期/时间_taskId(由 io_utils 统一创建)
std::string out_dir = make_task_dir(snap.task_id);
DetectResult res;
detect_analyze_slices(pts, out_dir, &res);
std::ostringstream r;
r << "{\"code\":200,\"message\":\"detect done\"";
if (res.valid)
{
if (res.start_fill_pct >= 0.0)
r << ",\"start\":{\"fill_pct\":"
<< std::fixed << std::setprecision(2) << res.start_fill_pct
<< ",\"y_max\":" << std::setprecision(1) << res.start_y_max
<< ",\"y_median\":" << res.start_y_med << "}";
if (res.end_fill_pct >= 0.0)
r << ",\"end\":{\"fill_pct\":"
<< std::fixed << std::setprecision(2) << res.end_fill_pct
<< ",\"y_max\":" << std::setprecision(1) << res.end_y_max
<< ",\"y_median\":" << res.end_y_med << "}";
}
if (!snap.task_id.empty())
r << ",\"task_id\":\"" << snap.task_id << "\"";
r << ",\"radar\":\"" << snap.radar << "\"}";
send_reply(target, resp, r.str());
std::cout << "[TCP][" << snap.radar << "] check结束: task_id=" << snap.task_id
<< ", 点数=" << pts.size()
<< (res.valid ? ", 检测有效" : ", 检测无效") << std::endl;
}
static void finish_and_reply(const SessionState& snap,
const std::vector<CountPoint>& pts,
SOCKET target, std::string* resp = nullptr)
{
if (snap.function == "count")
finish_count(snap, pts, target, resp);
else if (snap.function == "check")
finish_check(snap, pts, target, resp);
}
// ==================== 自动到时停止线程 ====================
// 多雷达:每 100ms 轮询一次所有雷达的会话,到点的各自独立停止并回包。
static void auto_stop_loop()
{
WorkerGuard wg; // 计入活跃工作线程,退出时被等待
while (!g_stop.load())
{
std::this_thread::sleep_for(std::chrono::milliseconds(100));
std::vector<std::string> due;
{
std::lock_guard<std::mutex> lk(g_life);
auto now = std::chrono::steady_clock::now();
for (const auto& kv : g_states)
{
const SessionState& st = kv.second;
if (st.active && st.auto_stop && now >= st.deadline)
due.push_back(kv.first);
}
}
for (const auto& id : due)
{
SessionState snap;
std::vector<CountPoint> pts;
if (!claim_stop_and_fetch(id, true, &snap, &pts)) continue;
std::cout << "[TCP][" << id << "] 扫描到时自动停止: function=" << snap.function
<< ", task_id=" << snap.task_id << std::endl;
// 异常必须就地吞掉:本线程是 detached,异常逃逸会直接终止进程
try
{
RadarScope scope(id);
if (snap.owner != INVALID_SOCKET)
finish_and_reply(snap, pts, snap.owner);
else
{
// HTTP 模式无长连接可推送,结果只打印并留痕
std::string out;
finish_and_reply(snap, pts, INVALID_SOCKET, &out);
std::cout << "[HTTP][" << id << "] scanTime到时自动停止, 无连接可推送, 结果: "
<< out << std::endl;
}
}
catch (const std::exception& e)
{
std::cerr << "[TCP][" << id << "] 自动停止处理结果异常: " << e.what() << std::endl;
}
catch (...) { std::cerr << "[TCP][" << id << "] 自动停止处理结果未知异常" << std::endl; }
}
}
}
// ==================== 命令处理 ====================
static void handle_scan_start(const std::string& msg, SOCKET client,
std::string* resp, const std::string& radar_id,
RadarNode* node)
{
std::string fn;
json_get_string(msg, "function", fn);
std::string task_id;
json_get_string(msg, "task_id", task_id);
double scan_time = 0.0;
json_get_number(msg, "scanTime", scan_time);
CountParams box;
json_get_string(msg, "task_id", box.task_id);
json_get_number(msg, "l", box.l);
json_get_number(msg, "w", box.w);
json_get_number(msg, "h", box.h);
double d = 0.0;
if (json_get_number(msg, "maxPerLayer", d) ||
json_get_number(msg, "max", d))
box.maxPerLayer = (int)d;
std::ostringstream ack;
ack << "{\"code\":200,\"message\":\"scan start\",\"function\":\"" << fn << "\"";
if (!task_id.empty())
ack << ",\"task_id\":\"" << task_id << "\"";
if (scan_time > 0)
ack << ",\"scanTime\":" << (int)scan_time;
ack << ",\"radar\":\"" << radar_id << "\"}";
if (!node->connected.load())
{
std::string fail = "radar not connected: " + radar_id;
std::cerr << "[TCP][" << radar_id << "] " << fail << std::endl;
send_reply(client, resp, "{\"code\":503,\"message\":\"" + fail + "\"}");
return;
}
bool started = false;
std::string fail;
{
std::lock_guard<std::mutex> lk(g_life);
// 该雷达扫描运行中收到新的 start:强制重启(旧扫描结果丢弃)
if (discard_current_locked(radar_id))
std::cout << "[TCP][" << radar_id << "] 扫描运行中被新命令强制重启: function=" << fn
<< ", task_id=" << task_id << std::endl;
node->session.reset();
// 传入 io_mtx:采集取帧与后台重连互斥,杜绝两线程并发操作同一 driver
if (!node->session.start(*node->driver, radar_id, &node->io_mtx))
{
fail = "scan start failed";
}
else
{
SessionState ns;
ns.active = true;
ns.radar = radar_id;
ns.function = fn;
ns.box = box;
ns.task_id = task_id;
// TCP 用连接作 owner(自动到时结果回发);HTTP 无长连接,置无效不推送
ns.owner = resp ? INVALID_SOCKET : client;
if (scan_time > 0)
{
ns.auto_stop = true;
ns.deadline = std::chrono::steady_clock::now() +
std::chrono::milliseconds((long)(scan_time * 1000.0));
}
g_states[radar_id] = ns;
g_cv.notify_all();
started = true;
}
}
if (!started)
{
std::cerr << "[TCP][" << radar_id << "] " << fail << std::endl;
send_reply(client, resp, "{\"code\":400,\"message\":\"" + fail + "\"}");
return;
}
std::cout << "[TCP][" << radar_id << "] 扫描开始: function=" << fn
<< ", task_id=" << task_id
<< (scan_time > 0 ? ", 到时自动停止" : ", 等待stop") << std::endl;
send_reply(client, resp, ack.str());
}
static void handle_scan_stop(const std::string& msg, SOCKET client,
std::string* resp, const std::string& radar_id)
{
std::string fn;
json_get_string(msg, "function", fn);
SessionState snap;
std::vector<CountPoint> pts;
if (!claim_stop_and_fetch(radar_id, false, &snap, &pts))
{
// 未在扫描(从未 start / 已停止 / 自动到时已完成)
std::ostringstream r;
r << "{\"code\":200,\"message\":\"scan not running\"";
if (fn == "count")
r << ",\"count\":0";
r << ",\"radar\":\"" << radar_id << "\"}";
send_reply(client, resp, r.str());
return;
}
std::cout << "[TCP][" << radar_id << "] 收到stop: function=" << snap.function
<< ", task_id=" << snap.task_id << std::endl;
finish_and_reply(snap, pts, client, resp);
}
static void handle_config(const std::string& msg, SOCKET client,
std::string* resp = nullptr)
{
(void)msg;
std::cout << "[TCP] 收到config命令,重新读取 config/config ..." << std::endl;
if (!load_config()) // 失败时保留原配置,绝不让进程退出
{
std::cerr << "[TCP] 配置重载失败,沿用现有配置" << std::endl;
send_reply(client, resp,
"{\"code\":500,\"message\":\"config reload failed\"}");
return;
}
reload_speed_profile(); // radar_cfg().scan.z_speed_type=configuration 时重新读取 speed.txt
radar_mgr.sync_config(); // 配置里的雷达增删:新增节点后台连接,删除的节点停用
std::cout << "[TCP] 配置已重载: 雷达" << cfg.ids.size() << "台:";
for (const auto& id : cfg.ids) std::cout << " " << id;
std::cout << std::endl;
std::cout << "[TCP] 配置已重载: 采集=" << radar_cfg().scan.collect_duration_s
<< "s, 速度模式=" << radar_cfg().scan.z_speed_type
<< ", 速度=" << radar_cfg().scan.z_speed << "cm/s, 插值=" << radar_cfg().count.z_interpolate_mode
<< ", 保留=" << cfg.log.retain_days << "天, 端口=" << cfg.server.port << std::endl;
std::cout << "[TCP] count容差: 基准=" << radar_cfg().count.base_tolerance
<< "mm + 每层" << radar_cfg().count.per_level_tolerance << "mm"
<< ", 地板高=" << radar_cfg().count.floor_height << "mm"
<< ", SparseCut=" << (radar_cfg().filter.sparse_cut.enable ? "开" : "关")
<< "(X±" << radar_cfg().filter.sparse_cut.x_radius << "mm,同圈>=" << radar_cfg().filter.sparse_cut.circle_min
<< ",跨圈>=" << radar_cfg().filter.sparse_cut.run_min << ")"
<< ", TailCut=" << (radar_cfg().filter.tail_cut.enable ? "开" : "关") << std::endl;
send_reply(client, resp, "{\"code\":200,\"message\":\"config reloaded\"}");
}
// ConfigureSpeed: 标定生成 config/speed.txt(雷达横放时,每帧Y最大值=行进距离)。
// 请求: {"function":"ConfigureSpeed","scanTime":N},N缺省用 radar_cfg().scan.collect_duration_s。
// 标定期间与扫描会话互斥(全程持 g_life,其他命令排队)。
static void handle_configure_speed(const std::string& msg, SOCKET client,
std::string* resp, const std::string& radar_id,
RadarNode* node)
{
double scan_time = 0.0;
json_get_number(msg, "scanTime", scan_time);
double seconds = (scan_time > 0) ? scan_time : (double)radar_cfg().scan.collect_duration_s;
if (seconds <= 0.0) seconds = 5.0;
std::cout << "[TCP][" << radar_id << "] 收到ConfigureSpeed: 采集" << seconds
<< "s 标定变速剖面(每帧Y最大值) ..." << std::endl;
if (!node->connected.load())
{
std::string fail = "radar not connected: " + radar_id;
std::cerr << "[TCP][" << radar_id << "] " << fail << std::endl;
send_reply(client, resp, "{\"code\":503,\"message\":\"" + fail + "\"}");
return;
}
int rows = -1;
std::string err;
{
std::lock_guard<std::mutex> lk(g_life);
if (discard_current_locked(radar_id)) // 扫描运行中先强制停止(本函数持锁)
std::cout << "[TCP][" << radar_id << "] 扫描运行中被ConfigureSpeed中断" << std::endl;
rows = collect_speed_profile(*node->driver, seconds, err);
}
if (rows < 0)
{
std::cerr << "[TCP][" << radar_id << "] ConfigureSpeed失败: " << err << std::endl;
send_reply(client, resp, "{\"code\":400,\"message\":\"configure speed failed\"}");
return;
}
reload_speed_profile(); // 重新载入新生成的剖面(若已是configuration则立即生效)
std::cout << "[TCP][" << radar_id << "] speed.txt 已更新: " << rows
<< " 帧样本 (采集" << seconds << "s)" << std::endl;
std::ostringstream r;
r << "{\"code\":200,\"message\":\"configure speed done\",\"rows\":" << rows
<< ",\"duration\":" << (int)seconds
<< ",\"radar\":\"" << radar_id << "\""
<< ",\"file\":\"config/speed.txt\"}";
send_reply(client, resp, r.str());
}
// 处理一条命令。resp==nullptr: 直接发送(TCP);resp!=nullptr: 响应写入该串(HTTP)。
// 返回值: 是否产生响应。
static bool dispatch(const std::string& raw, SOCKET client,
std::string* resp = nullptr)
{
std::string msg = json_trim(raw);
if (msg.empty()) return false;
std::string fn;
if (!json_get_string(msg, "function", fn)) return false; // 未知消息忽略
if (fn == "config")
{
handle_config(msg, client, resp);
return true;
}
// 其余命令需要指定雷达:{"radar":"radar1"};缺省用第一个雷达(兼容旧客户端)
std::string req_id;
json_get_string(msg, "radar", req_id);
RadarNode* node = radar_mgr.find(req_id);
if (!node)
{
std::string m = "unknown radar: " + req_id;
std::cerr << "[TCP] " << m << std::endl;
send_reply(client, resp, "{\"code\":400,\"message\":\"" + m + "\"}");
return true;
}
const std::string radar_id = node->id;
RadarScope scope(radar_id); // 本请求全程使用该雷达的配置
if (fn == "ConfigureSpeed" || fn == "configurespeed")
{
handle_configure_speed(msg, client, resp, radar_id, node);
return true;
}
if (fn != "count" && fn != "check") return false; // 未知功能忽略
std::string scan_cmd;
json_get_string(msg, "scan", scan_cmd);
if (scan_cmd == "stop")
{
handle_scan_stop(msg, client, resp, radar_id);
return true;
}
else if (scan_cmd == "start")
{
handle_scan_start(msg, client, resp, radar_id, node);
return true;
}
return false; // 无scan/未知scan值忽略
}
static void handle_client(SOCKET client)
{
WorkerGuard wg; // 计入活跃工作线程,退出时被等待
ConnGuard guard(client); // 任何路径退出都关闭 socket 并归还配额
set_sock_timeouts(client, TCP_RECV_TIMEOUT_MS, SEND_TIMEOUT_MS);
char buf[4096];
std::string rbuf;
while (true)
{
int n = recv(client, buf, sizeof(buf) - 1, 0);
if (n <= 0) break; // 连接关闭 / 超时 / 出错
buf[n] = '\0';
rbuf.append(buf, n);
// 按 '>' 分帧处理(一条消息以 '>' 结尾)
size_t pos;
while ((pos = rbuf.find('>')) != std::string::npos)
{
std::string raw = rbuf.substr(0, pos);
rbuf.erase(0, pos + 1);
try { dispatch(raw, client); }
catch (const std::exception& e)
{
std::cerr << "[TCP] 处理命令异常: " << e.what() << std::endl;
}
catch (...) { std::cerr << "[TCP] 处理命令发生未知异常" << std::endl; }
}
// 一直没出现 '>':异常客户端灌数据,防止 rbuf 无限膨胀
if (rbuf.size() > 65536) rbuf.clear();
}
// 客户端断开:若它是某自动扫描的 owner,扫描仍会到时自动停止;
// 未设置 scanTime 的扫描则持续到下一次 start 强制重启或 stop。
}
} // namespace
void run_tcp_server()
{
std::cout << "\n========== TCP服务端模式 ==========" << std::endl;
std::cout << "[TCP] 监听端口: " << cfg.server.port
<< " | 消息以 '>' 结尾" << std::endl;
std::cout << "[TCP] 已绑定雷达" << radar_mgr.nodes().size() << "台:";
for (const auto& id : cfg.ids) std::cout << " " << id;
std::cout << " (请求用 \"radar\" 指定,缺省用第一个)" << std::endl;
SOCKET listen_sock = socket(AF_INET, SOCK_STREAM, 0);
if (listen_sock == INVALID_SOCKET)
{
std::cerr << "[错误] 创建监听socket失败!" << std::endl;
return;
}
sockaddr_in addr;
addr.sin_family = AF_INET;
addr.sin_addr.s_addr = INADDR_ANY;
addr.sin_port = htons(cfg.server.port);
if (bind(listen_sock, (sockaddr*)&addr, sizeof(addr)) == SOCKET_ERROR)
{
std::cerr << "[错误] 绑定端口 " << cfg.server.port << " 失败!" << std::endl;
closesocket(listen_sock);
return;
}
if (listen(listen_sock, 5) == SOCKET_ERROR)
{
std::cerr << "[错误] 监听失败!" << std::endl;
closesocket(listen_sock);
return;
}
// 自动到时停止线程(scanTime 模式)
std::thread(auto_stop_loop).detach();
std::cout << "[TCP] 初始化完成,等待命令 "
"{\"function\":\"count\",\"scan\":\"start\",...} ..." << std::endl;
// 主循环必须吞掉一切异常:线程创建失败等异常若逃出 main 会直接终止进程
try
{
while (true)
{
SOCKET client = accept(listen_sock, nullptr, nullptr);
if (client == INVALID_SOCKET)
{
std::cout << "[TCP] accept失败,重试..." << std::endl;
std::this_thread::sleep_for(std::chrono::milliseconds(200));
continue;
}
set_sock_timeouts(client, TCP_RECV_TIMEOUT_MS, SEND_TIMEOUT_MS);
if (g_conn_count.load() >= MAX_CONNECTIONS)
{
std::cerr << "[TCP] 并发连接数已达上限 " << MAX_CONNECTIONS
<< ",拒绝新连接(可能有客户端连上后不发数据)" << std::endl;
closesocket(client);
continue;
}
++g_conn_count;
try { std::thread(handle_client, client).detach(); }
catch (...)
{
--g_conn_count;
std::cerr << "[TCP] 创建客户端处理线程失败,关闭该连接" << std::endl;
closesocket(client);
}
}
}
catch (const std::exception& e)
{
std::cerr << "[TCP] 服务端主循环异常退出: " << e.what() << std::endl;
}
catch (...) { std::cerr << "[TCP] 服务端主循环未知异常退出" << std::endl; }
closesocket(listen_sock);
stop_server_workers(); // 通知并等待 detach 的工作线程退出,再交回 main 收尾
}
// ==================== HTTP 服务端模式 ====================
// 请求: POST /<任意路径>, body = 与 TCP 协议相同的 JSON(不带结尾 '>')
// 响应: HTTP 200 + application/json, body = 与 TCP 协议相同的响应文本(不带 '>')
// 注: HTTP 请求是"无连接"的。scanTime>0 的扫描到点自动停止后无法主动推送
// 结果(结果打印到日志);客户端应在需要结果时再发一次 stop(会返回
// "scan not running" + count:0) 或通过文件目录查看。建议流程:
// start(不带scanTime) → 采集完成 → stop 拿结果。
namespace
{
// 读 HTTP 请求头(依赖 socket 上的 SO_RCVTIMEO,超时 recv 返回 <0)。
// 返回: 0 = 收到完整头(head=头, rest=已收到的 body 片段)
// 1 = 超时/对端半关且未收到 "\r\n\r\n"(head=已收到的全部数据, 用于TCP协议兜底)
// -1 = 出错或没有任何数据,应关闭连接
static int http_read_head(SOCKET c, std::string& head, std::string& rest)
{
char buf[8192];
head.clear();
rest.clear();
while (head.size() < 65536) // 头上限 64KB,防呆
{
int n = recv(c, buf, (int)sizeof(buf), 0);
if (n < 0) return head.empty() ? -1 : 1; // 超时/出错
if (n == 0) return head.empty() ? -1 : 1; // 对端关闭(可能已发部分数据)
head.append(buf, n);
size_t p = head.find("\r\n\r\n");
if (p != std::string::npos)
{
rest = head.substr(p + 4); // 头部之后可能已带部分 body
head = head.substr(0, p);
return 0;
}
}
return -1;
}
static long http_find_content_length(const std::string& head)
{
std::istringstream iss(head);
std::string line;
while (std::getline(iss, line))
{
if (!line.empty() && line.back() == '\r') line.pop_back();
size_t c = line.find(':');
if (c == std::string::npos) continue;
std::string key = line.substr(0, c);
for (auto& ch : key) ch = (char)tolower((unsigned char)ch);
if (key == "content-length")
return strtol(line.c_str() + c + 1, nullptr, 10);
}
return -1;
}
static void http_send_response(SOCKET c, int code, const char* reason,
const std::string& body)
{
std::ostringstream h;
h << "HTTP/1.1 " << code << " " << reason << "\r\n"
<< "Content-Type: application/json; charset=utf-8\r\n"
<< "Content-Length: " << body.size() << "\r\n"
<< "Connection: close\r\n\r\n";
std::string out = h.str() + body;
int off = 0;
while (off < (int)out.size())
{
int n = send(c, out.c_str() + off, (int)out.size() - off, 0);
if (n <= 0) break;
off += n;
}
}
// 每个连接处理一个请求后即关闭(Connection: close)
static void handle_http_client(SOCKET c)
{
WorkerGuard wg; // 计入活跃工作线程,退出时被等待
ConnGuard guard(c); // 任何路径退出都关闭 socket 并归还配额
set_sock_timeouts(c, HTTP_RECV_TIMEOUT_MS, SEND_TIMEOUT_MS);
std::string head, rest;
int hr = http_read_head(c, head, rest);
if (hr < 0) return;
// 兼容兜底:旧 TCP 客户端(直接发 "{...}" 无 HTTP 头)连到 HTTP 端口时,
// 永远等不到 "\r\n\r\n",原实现会让线程永久卡死且不打印任何日志。
// 这里改为按 TCP 协议解析并直接回包,避免线程泄漏。
if (hr != 0)
{
std::string raw = head;
std::cout << "[HTTP] 收到非HTTP报文(" << raw.size() << "B),按TCP协议处理" << std::endl;
size_t pos;
bool replied = false;
while ((pos = raw.find('>')) != std::string::npos)
{
std::string one = raw.substr(0, pos);
raw.erase(0, pos + 1);
try { if (dispatch(one, c, nullptr)) replied = true; }
catch (const std::exception& e)
{
std::cerr << "[HTTP] 处理命令异常: " << e.what() << std::endl;
}
catch (...) { std::cerr << "[HTTP] 处理命令发生未知异常" << std::endl; }
}
if (!replied)
send_text(c, "{\"code\":400,\"message\":\"unknown command, expect HTTP POST or TCP json ending with '>'\"}");
return;
}
std::string line0 = head.substr(0, head.find("\r\n"));
std::istringstream iss(line0);
std::string method, path;
iss >> method >> path;
if (method != "POST")
{
http_send_response(c, 405, "Method Not Allowed",
"{\"code\":405,\"message\":\"use POST, body same as TCP protocol json\"}");
return;
}
long cl = http_find_content_length(head);
if (cl < 0 || cl > 4 * 1024 * 1024) // body 上限 4MB
{
http_send_response(c, 411, "Length Required",
"{\"code\":411,\"message\":\"missing/invalid Content-Length\"}");
return;
}
std::string body = rest;
while ((long)body.size() < cl)
{
char buf[8192];
int n = recv(c, buf, (int)sizeof(buf), 0);
if (n <= 0) break;
body.append(buf, n);
}
if ((long)body.size() > cl) body.resize(cl);
std::string msg = json_trim(body); // 完整请求体,用于分发
std::string brief = msg; // 仅用于日志,避免污染真实报文
if (brief.size() > 120) brief = brief.substr(0, 120) + "...";
std::cout << "[HTTP] POST " << path << " (" << cl << "B): "
<< brief << std::endl;
std::string out;
bool handled = false;
try { handled = dispatch(msg, INVALID_SOCKET, &out); }
catch (const std::exception& e)
{
std::cerr << "[HTTP] 处理命令异常: " << e.what() << std::endl;
out = "{\"code\":500,\"message\":\"internal error\"}";
handled = true;
}
catch (...)
{
std::cerr << "[HTTP] 处理命令发生未知异常" << std::endl;
out = "{\"code\":500,\"message\":\"internal error\"}";
handled = true;
}
if (handled)
http_send_response(c, 200, "OK", out);
else
http_send_response(c, 400, "Bad Request",
"{\"code\":400,\"message\":\"unknown command, see TCP protocol\"}");
}
static void run_http_server_internal()
{
SOCKET listen_sock = socket(AF_INET, SOCK_STREAM, 0);
if (listen_sock == INVALID_SOCKET)
{
std::cerr << "[错误] 创建HTTP监听socket失败!" << std::endl;
return;
}
sockaddr_in addr;
addr.sin_family = AF_INET;
addr.sin_addr.s_addr = INADDR_ANY;
addr.sin_port = htons(cfg.server.port);
if (bind(listen_sock, (sockaddr*)&addr, sizeof(addr)) == SOCKET_ERROR)
{
std::cerr << "[错误] 绑定HTTP端口 " << cfg.server.port << " 失败!" << std::endl;
closesocket(listen_sock);
return;
}
if (listen(listen_sock, 5) == SOCKET_ERROR)
{
std::cerr << "[错误] HTTP监听失败!" << std::endl;
closesocket(listen_sock);
return;
}
// 自动到时停止线程(scanTime 模式; HTTP 下结果打印日志不推送)
std::thread(auto_stop_loop).detach();
std::cout << "[HTTP] 初始化完成,等待 POST 命令 "
"{\"function\":\"count\",\"scan\":\"start\",...} ..." << std::endl;
// 主循环必须吞掉一切异常:线程创建失败等异常逃出 main 会直接终止进程
try
{
while (true)
{
SOCKET client = accept(listen_sock, nullptr, nullptr);
if (client == INVALID_SOCKET)
{
std::cout << "[HTTP] accept失败,重试..." << std::endl;
std::this_thread::sleep_for(std::chrono::milliseconds(200));
continue;
}
set_sock_timeouts(client, HTTP_RECV_TIMEOUT_MS, SEND_TIMEOUT_MS);
if (g_conn_count.load() >= MAX_CONNECTIONS)
{
std::cerr << "[HTTP] 并发连接数已达上限 " << MAX_CONNECTIONS
<< ",拒绝新连接(可能有客户端连上后不发数据)" << std::endl;
closesocket(client);
continue;
}
++g_conn_count;
try { std::thread(handle_http_client, client).detach(); }
catch (...)
{
--g_conn_count;
std::cerr << "[HTTP] 创建客户端处理线程失败,关闭该连接" << std::endl;
closesocket(client);
}
}
}
catch (const std::exception& e)
{
std::cerr << "[HTTP] 服务端主循环异常退出: " << e.what() << std::endl;
}
catch (...) { std::cerr << "[HTTP] 服务端主循环未知异常退出" << std::endl; }
closesocket(listen_sock);
}
}
void run_http_server()
{
std::cout << "\n========== HTTP服务端模式 ==========" << std::endl;
std::cout << "[HTTP] 监听端口: " << cfg.server.port
<< " | POST json body(与TCP协议相同, 不带结尾'>')" << std::endl;
std::cout << "[HTTP] 已绑定雷达" << radar_mgr.nodes().size() << "台:";
for (const auto& id : cfg.ids) std::cout << " " << id;
std::cout << " (请求用 \"radar\" 指定,缺省用第一个)" << std::endl;
run_http_server_internal();
stop_server_workers(); // 通知并等待 detach 的工作线程退出,再交回 main 收尾
}