mobile_comm/luatos/air780epm/module/Air780EPM/demo/vd960Air/evt_queue.lua

--[[
@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