--[[ @module evt_queue @summary event_report 待发队列 + ACK 状态机(Lua 版 iot_evt_*,对齐 DBN 语义) @version 0.1 @date 2026.08.31 @usage 复刻 vd960DBN iot_mqtt_srv.c 的 event_report 机制(协议 V1.04 / §5.3): - 16 深环形队列,溢出丢最旧 - 平台 ACK(code=0, 回显 msg_id) → 出队 - 5s 超时重发,同 msg_id/原始 ts,最多 3 次;耗尽丢弃 - 跨 MQTT 重连保持同 msg_id 立即重发(平台按 (sn,msg_id) 去重) - 多事件合并一条 publish - 独立 msg_id(不复用 MQTT packet id) 用法: local eq = require "evt_queue" eq.init() -- 上电初始化 -- 沿检测出事件: local need_pub = eq.enqueue({type="car_enter", ch=1, value=350}) if need_pub then local ok = publish_now() end -- 主循环/定时器驱动: local action = eq.poll(now_ms, now_ts) -- "resend" | nil -- MQTT 重连后: local action = eq.on_reconnect(now_ms) -- 收到平台 ACK: eq.handle_ack(msg_id, code) ]] local evt_queue = {} -- 事件类型名(与协议 §5.3 一致) local EVT_NAMES = { "car_enter", "car_leave", "loop_cut", "loop_restore" } local DEPTH = 16 local ACK_TIMEOUT_MS = 5000 local MAX_RETRY = 3 -- 环形队列 local queue = {} -- table,逻辑索引 0..DEPTH-1 local head = 0 -- 队首 local count = 0 -- 有效条数 -- 待确认包状态 local msg_id = 0 -- event_report 独立 msg_id(递增) local pend_id = 0 -- 待确认包 msg_id,0=无 local pend_n = 0 -- 待确认包含事件数 local pend_ts = 0 -- 原始 ts(重发不刷新) local retry = 0 -- 已重发次数 local sent_ms = 0 -- 上次发送时刻 function evt_queue.init() queue = {} head = 0 count = 0 msg_id = 0 pend_id = 0 pend_n = 0 pend_ts = 0 retry = 0 sent_ms = 0 end -- 入队;返回 true = 当前无未决包,可立即发布 function evt_queue.enqueue(evt) if count >= DEPTH then head = (head + 1) % DEPTH -- 溢出丢最旧 count = count - 1 end queue[(head + count) % DEPTH] = evt count = count + 1 return (pend_id == 0) end -- 组 event_report JSON(合并队列中全部事件) -- @param ts number: 时间戳(首发用 now_ts;重发用 pend_ts 由调用方传入) -- @return payload string 或 nil(队列空) function evt_queue.build_report(ts) if count == 0 then return nil end msg_id = msg_id + 1 local parts = {} for i = 0, count - 1 do local e = queue[(head + i) % DEPTH] local tname = EVT_NAMES[e.type] or "car_enter" parts[#parts + 1] = string.format('{"type":"%s","ch":%d,"value":%d}', tname, e.ch, e.value or 0) end local payload = string.format('{"msg_id":%d,"cmd":"event_report","ts":%d,' .. '"data":{"events":[%s]}}', msg_id, ts, table.concat(parts, ",")) return payload end -- 发送后登记待确认(首发/重发后调用) function evt_queue.mark_sent(id, n, ts, now_ms) pend_id = id pend_n = n pend_ts = ts retry = 0 sent_ms = now_ms end -- 平台 ACK 处理(协议 §5.3: 平台回显 msg_id + code) function evt_queue.handle_ack(ack_id, code) if pend_id == 0 or ack_id ~= pend_id then return end if code ~= 0 then return -- 非 0 保持挂起 end head = (head + pend_n) % DEPTH count = count - pend_n if count < 0 then count = 0 end pend_id = 0 pend_n = 0 retry = 0 end -- 主循环/定时器驱动: 超时重发判定 -- @return "resend"(需要重发) | nil function evt_queue.poll(now_ms, now_ts) if pend_id == 0 then return nil end if now_ms - sent_ms < ACK_TIMEOUT_MS then return nil end if retry >= MAX_RETRY then -- 重试耗尽: 丢弃本包(可扩展 offlog 记录) head = (head + pend_n) % DEPTH count = count - pend_n if count < 0 then count = 0 end pend_id = 0 pend_n = 0 retry = 0 return nil end retry = retry + 1 sent_ms = now_ms return "resend" end -- MQTT 重连后: 未决包立即重发(同 msg_id/原始 ts) function evt_queue.on_reconnect(now_ms) if pend_id ~= 0 then retry = 0 sent_ms = now_ms return "resend" end return nil end -- 取待确认信息(重发时用) function evt_queue.pending_info() return pend_id, pend_n, pend_ts end -- 最近一次 build_report 生成的 msg_id(发布后登记用) function evt_queue.last_msg_id() return msg_id end -- 队列事件数 function evt_queue.count() return count end return evt_queue