init: Air780EPM 官方 LuatOS 项目代码基线

- 来源: 合宙 LuatOS 官方仓库 air780epm 模块完整代码
- 路径: luatos/air780epm/module/Air780EPM/demo 含官方 demo(含 mqtt/mqtts/socket/uart 等)
- 后续: 基于 demo 开发 UART<->MQTT 数据上报功能
This commit is contained in:
wangfq
2026-08-31 08:53:49 +08:00
commit 173dd6874f
813 changed files with 125484 additions and 0 deletions
@@ -0,0 +1,98 @@
--[[
@module main
@summary LuatOS用户应用脚本文件入口,总体调度应用逻辑
@version 1.0
@date 2025.07.28
@author 马梦阳
@usage
本demo演示的核心功能为:
1、创建四路mqtt连接,详情如下
- 创建一个mqtt client,连接mqtt server
- 创建一个mqtt ssl client,连接mqtt ssl server,不做证书校验;
- 创建一个mqtt ssl client,连接mqtt ssl serverclient仅单向校验server的证书,server不校验client的证书和密钥文件;
- 创建一个mqtt ssl client,连接mqtt ssl serverclient校验server的证书,server校验client的证书和密钥文件;
2、每一路mqtt连接出现异常后,自动重连;
3、每一路mqtt连接,client按照以下几种逻辑发送数据给server
- 串口应用功能模块uart_app.lua,通过uart1接收到串口数据,将串口数据增加send from uart: 前缀后,使用mobile.imei().."/uart/up"主题,发送给server
- 定时器应用功能模块timer_app.lua,定时产生数据,将数据增加send from timer:前缀后,使用mobile.imei().."/timer/up"主题,发送给server
4、每一路mqtt连接,client收到server数据后,将数据增加recv from mqtt/mqtt ssl/mqtt ssl ca/mqtt ssl mutual ca(四选一)server: 前缀后,通过uart1发送出去;
5、启动一个网络业务逻辑看门狗task,用来监控网络环境,如果连续长时间工作不正常,重启整个软件系统;
6、netdrv_device:配置连接外网使用的网卡,目前支持以下三种选择(三选一)
(1) netdrv_4g4G网卡
(2) netdrv_eth_spi:通过SPI外挂CH390H芯片的以太网卡
(3) netdrv_multiple:支持以上两种网卡,可以配置两种网卡的优先级
更多说明参考本目录下的readme.md文件
]]
--[[
必须定义PROJECT和VERSION变量,Luatools工具会用到这两个变量,远程升级功能也会用到这两个变量
PROJECT:项目名,ascii string类型
可以随便定义,只要不使用,就行
VERSION:项目版本号,ascii string类型
如果使用合宙iot.openluat.com进行远程升级,必须按照"XXX.YYY.ZZZ"三段格式定义:
X、Y、Z各表示1位数字,三个X表示的数字可以相同,也可以不同,同理三个Y和三个Z表示的数字也是可以相同,可以不同
因为历史原因,YYY这三位数字必须存在,但是没有任何用处,可以一直写为999
如果不使用合宙iot.openluat.com进行远程升级,根据自己项目的需求,自定义格式即可
]]
PROJECT = "MQTT"
VERSION = "001.999.000"
-- 在日志中打印项目名和项目版本号
log.info("main", PROJECT, VERSION)
-- 如果内核固件支持errDump功能,此处进行配置,【强烈建议打开此处的注释】
-- 因为此功能模块可以记录并且上传脚本在运行过程中出现的语法错误或者其他自定义的错误信息,可以初步分析一些设备运行异常的问题
-- 以下代码是最基本的用法,更复杂的用法可以详细阅读API说明文档
-- 启动errDump日志存储并且上传功能,600秒上传一次
-- if errDump then
-- errDump.config(true, 600)
-- end
-- 使用LuatOS开发的任何一个项目,都强烈建议使用远程升级FOTA功能
-- 可以使用合宙的iot.openluat.com平台进行远程升级
-- 也可以使用客户自己搭建的平台进行远程升级
-- 远程升级的详细用法,可以参考fota的demo进行使用
-- 启动一个循环定时器
-- 每隔3秒钟打印一次总内存,实时的已使用内存,历史最高的已使用内存情况
-- 方便分析内存使用是否有异常
-- sys.timerLoopStart(function()
-- log.info("mem.lua", rtos.meminfo())
-- log.info("mem.sys", rtos.meminfo("sys"))
-- end, 3000)
-- 加载网络环境检测看门狗功能模块
require "network_watchdog"
-- 加载网络驱动设备功能模块
require "netdrv_device"
-- 加载串口应用功能模块
require "uart_app"
-- 加载定时器应用功能模块
require "timer_app"
-- 加载mqtt client 主应用功能模块
require "mqtt_main"
-- 加载mqtt ssl client 主应用功能模块(mqtt ssl 无证书校验)
-- require "mqtts_main"
-- 加载mqtt ssl ca client 主应用功能模块(mqtt ssl 单向证书校验)
-- require "mqtts_ca_main"
-- 加载mqtt ssl mutual ca client 主应用功能模块(mqtt ssl 双向证书校验)
-- require "mqtts_m_ca_main"
-- 用户代码已结束---------------------------------------------
-- 结尾总是这一句
sys.run()
-- sys.run()之后不要加任何语句!!!!!因为添加的任何语句都不会被执行
@@ -0,0 +1,265 @@
--[[
@module mqtt_main
@summary mqtt client 主应用功能模块
@version 1.0
@date 2025.07.28
@author 马梦阳
@usage
本文件为mqtt client 主应用功能模块,核心业务逻辑为:
1、创建一个mqtt client,连接server
2、处理连接/订阅/取消订阅/异常逻辑,出现异常后执行重连动作;
3、调用mqtt_receiver的外部接口mqtt_receiver.proc,对接收到的publish数据进行处理;
4、调用sys.sendMsg接口,发送"CONNECT OK"、"PUBLISH OK"和"DISCONNECTED"三种类型的"MQTT_EVENT"消息到mqtt_sender的task,控制publish数据发送逻辑;
5、收到MQTT心跳应答后,执行sys.publish("FEED_NETWORK_WATCHDOG") 对网络环境检测看门狗功能模块进行喂狗;
本文件没有对外接口,直接在main.lua中require "mqtt_main"就可以加载运行;
]]
-- 加载mqtt client数据接收功能模块
local mqtt_receiver = require "mqtt_receiver"
-- 加载mqtt client数据发送功能模块
local mqtt_sender = require "mqtt_sender"
-- mqtt服务器地址和端口
-- 这里使用的地址和端口,会不定期重启或维护,仅能用作测试用途,不可商用,说不定哪一天就关闭了
-- 用户开发项目时,替换为自己的商用服务器地址和端口
local SERVER_ADDR = "lbsmqtt.airm2m.com"
local SERVER_PORT = 1884
-- mqtt_main的任务名
local TASK_NAME = mqtt_sender.TASK_NAME_PREFIX.."main"
-- mqtt主题的前缀:IMEI号
local TOPIC_PREFIX = mobile.imei()
-- mqtt client的事件回调函数
local function mqtt_client_event_cbfunc(mqtt_client, event, data, payload, metas)
log.info("mqtt_client_event_cbfunc", mqtt_client, event, data, payload, json.encode(metas))
-- mqtt连接成功
if event == "conack" then
sys.sendMsg(TASK_NAME, "MQTT_EVENT", "CONNECT", true)
-- 订阅单主题
-- 第二个参数表示qos,取值范围为0,1,2,如果不设置,默认为0
if not mqtt_client:subscribe(TOPIC_PREFIX .. "/down") then
sys.sendMsg(TASK_NAME, "MQTT_EVENT", "SUBSCRIBE", false, -1)
end
-- 订阅多主题,如果有需要,打开注释
-- 表中的每一个订阅主题的格式为[topic]=qos
-- if not mqtt_client:subscribe(
-- {
-- [(TOPIC_PREFIX .. "/data"]=0,
-- [(TOPIC_PREFIX .. "/cmd"]=1
-- }
-- ) then
-- sys.sendMsg(TASK_NAME, "MQTT_EVENT", "SUBSCRIBE", false, -1)
-- end
-- 订阅结果
-- data:订阅应答结果,true为成功,false为失败
-- payloadnumber类型;成功时表示qos,取值范围为0,1,2;失败时表示失败码,一般是0x80
elseif event == "suback" then
-- 发送消息通知 mqtt main task
sys.sendMsg(TASK_NAME, "MQTT_EVENT", "SUBSCRIBE", data, payload)
-- 取消订阅成功
elseif event == "unsuback" then
-- 发送消息通知 mqtt main task
sys.sendMsg(TASK_NAME, "MQTT_EVENT", "UNSUBSCRIBE", true)
-- 接收到服务器下发的publish数据
-- datastring类型,表示topic
-- payloadstring类型,表示payload
-- metastable类型,数据内容如下
-- {
-- qos: number类型,取值范围0,1,2
-- retainnumber类型,取值范围0,1
-- dupnumber类型,取值范围0,1
-- message_id: number类型
-- }
elseif event == "recv" then
-- 对接收到的publish数据处理
mqtt_receiver.proc(data, payload, metas)
-- 发送成功publish数据
-- datanumber类型,表示message id
elseif event == "sent" then
-- 发送消息通知 mqtt sender task
sys.sendMsg(mqtt_sender.TASK_NAME, "MQTT_EVENT", "PUBLISH_OK", data)
-- 服务器断开mqtt连接
elseif event == "disconnect" then
-- 发送消息通知 mqtt main task
sys.sendMsg(TASK_NAME, "MQTT_EVENT", "DISCONNECTED", false)
-- 收到服务器的心跳应答
elseif event == "pong" then
-- 接收到数据,通知网络环境检测看门狗功能模块进行喂狗
sys.publish("FEED_NETWORK_WATCHDOG")
-- 严重异常,本地会主动断开连接
-- data:string类型,表示具体的异常,有以下几种:
-- "connect"tcp连接失败
-- "tx":数据发送失败
-- "conack"mqtt connect后,服务器应答CONNACK鉴权失败,失败码为payloadnumber类型)
-- "other":其他异常
elseif event == "error" then
if data == "connect" or data == "conack" then
-- 发送消息通知 mqtt main task,连接失败
sys.sendMsg(TASK_NAME, "MQTT_EVENT", "CONNECT", false)
elseif data == "other" or data == "tx" then
-- 发送消息通知 mqtt main task,出现异常
sys.sendMsg(TASK_NAME, "MQTT_EVENT", "ERROR")
end
end
end
-- mqtt main task 的任务处理函数
local function mqtt_client_main_task_func()
local mqtt_client
local result, msg
while true do
-- 如果当前时间点设置的默认网卡还没有连接成功,一直在这里循环等待
while not socket.adapter(socket.dft()) do
log.warn("mqtt_client_main_task_func", "wait IP_READY", socket.dft())
-- 在此处阻塞等待默认网卡连接成功的消息"IP_READY"
-- 或者等待1秒超时退出阻塞等待状态;
-- 注意:此处的1000毫秒超时不要修改的更长;
-- 因为当使用exnetif.set_priority_order配置多个网卡连接外网的优先级时,会隐式的修改默认使用的网卡
-- 当exnetif.set_priority_order的调用时序和此处的socket.adapter(socket.dft())判断时序有可能不匹配
-- 此处的1秒,能够保证,即使时序不匹配,也能1秒钟退出阻塞状态,再去判断socket.adapter(socket.dft())
sys.waitUntil("IP_READY", 1000)
end
-- 检测到了IP_READY消息
log.info("mqtt_client_main_task_func", "recv IP_READY", socket.dft())
-- 清空此task绑定的消息队列中的未处理的消息
sys.cleanMsg(TASK_NAME)
-- 创建mqtt client对象
mqtt_client = mqtt.create(nil, SERVER_ADDR, SERVER_PORT)
-- 如果创建mqtt client对象失败
if not mqtt_client then
log.error("mqtt_client_main_task_func", "mqtt.create error")
goto EXCEPTION_PROC
end
-- 配置mqtt client对象的client idusernamepassword和clean session标志
result = mqtt_client:auth(TASK_NAME..mobile.imei(), "", "", true)
-- 如果配置失败
if not result then
log.error("mqtt_client_main_task_func", "mqtt_client:auth error")
goto EXCEPTION_PROC
end
-- 注册mqtt client对象的事件回调函数
mqtt_client:on(mqtt_client_event_cbfunc)
-- 设置mqtt keepalive时间为120秒
-- 如果没有设置,内核固件中默认为180秒
-- 有需要的话,可以打开注释
-- mqtt_client:keepalive(120)
-- 设置遗嘱消息,有需要的话,可以打开注释
-- mqtt_client:will(TOPIC_PREFIX .. "/status", "offline")
-- 连接server
result = mqtt_client:connect()
-- 如果连接server失败
if not result then
log.error("mqtt_client_main_task_func", "mqtt_client:connect error")
goto EXCEPTION_PROC
end
-- 连接、断开连接、订阅、取消订阅、异常等各种事件的处理调度逻辑
while true do
-- 等待"MQTT_EVENT"消息
msg = sys.waitMsg(TASK_NAME, "MQTT_EVENT")
log.info("mqtt_client_main_task_func waitMsg", msg[2], msg[3], msg[4])
-- connect连接结果
-- msg[3]表示连接结果,true为连接成功,false为连接失败
if msg[2] == "CONNECT" then
-- mqtt连接成功
if msg[3] then
log.info("mqtt_client_main_task_func", "connect success")
-- 通知mqtt sender数据发送应用模块的task,MQTT连接成功
sys.sendMsg(mqtt_sender.TASK_NAME, "MQTT_EVENT", "CONNECT_OK", mqtt_client)
-- mqtt连接失败
else
log.info("mqtt_client_main_task_func", "connect error")
-- 退出循环,发起重连
break
end
-- subscribe订阅结果
-- msg[3]表示订阅结果,true为订阅成功,false为订阅失败
elseif msg[2] == "SUBSCRIBE" then
-- 订阅成功
if msg[3] then
log.info("mqtt_client_main_task_func", "subscribe success", "qos: "..(msg[4] or "nil"))
-- 订阅失败
else
log.error("mqtt_client_main_task_func", "subscribe error", "code", msg[4])
-- 主动断开mqtt client连接
mqtt_client:disconnect()
-- 发送disconnect之后,此处延时1秒,给数据发送预留一点儿时间,发送到服务器;
-- 即使1秒的时间不足以发送给服务器也没关系;对服务器来说,mqtt客户端只是没有优雅的断开,不影响什么实质功能;
sys.wait(1000)
break
end
-- unsubscribe取消订阅成功
elseif msg[2] == "UNSUBSCRIBE" then
log.info("mqtt_client_main_task_func", "unsubscribe success")
-- 需要主动关闭mqtt连接
-- 用户需要主动关闭mqtt连接时,可以调用sys.sendMsg(TASK_NAME, "MQTT_EVENT", "CLOSE")
elseif msg[2] == "CLOSE" then
-- 主动断开mqtt client连接
mqtt_client:disconnect()
-- 发送disconnect之后,此处延时1秒,给数据发送预留一点儿时间,发送到服务器;
-- 即使1秒的时间不足以发送给服务器也没关系;对服务器来说,mqtt客户端只是没有优雅的断开,不影响什么实质功能;
sys.wait(1000)
break
-- 被动关闭了mqtt连接
-- 被网络或者服务器断开了连接
elseif msg[2] == "DISCONNECTED" then
break
-- 出现了其他异常
elseif msg[2] == "ERROR" then
break
end
end
-- 出现异常
::EXCEPTION_PROC::
-- 清空此task绑定的消息队列中的未处理的消息
sys.cleanMsg(TASK_NAME)
-- 通知mqtt sender数据发送应用模块的task,MQTT连接已经断开
sys.sendMsg(mqtt_sender.TASK_NAME, "MQTT_EVENT", "DISCONNECTED")
-- 如果存在mqtt client对象
if mqtt_client then
-- 关闭mqtt client,并且释放mqtt client对象
mqtt_client:close()
mqtt_client = nil
end
-- 5秒后跳转到循环体开始位置,自动发起重连
sys.wait(5000)
end
end
--创建并且启动一个task
--运行这个task的处理函数mqtt_client_main_task_func
sys.taskInitEx(mqtt_client_main_task_func, TASK_NAME)
@@ -0,0 +1,63 @@
--[[
@module mqtt_receiver
@summary mqtt client数据接收处理应用功能模块
@version 1.0
@date 2025.07.29
@author 马梦阳
@usage
本文件为mqtt client 数据接收应用功能模块,核心业务逻辑为:
处理接收到的publish数据,同时将数据发送给其他应用功能模块做进一步处理;
本文件的对外接口有2个:
1、mqtt_receiver.proc(topic, payload, metas)publish数据处理入口,在mqtt_main.lua中调用;
2、sys.publish("RECV_DATA_FROM_SERVER", "recv from mqtt server: ", topic, payload)
将接收到的publish中的topic和payload数据通过消息"RECV_DATA_FROM_SERVER"发布出去;
需要处理数据的应用功能模块订阅处理此消息即可,本demo项目中uart_app.lua中订阅处理了本消息;
]]
local mqtt_receiver = {}
--[[
处理接收到的publish数据
@api mqtt_receiver.proc(topic, payload, metas)
@param1 topic string
表示publish主题
@param2 payload string
表示publish数据负载
@param2 payload string
表示publish数据负载
@param3 metas table
表示publish报文的一些参数;格式如下:
{
qos: number类型,取值范围0,1,2
retainnumber类型,取值范围0,1
dupnumber类型,取值范围0,1
message_id: number类型
}
@return1 result nil
@usage
mqtt_receiver.proc(topic, payload, metas)
]]
function mqtt_receiver.proc(topic, payload, metas)
log.info("mqtt_receiver.proc", topic, payload:len(), json.encode(metas))
-- 接收到数据,通知网络环境检测看门狗功能模块进行喂狗
sys.publish("FEED_NETWORK_WATCHDOG")
-- 将topic和payload通过"RECV_DATA_FROM_SERVER"消息publish出去,给其他应用模块处理
sys.publish("RECV_DATA_FROM_SERVER", "recv from mqtt server: ", topic, payload)
-- 也可以直接在此处编写代码,处理topic和payload
end
return mqtt_receiver
@@ -0,0 +1,155 @@
--[[
@module mqtt_sender
@summary mqtt client数据发送应用功能模块
@version 1.0
@date 2025.07.29
@author 马梦阳
@usage
本文件为mqtt client 数据发送应用功能模块,核心业务逻辑为:
1、sys.subscribe("SEND_DATA_REQ", send_data_req_proc_func)订阅"SEND_DATA_REQ"消息,将其他应用模块需要发送的数据存储到队列send_queue中;
2、mqtt sender task接收"CONNECT OK"、"PUBLISH_REQ"、"PUBLISH OK"三种类型的"MQTT_EVENT"消息,遍历队列send_queue,逐条发送数据到server
3、mqtt sender task接收"DISCONNECTED"类型的"MQTT_EVENT"消息,丢弃掉队列send_queue中未发送的数据;
4、任何一条数据无论发送成功还是失败,只要这条数据有回调函数,都会通过回调函数通知数据发送方;
本文件的对外接口有1个:
1、sys.subscribe("SEND_DATA_REQ", send_data_req_proc_func):订阅"SEND_DATA_REQ"消息;
其他应用模块如果需要发送数据,直接sys.publish这个消息即可,将需要发送的topic,payload和qos以及回调函数和回调参数一起publish出去;
本demo项目中uart_app.lua和timer_app.lua中publish了这个消息;
]]
local mqtt_sender = {}
--[[
数据发送队列,数据结构为:
{
[1] = {topic="topic1", payload="payload1", qos=0, cb={func=callback_function1, para=callback_para1}},
[2] = {topic="topic2", payload="payload2", qos=1, cb={func=callback_function2, para=callback_para2}},
[3] = {topic="topic3", payload="payload3", qos=2, cb={func=callback_function3, para=callback_para3}},
}
topic的内容为publish的主题,string类型,必须存在;
payload的内容为publish的负载数据,string类型,必须存在;
qos的内容为publish的质量等级,number类型,取值范围0,1,2,可选,如果用户没有指定,默认为0;
cb.func的内容为数据发送结果的用户回调函数,可以不存在;
cb.para的内容为数据发送结果的用户回调函数的回调参数,可以不存在;
]]
local send_queue = {}
-- mqtt client的任务名前缀
mqtt_sender.TASK_NAME_PREFIX = "mqtt_"
-- mqtt_client_sender的任务名
mqtt_sender.TASK_NAME = mqtt_sender.TASK_NAME_PREFIX.."sender"
-- "SEND_DATA_REQ"消息的处理函数
local function send_data_req_proc_func(tag, topic, payload, qos, cb)
-- 将原始数据增加前缀,然后插入到发送队列send_queue中
table.insert(send_queue, {topic=topic, payload="send from "..tag..": "..payload, qos=qos or 0, cb=cb})
-- 发送消息通知 mqtt sender task,有新数据等待发送
sys.sendMsg(mqtt_sender.TASK_NAME, "MQTT_EVENT", "PUBLISH_REQ")
end
-- 按照顺序发送send_queue中的数据
-- 如果调用publish接口成功,则返回当前正在发送的数据项
-- 如果调用publish接口失败,通知回调函数发送失败后,继续发送下一条数据
local function publish_item(mqtt_client)
local item
-- 如果发送队列中有数据等待发送
while #send_queue>0 do
-- 取出来第一条数据赋值给item
-- 同时从队列send_queue中删除这一条数据
item = table.remove(send_queue, 1)
-- publish数据
-- result表示调用publish接口的同步结果,返回值有以下几种:
-- 如果失败,返回nil
-- 如果成功,number类型,qos为0时直接返回0qos为1或者2时返回publish报文的message id
result = mqtt_client:publish(item.topic, item.payload, item.qos)
-- publish接口调用成功
if result then
return item
-- publish接口调用失败
else
-- 如果当前发送的数据有用户回调函数,则执行用户回调函数
if item.cb and item.cb.func then
item.cb.func(false, item.cb.para)
end
end
end
end
local function publish_item_cbfunc(item, result)
if item then
-- 如果当前发送的数据有用户回调函数,则执行用户回调函数
if item.cb and item.cb.func then
item.cb.func(result, item.cb.para)
end
end
end
-- mqtt client sender的任务处理函数
local function mqtt_client_sender_task_func()
local mqtt_client
local send_item
local result, msg
while true do
-- 等待"MQTT_EVENT"消息
msg = sys.waitMsg(mqtt_sender.TASK_NAME, "MQTT_EVENT")
-- mqtt连接成功
-- msg[3]表示mqtt client对象
if msg[2] == "CONNECT_OK" then
mqtt_client = msg[3]
-- 发送send_queue中的数据
send_item = publish_item(mqtt_client)
-- mqtt publish数据请求
elseif msg[2] == "PUBLISH_REQ" then
-- 如果mqtt client对象存在,并且没有正在等待发送结果的发送数据项
if mqtt_client and not send_item then
-- 发送send_queue中的数据
send_item = publish_item(mqtt_client)
end
-- mqtt publish数据成功
elseif msg[2] == "PUBLISH_OK" then
-- publish成功,执行回调函数通知发送方
publish_item_cbfunc(send_item, true)
-- publish成功,通知网络环境检测看门狗功能模块进行喂狗
sys.publish("FEED_NETWORK_WATCHDOG")
-- 发送send_queue中的数据
send_item = publish_item(mqtt_client)
-- mqtt断开连接
elseif msg[2] == "DISCONNECTED" then
-- 清空mqtt client对象
mqtt_client = nil
-- 如果存在正在等待发送结果的发送项,执行回调函数通知发送方失败
publish_item_cbfunc(send_item, false)
-- 如果发送队列中有数据等待发送
while #send_queue>0 do
-- 取出来第一条数据赋值给send_item
-- 同时从队列send_queue中删除这一条数据
send_item = table.remove(send_queue,1)
-- 执行回调函数通知发送方失败
publish_item_cbfunc(send_item, false)
end
-- 当前没有正在等待发送结果的发送项
send_item = nil
end
end
end
-- 订阅"SEND_DATA_REQ"消息;
-- 其他应用模块如果需要发送数据,直接sys.publish这个消息即可,将需要发送的数据以及回调函数和回调参数一起publish出去;
-- 本demo项目中uart_app.lua和timer_app.lua中publish了这个消息;
sys.subscribe("SEND_DATA_REQ", send_data_req_proc_func)
--创建并且启动一个task
--运行这个task的处理函数mqtt_client_sender_task_func
sys.taskInitEx(mqtt_client_sender_task_func, mqtt_sender.TASK_NAME)
return mqtt_sender
@@ -0,0 +1,269 @@
--[[
@module mqtts_main
@summary mqtts client 主应用功能模块
@version 1.0
@date 2025.07.28
@author 马梦阳
@usage
本文件为mqtts client 主应用功能模块,核心业务逻辑为:
1、创建一个mqtts client,连接server
2、处理连接/订阅/取消订阅/异常逻辑,出现异常后执行重连动作;
3、调用mqtts_receiver的外部接口mqtts_receiver.proc,对接收到的publish数据进行处理;
4、调用sys.sendMsg接口,发送"CONNECT OK"、"PUBLISH OK"和"DISCONNECTED"三种类型的"MQTT_EVENT"消息到mqtts_sender的task,控制publish数据发送逻辑;
5、收到MQTT心跳应答后,执行sys.publish("FEED_NETWORK_WATCHDOG") 对网络环境检测看门狗功能模块进行喂狗;
本文件没有对外接口,直接在main.lua中require "mqtts_main"就可以加载运行;
]]
-- 加载mqtts client数据接收功能模块
local mqtts_receiver = require "mqtts_receiver"
-- 加载mqtts client数据发送功能模块
local mqtts_sender = require "mqtts_sender"
-- mqtts服务器地址和端口
-- 这里使用的地址和端口,会不定期重启或维护,仅能用作测试用途,不可商用,说不定哪一天就关闭了
-- 用户开发项目时,替换为自己的商用服务器地址和端口
local SERVER_ADDR = "airtest.openluat.com"
local SERVER_PORT = 8888
-- mqtt连接时的用户名和密码
-- 必须填写,否则会连接失败
local USERNAME = "mqtt_hz_test_2"
local PASSWORD = "3bEKUb"
-- mqtts_main的任务名
local TASK_NAME = mqtts_sender.TASK_NAME_PREFIX.."main"
-- mqtt主题的前缀:IMEI号
local TOPIC_PREFIX = mobile.imei()
-- mqtts client的事件回调函数
local function mqtts_client_event_cbfunc(mqtt_client, event, data, payload, metas)
log.info("mqtts_client_event_cbfunc", mqtt_client, event, data, payload, json.encode(metas))
-- mqtt连接成功
if event == "conack" then
sys.sendMsg(TASK_NAME, "MQTT_EVENT", "CONNECT", true)
-- 订阅单主题
-- 第二个参数表示qos,取值范围为0,1,2,如果不设置,默认为0
if not mqtt_client:subscribe(TOPIC_PREFIX .. "/down") then
sys.sendMsg(TASK_NAME, "MQTT_EVENT", "SUBSCRIBE", false, -1)
end
-- 订阅多主题,如果有需要,打开注释
-- 表中的每一个订阅主题的格式为[topic]=qos
-- if not mqtt_client:subscribe(
-- {
-- [(TOPIC_PREFIX .. "/data"]=0,
-- [(TOPIC_PREFIX .. "/cmd"]=1
-- }
-- ) then
-- sys.sendMsg(TASK_NAME, "MQTT_EVENT", "SUBSCRIBE", false, -1)
-- end
-- 订阅结果
-- data:订阅应答结果,true为成功,false为失败
-- payloadnumber类型;成功时表示qos,取值范围为0,1,2;失败时表示失败码,一般是0x80
elseif event == "suback" then
-- 发送消息通知 mqtts main task
sys.sendMsg(TASK_NAME, "MQTT_EVENT", "SUBSCRIBE", data, payload)
-- 取消订阅成功
elseif event == "unsuback" then
-- 发送消息通知 mqtts main task
sys.sendMsg(TASK_NAME, "MQTT_EVENT", "UNSUBSCRIBE", true)
-- 接收到服务器下发的publish数据
-- datastring类型,表示topic
-- payloadstring类型,表示payload
-- metastable类型,数据内容如下
-- {
-- qos: number类型,取值范围0,1,2
-- retainnumber类型,取值范围0,1
-- dupnumber类型,取值范围0,1
-- message_id: number类型
-- }
elseif event == "recv" then
-- 对接收到的publish数据处理
mqtts_receiver.proc(data, payload, metas)
-- 发送成功publish数据
-- datanumber类型,表示message id
elseif event == "sent" then
-- 发送消息通知 mqtts sender task
sys.sendMsg(mqtts_sender.TASK_NAME, "MQTT_EVENT", "PUBLISH_OK", data)
-- 服务器断开mqtt连接
elseif event == "disconnect" then
-- 发送消息通知 mqtts main task
sys.sendMsg(TASK_NAME, "MQTT_EVENT", "DISCONNECTED", false)
-- 收到服务器的心跳应答
elseif event == "pong" then
-- 接收到数据,通知网络环境检测看门狗功能模块进行喂狗
sys.publish("FEED_NETWORK_WATCHDOG")
-- 严重异常,本地会主动断开连接
-- data:string类型,表示具体的异常,有以下几种:
-- "connect"tcp连接失败
-- "tx":数据发送失败
-- "conack"mqtt connect后,服务器应答CONNACK鉴权失败,失败码为payloadnumber类型)
-- "other":其他异常
elseif event == "error" then
if data == "connect" or data == "conack" then
-- 发送消息通知 mqtts main task,连接失败
sys.sendMsg(TASK_NAME, "MQTT_EVENT", "CONNECT", false)
elseif data == "other" or data == "tx" then
-- 发送消息通知 mqtts main task,出现异常
sys.sendMsg(TASK_NAME, "MQTT_EVENT", "ERROR")
end
end
end
-- mqtts main task 的任务处理函数
local function mqtts_client_main_task_func()
local mqtt_client
local result, msg
while true do
-- 如果当前时间点设置的默认网卡还没有连接成功,一直在这里循环等待
while not socket.adapter(socket.dft()) do
log.warn("mqtts_client_main_task_func", "wait IP_READY", socket.dft())
-- 在此处阻塞等待默认网卡连接成功的消息"IP_READY"
-- 或者等待1秒超时退出阻塞等待状态;
-- 注意:此处的1000毫秒超时不要修改的更长;
-- 因为当使用exnetif.set_priority_order配置多个网卡连接外网的优先级时,会隐式的修改默认使用的网卡
-- 当exnetif.set_priority_order的调用时序和此处的socket.adapter(socket.dft())判断时序有可能不匹配
-- 此处的1秒,能够保证,即使时序不匹配,也能1秒钟退出阻塞状态,再去判断socket.adapter(socket.dft())
sys.waitUntil("IP_READY", 1000)
end
-- 检测到了IP_READY消息
log.info("mqtts_client_main_task_func", "recv IP_READY", socket.dft())
-- 清空此task绑定的消息队列中的未处理的消息
sys.cleanMsg(TASK_NAME)
-- 创建mqtt client对象,ssl连接,不需要证书校验
mqtt_client = mqtt.create(nil, SERVER_ADDR, SERVER_PORT, true)
-- 如果创建mqtt client对象失败
if not mqtt_client then
log.error("mqtts_client_main_task_func", "mqtt.create error")
goto EXCEPTION_PROC
end
-- 配置mqtt client对象的client idusernamepassword和clean session标志
result = mqtt_client:auth(TASK_NAME..mobile.imei(), USERNAME, PASSWORD, true)
-- 如果配置失败
if not result then
log.error("mqtts_client_main_task_func", "mqtt_client:auth error")
goto EXCEPTION_PROC
end
-- 注册mqtt client对象的事件回调函数
mqtt_client:on(mqtts_client_event_cbfunc)
-- 设置mqtt keepalive时间为120秒
-- 如果没有设置,内核固件中默认为180秒
-- 有需要的话,可以打开注释
-- mqtt_client:keepalive(120)
-- 设置遗嘱消息,有需要的话,可以打开注释
-- mqtt_client:will(TOPIC_PREFIX .. "/status", "offline")
-- 连接server
result = mqtt_client:connect()
-- 如果连接server失败
if not result then
log.error("mqtts_client_main_task_func", "mqtt_client:connect error")
goto EXCEPTION_PROC
end
-- 连接、断开连接、订阅、取消订阅、异常等各种事件的处理调度逻辑
while true do
-- 等待"MQTT_EVENT"消息
msg = sys.waitMsg(TASK_NAME, "MQTT_EVENT")
log.info("mqtts_client_main_task_func waitMsg", msg[2], msg[3], msg[4])
-- connect连接结果
-- msg[3]表示连接结果,true为连接成功,false为连接失败
if msg[2] == "CONNECT" then
-- mqtt连接成功
if msg[3] then
log.info("mqtts_client_main_task_func", "connect success")
-- 通知数据发送应用模块,MQTT连接成功
sys.sendMsg(mqtts_sender.TASK_NAME, "MQTT_EVENT", "CONNECT_OK", mqtt_client)
-- mqtt连接失败
else
log.info("mqtts_client_main_task_func", "connect error")
-- 退出循环,发起重连
break
end
-- subscribe订阅结果
-- msg[3]表示订阅结果,true为订阅成功,false为订阅失败
elseif msg[2] == "SUBSCRIBE" then
-- 订阅成功
if msg[3] then
log.info("mqtts_client_main_task_func", "subscribe success", "qos: "..(msg[4] or "nil"))
-- 订阅失败
else
log.error("mqtts_client_main_task_func", "subscribe error", "code", msg[4])
-- 主动断开mqtt client连接
mqtt_client:disconnect()
-- 发送disconnect之后,此处延时1秒,给数据发送预留一点儿时间,发送到服务器;
-- 即使1秒的时间不足以发送给服务器也没关系;对服务器来说,mqtt客户端只是没有优雅的断开,不影响什么实质功能;
sys.wait(1000)
break
end
-- unsubscribe取消订阅成功
elseif msg[2] == "UNSUBSCRIBE" then
log.info("mqtts_client_main_task_func", "unsubscribe success")
-- 需要主动关闭mqtt连接
-- 用户需要主动关闭mqtt连接时,可以调用sys.sendMsg(TASK_NAME, "MQTT_EVENT", "CLOSE")
elseif msg[2] == "CLOSE" then
-- 主动断开mqtt client连接
mqtt_client:disconnect()
-- 发送disconnect之后,此处延时1秒,给数据发送预留一点儿时间,发送到服务器;
-- 即使1秒的时间不足以发送给服务器也没关系;对服务器来说,mqtt客户端只是没有优雅的断开,不影响什么实质功能;
sys.wait(1000)
break
-- 被动关闭了mqtt连接
-- 被网络或者服务器断开了连接
elseif msg[2] == "DISCONNECTED" then
break
-- 出现了其他异常
elseif msg[2] == "ERROR" then
break
end
end
-- 出现异常
::EXCEPTION_PROC::
-- 清空此task绑定的消息队列中的未处理的消息
sys.cleanMsg(TASK_NAME)
-- 通知mqtts sender数据发送应用模块的task,MQTT连接已经断开
sys.sendMsg(mqtts_sender.TASK_NAME, "MQTT_EVENT", "DISCONNECTED")
-- 如果存在mqtt client对象
if mqtt_client then
-- 关闭mqtt client,并且释放mqtt client对象
mqtt_client:close()
mqtt_client = nil
end
-- 5秒后跳转到循环体开始位置,自动发起重连
sys.wait(5000)
end
end
--创建并且启动一个task
--运行这个task的处理函数mqtts_client_main_task_func
sys.taskInitEx(mqtts_client_main_task_func, TASK_NAME)
@@ -0,0 +1,64 @@
--[[
@module mqtts_receiver
@summary mqtts cient数据接收处理应用功能模块
@version 1.0
@date 2025.07.29
@author 马梦阳
@usage
本文件为mqtts client 数据接收应用功能模块,核心业务逻辑为:
处理接收到的publish数据,同时将数据发送给其他应用功能模块做进一步处理;
本文件的对外接口有2个:
1、mqtts_receiver.proc(topic, payload, metas)publish数据处理入口,在mqtts_main.lua中调用;
2、sys.publish("RECV_DATA_FROM_SERVER", "recv from mqtt ssl server: ", topic, payload)
将接收到的publish中的topic和payload数据通过消息"RECV_DATA_FROM_SERVER"发布出去;
需要处理数据的应用功能模块订阅处理此消息即可,本demo项目中uart_app.lua中订阅处理了本消息;
]]
local mqtts_receiver = {}
--[[
处理接收到的publish数据
@api mqtts_receiver.proc(topic, payload, metas)
@param1 topic string
表示publish主题
@param2 payload string
表示publish数据负载
@param2 payload string
表示publish数据负载
@param3 metas table
表示publish报文的一些参数;格式如下:
{
qos: number类型,取值范围0,1,2
retainnumber类型,取值范围0,1
dupnumber类型,取值范围0,1
message_id: number类型
}
@return1 result nil
@usage
mqtts_receiver.proc(topic, payload, metas)
]]
function mqtts_receiver.proc(topic, payload, metas)
log.info("mqtts_receiver.proc", topic, payload:len(), json.encode(metas))
-- 接收到数据,通知网络环境检测看门狗功能模块进行喂狗
sys.publish("FEED_NETWORK_WATCHDOG")
-- 将topic和payload通过"RECV_DATA_FROM_SERVER"消息publish出去,给其他应用模块处理
sys.publish("RECV_DATA_FROM_SERVER", "recv from mqtt ssl server: ", topic, payload)
-- 也可以直接在此处编写代码,处理topic和payload
end
return mqtts_receiver
@@ -0,0 +1,154 @@
--[[
@module mqtts_sender
@summary mqtts client数据发送应用功能模块
@version 1.0
@date 2025.07.29
@author 马梦阳
@usage
本文件为mqtts client 数据发送应用功能模块,核心业务逻辑为:
1、sys.subscribe("SEND_DATA_REQ", send_data_req_proc_func)订阅"SEND_DATA_REQ"消息,将其他应用模块需要发送的数据存储到队列send_queue中;
2、mqtts sender task接收"CONNECT OK"、"PUBLISH_REQ"、"PUBLISH OK"三种类型的"MQTT_EVENT"消息,遍历队列send_queue,逐条发送数据到server
3、mqtts sender task接收"DISCONNECTED"类型的"MQTT_EVENT"消息,丢弃掉队列send_queue中未发送的数据;
4、任何一条数据无论发送成功还是失败,只要这条数据有回调函数,都会通过回调函数通知数据发送方;
本文件的对外接口有1个:
1、sys.subscribe("SEND_DATA_REQ", send_data_req_proc_func):订阅"SEND_DATA_REQ"消息;
其他应用模块如果需要发送数据,直接sys.publish这个消息即可,将需要发送的topic,payload和qos以及回调函数和回调参数一起publish出去;
本demo项目中uart_app.lua和timer_app.lua中publish了这个消息;
]]
local mqtts_sender = {}
--[[
数据发送队列,数据结构为:
{
[1] = {topic="topic1", payload="payload1", qos=0, cb={func=callback_function1, para=callback_para1}},
[2] = {topic="topic2", payload="payload2", qos=1, cb={func=callback_function2, para=callback_para2}},
[3] = {topic="topic3", payload="payload3", qos=2, cb={func=callback_function3, para=callback_para3}},
}
topic的内容为publish的主题,string类型,必须存在;
payload的内容为publish的负载数据,string类型,必须存在;
qos的内容为publish的质量等级,number类型,取值范围0,1,2,可选,如果用户没有指定,默认为0;
cb.func的内容为数据发送结果的用户回调函数,可以不存在;
cb.para的内容为数据发送结果的用户回调函数的回调参数,可以不存在;
]]
local send_queue = {}
-- mqtts client的任务名前缀
mqtts_sender.TASK_NAME_PREFIX = "mqtts_"
-- mqtts_client_sender的任务名
mqtts_sender.TASK_NAME = mqtts_sender.TASK_NAME_PREFIX.."sender"
-- "SEND_DATA_REQ"消息的处理函数
local function send_data_req_proc_func(tag, topic, payload, qos, cb)
-- 将原始数据增加前缀,然后插入到发送队列send_queue中
table.insert(send_queue, {topic=topic, payload="send from "..tag..": "..payload, qos=qos or 0, cb=cb})
-- 发送消息通知 mqtts sender task,有新数据等待发送
sys.sendMsg(mqtts_sender.TASK_NAME, "MQTT_EVENT", "PUBLISH_REQ")
end
-- 按照顺序发送send_queue中的数据
-- 如果调用publish接口成功,则返回当前正在发送的数据项
-- 如果调用publish接口失败,通知回调函数发送失败后,继续发送下一条数据
local function publish_item(mqtt_client)
local item
-- 如果发送队列中有数据等待发送
while #send_queue>0 do
-- 取出来第一条数据赋值给item
-- 同时从队列send_queue中删除这一条数据
item = table.remove(send_queue, 1)
-- publish数据
-- result表示调用publish接口的同步结果,返回值有以下几种:
-- 如果失败,返回nil
-- 如果成功,number类型,qos为0时直接返回0qos为1或者2时返回publish报文的message id
result = mqtt_client:publish(item.topic, item.payload, item.qos)
-- publish接口调用成功
if result then
return item
-- publish接口调用失败
else
-- 如果当前发送的数据有用户回调函数,则执行用户回调函数
if item.cb and item.cb.func then
item.cb.func(false, item.cb.para)
end
end
end
end
local function publish_item_cbfunc(item, result)
if item then
-- 如果当前发送的数据有用户回调函数,则执行用户回调函数
if item.cb and item.cb.func then
item.cb.func(result, item.cb.para)
end
end
end
-- mqtts client sender的任务处理函数
local function mqtts_client_sender_task_func()
local mqtt_client
local send_item
local result, msg
while true do
-- 等待"MQTT_EVENT"消息
msg = sys.waitMsg(mqtts_sender.TASK_NAME, "MQTT_EVENT")
-- mqtt连接成功
-- msg[3]表示mqtt client对象
if msg[2] == "CONNECT_OK" then
mqtt_client = msg[3]
-- 发送send_queue中的数据
send_item = publish_item(mqtt_client)
-- mqtt publish数据请求
elseif msg[2] == "PUBLISH_REQ" then
-- 如果mqtt client对象存在,并且没有正在等待发送结果的发送数据项
if mqtt_client and not send_item then
-- 发送send_queue中的数据
send_item = publish_item(mqtt_client)
end
-- mqtt publish数据成功
elseif msg[2] == "PUBLISH_OK" then
-- publish成功,执行回调函数通知发送方
publish_item_cbfunc(send_item, true)
-- publish成功,通知网络环境检测看门狗功能模块进行喂狗
sys.publish("FEED_NETWORK_WATCHDOG")
-- 发送send_queue中的数据
send_item = publish_item(mqtt_client)
-- mqtt断开连接
elseif msg[2] == "DISCONNECTED" then
-- 清空mqtt client对象
mqtt_client = nil
-- 如果存在正在等待发送结果的发送项,执行回调函数通知发送方失败
publish_item_cbfunc(send_item, false)
-- 如果发送队列中有数据等待发送
while #send_queue>0 do
-- 取出来第一条数据赋值给send_item
-- 同时从队列send_queue中删除这一条数据
send_item = table.remove(send_queue,1)
-- 执行回调函数通知发送方失败
publish_item_cbfunc(send_item, false)
end
-- 当前没有正在等待发送结果的发送项
send_item = nil
end
end
end
-- 订阅"SEND_DATA_REQ"消息;
-- 其他应用模块如果需要发送数据,直接sys.publish这个消息即可,将需要发送的数据以及回调函数和回调参数一起publish出去;
-- 本demo项目中uart_app.lua和timer_app.lua中publish了这个消息;
sys.subscribe("SEND_DATA_REQ", send_data_req_proc_func)
--创建并且启动一个task
--运行这个task的处理函数mqtts_client_sender_task_func
sys.taskInitEx(mqtts_client_sender_task_func, mqtts_sender.TASK_NAME)
return mqtts_sender
@@ -0,0 +1,289 @@
--[[
@module mqtts_ca_main
@summary mqtts ca client 主应用功能模块
@version 1.0
@date 2025.07.28
@author 马梦阳
@usage
本文件为mqtts ca client 主应用功能模块,核心业务逻辑为:
1、创建一个mqtts ca client,连接server
2、处理连接/订阅/取消订阅/异常逻辑,出现异常后执行重连动作;
3、调用mqtts_ca_receiver的外部接口mqtts_ca_receiver.proc,对接收到的publish数据进行处理;
4、调用sys.sendMsg接口,发送"CONNECT OK"、"PUBLISH OK"和"DISCONNECTED"三种类型的"MQTT_EVENT"消息到mqtts_ca_sender的task,控制publish数据发送逻辑;
5、收到MQTT心跳应答后,执行sys.publish("FEED_NETWORK_WATCHDOG") 对网络环境检测看门狗功能模块进行喂狗;
本文件没有对外接口,直接在main.lua中require "mqtts_ca_main"就可以加载运行;
]]
-- 加载sntp时间同步应用功能模块(证书校验的mqtt ssl需要时间同步功能)
require "sntp_app"
-- 加载mqtts ca client数据接收功能模块
local mqtts_ca_receiver = require "mqtts_ca_receiver"
-- 加载mqtts ca client数据发送功能模块
local mqtts_ca_sender = require "mqtts_ca_sender"
-- mqtts ca服务器地址和端口
-- 这里使用的地址和端口,会不定期重启或维护,仅能用作测试用途,不可商用,说不定哪一天就关闭了
-- 用户开发项目时,替换为自己的商用服务器地址和端口
local SERVER_ADDR = "airtest.openluat.com"
local SERVER_PORT = 8888
-- mqtt连接时的用户名和密码
-- 必须填写,否则会连接失败
local USERNAME = "mqtt_hz_test_2"
local PASSWORD = "3bEKUb"
-- mqtts_ca_main的任务名
local TASK_NAME = mqtts_ca_sender.TASK_NAME_PREFIX.."main"
-- mqtt主题的前缀:IMEI号
local TOPIC_PREFIX = mobile.imei()
-- mqtts ca client的事件回调函数
local function mqtts_ca_client_event_cbfunc(mqtt_client, event, data, payload, metas)
log.info("mqtts_ca_client_event_cbfunc", mqtt_client, event, data, payload, json.encode(metas))
-- mqtt连接成功
if event == "conack" then
sys.sendMsg(TASK_NAME, "MQTT_EVENT", "CONNECT", true)
-- 订阅单主题
-- 第二个参数表示qos,取值范围为0,1,2,如果不设置,默认为0
if not mqtt_client:subscribe(TOPIC_PREFIX .. "/down") then
sys.sendMsg(TASK_NAME, "MQTT_EVENT", "SUBSCRIBE", false, -1)
end
-- 订阅多主题,如果有需要,打开注释
-- 表中的每一个订阅主题的格式为[topic]=qos
-- if not mqtt_client:subscribe(
-- {
-- [(TOPIC_PREFIX .. "/data"]=0,
-- [(TOPIC_PREFIX .. "/cmd"]=1
-- }
-- ) then
-- sys.sendMsg(TASK_NAME, "MQTT_EVENT", "SUBSCRIBE", false, -1)
-- end
-- 订阅结果
-- data:订阅应答结果,true为成功,false为失败
-- payloadnumber类型;成功时表示qos,取值范围为0,1,2;失败时表示失败码,一般是0x80
elseif event == "suback" then
-- 发送消息通知 mqtts ca main task
sys.sendMsg(TASK_NAME, "MQTT_EVENT", "SUBSCRIBE", data, payload)
-- 取消订阅成功
elseif event == "unsuback" then
-- 发送消息通知 mqtts ca main task
sys.sendMsg(TASK_NAME, "MQTT_EVENT", "UNSUBSCRIBE", true)
-- 接收到服务器下发的publish数据
-- datastring类型,表示topic
-- payloadstring类型,表示payload
-- metastable类型,数据内容如下
-- {
-- qos: number类型,取值范围0,1,2
-- retainnumber类型,取值范围0,1
-- dupnumber类型,取值范围0,1
-- message_id: number类型
-- }
elseif event == "recv" then
-- 对接收到的publish数据处理
mqtts_ca_receiver.proc(data, payload, metas)
-- 发送成功publish数据
-- datanumber类型,表示message id
elseif event == "sent" then
-- 发送消息通知 mqtts ca sender task
sys.sendMsg(mqtts_ca_sender.TASK_NAME, "MQTT_EVENT", "PUBLISH_OK", data)
-- 服务器断开mqtt连接
elseif event == "disconnect" then
-- 发送消息通知 mqtts ca main task
sys.sendMsg(TASK_NAME, "MQTT_EVENT", "DISCONNECTED", false)
-- 收到服务器的心跳应答
elseif event == "pong" then
-- 接收到数据,通知网络环境检测看门狗功能模块进行喂狗
sys.publish("FEED_NETWORK_WATCHDOG")
-- 严重异常,本地会主动断开连接
-- data:string类型,表示具体的异常,有以下几种:
-- "connect"tcp连接失败
-- "tx":数据发送失败
-- "conack"mqtt connect后,服务器应答CONNACK鉴权失败,失败码为payloadnumber类型)
-- "other":其他异常
elseif event == "error" then
if data == "connect" or data == "conack" then
-- 发送消息通知 mqtts ca main task,连接失败
sys.sendMsg(TASK_NAME, "MQTT_EVENT", "CONNECT", false)
elseif data == "other" or data == "tx" then
-- 发送消息通知 mqtts ca main task,出现异常
sys.sendMsg(TASK_NAME, "MQTT_EVENT", "ERROR")
end
end
end
-- mqtts ca main task 的任务处理函数
local function mqtts_ca_client_main_task_func()
local mqtt_client
local result, msg
-- 用来验证server证书是否合法的ca证书文件为openluat_root_ca.crt,此证书永久有效
-- 将这个ca证书文件的内容读取出来,赋值给server_ca_cert
-- 注意:此处的ca证书文件仅用来验证airtest.openluat.com的server证书
-- 在有效期之前,airtest.openluat.com可能会更换server证书
-- 如果server证书更换后,此处验证使用的openluat_root_ca.crt也可能需要更换
-- 如果你使用的是自己的server,要替换为自己server证书对应的ca证书文件
local server_ca_cert = io.readFile("/luadb/openluat_root_ca.crt")
while true do
-- 如果当前时间点设置的默认网卡还没有连接成功,一直在这里循环等待
while not socket.adapter(socket.dft()) do
log.warn("mqtts_ca_client_main_task_func", "wait IP_READY", socket.dft())
-- 在此处阻塞等待默认网卡连接成功的消息"IP_READY"
-- 或者等待1秒超时退出阻塞等待状态;
-- 注意:此处的1000毫秒超时不要修改的更长;
-- 因为当使用exnetif.set_priority_order配置多个网卡连接外网的优先级时,会隐式的修改默认使用的网卡
-- 当exnetif.set_priority_order的调用时序和此处的socket.adapter(socket.dft())判断时序有可能不匹配
-- 此处的1秒,能够保证,即使时序不匹配,也能1秒钟退出阻塞状态,再去判断socket.adapter(socket.dft())
sys.waitUntil("IP_READY", 1000)
end
-- 检测到了IP_READY消息
log.info("mqtts_ca_client_main_task_func", "recv IP_READY", socket.dft())
-- 清空此task绑定的消息队列中的未处理的消息
sys.cleanMsg(TASK_NAME)
-- 创建mqtt client对象,ssl连接,单向证书校验
-- client仅单向校验server的证书,server不校验client的证书和密钥文件
-- 如果做证书校验,需要特别注意以下几点:
-- 1、证书校验前,设备端必须同步为正确的时间,因为校验过程中会检查ca证书以及server证书中的有效期是否合法;本demo中的sntp_app.lua会同步时间;
-- 2、任何证书都有有效期,无论是ca证书还是server证书,必须在有效期截止之前,及时更换证书,延长有效期,否则证书校验会失败;
-- 3、如果要更换ca证书,需要在设备端远程升级,必须保证ca证书失效之前升级成功,否则校验失败,就无法连接server;
-- 综上所述,证书校验虽然安全,可以验证身份,但是后续维护成本比较高;除非有需要,否则可以不配置证书校验功能;
mqtt_client = mqtt.create(nil, SERVER_ADDR, SERVER_PORT, {server_cert = server_ca_cert})
-- 如果创建mqtt client对象失败
if not mqtt_client then
log.error("mqtts_ca_client_main_task_func", "mqtt.create error")
goto EXCEPTION_PROC
end
-- 配置mqtt client对象的client idusernamepassword和clean session标志
result = mqtt_client:auth(TASK_NAME..mobile.imei(), USERNAME, PASSWORD, true)
-- 如果配置失败
if not result then
log.error("mqtts_ca_client_main_task_func", "mqtt_client:auth error")
goto EXCEPTION_PROC
end
-- 注册mqtt client对象的事件回调函数
mqtt_client:on(mqtts_ca_client_event_cbfunc)
-- 设置mqtt keepalive时间为120秒
-- 如果没有设置,内核固件中默认为180秒
-- 有需要的话,可以打开注释
-- mqtt_client:keepalive(120)
-- 设置遗嘱消息,有需要的话,可以打开注释
-- mqtt_client:will(TOPIC_PREFIX .. "/status", "offline")
-- 配置开启debug信息,有需要的话,可以打开注释
-- mqtt_client:debug(true)
-- socket.sslLog(3)
-- 连接server
result = mqtt_client:connect()
-- 如果连接server失败
if not result then
log.error("mqtts_ca_client_main_task_func", "mqtt_client:connect error")
goto EXCEPTION_PROC
end
-- 连接、断开连接、订阅、取消订阅、异常等各种事件的处理调度逻辑
while true do
-- 等待"MQTT_EVENT"消息
msg = sys.waitMsg(TASK_NAME, "MQTT_EVENT")
log.info("mqtts_ca_client_main_task_func waitMsg", msg[2], msg[3], msg[4])
-- connect连接结果
-- msg[3]表示连接结果,true为连接成功,false为连接失败
if msg[2] == "CONNECT" then
-- mqtt连接成功
if msg[3] then
log.info("mqtts_ca_client_main_task_func", "connect success")
-- 通知数据发送应用模块,MQTT连接成功
sys.sendMsg(mqtts_ca_sender.TASK_NAME, "MQTT_EVENT", "CONNECT_OK", mqtt_client)
-- mqtt连接失败
else
log.info("mqtts_ca_client_main_task_func", "connect error")
-- 退出循环,发起重连
break
end
-- subscribe订阅结果
-- msg[3]表示订阅结果,true为订阅成功,false为订阅失败
elseif msg[2] == "SUBSCRIBE" then
-- 订阅成功
if msg[3] then
log.info("mqtts_ca_client_main_task_func", "subscribe success", "qos: "..(msg[4] or "nil"))
-- 订阅失败
else
log.error("mqtts_ca_client_main_task_func", "subscribe error", "code", msg[4])
-- 主动断开mqtt client连接
mqtt_client:disconnect()
-- 发送disconnect之后,此处延时1秒,给数据发送预留一点儿时间,发送到服务器;
-- 即使1秒的时间不足以发送给服务器也没关系;对服务器来说,mqtt客户端只是没有优雅的断开,不影响什么实质功能;
sys.wait(1000)
break
end
-- unsubscribe取消订阅成功
elseif msg[2] == "UNSUBSCRIBE" then
log.info("mqtts_ca_client_main_task_func", "unsubscribe success")
-- 需要主动关闭mqtt连接
-- 用户需要主动关闭mqtt连接时,可以调用sys.sendMsg(TASK_NAME, "MQTT_EVENT", "CLOSE")
elseif msg[2] == "CLOSE" then
-- 主动断开mqtt client连接
mqtt_client:disconnect()
-- 发送disconnect之后,此处延时1秒,给数据发送预留一点儿时间,发送到服务器;
-- 即使1秒的时间不足以发送给服务器也没关系;对服务器来说,mqtt客户端只是没有优雅的断开,不影响什么实质功能;
sys.wait(1000)
break
-- 被动关闭了mqtt连接
-- 被网络或者服务器断开了连接
elseif msg[2] == "DISCONNECTED" then
break
-- 出现了其他异常
elseif msg[2] == "ERROR" then
break
end
end
-- 出现异常
::EXCEPTION_PROC::
-- 清空此task绑定的消息队列中的未处理的消息
sys.cleanMsg(TASK_NAME)
-- 通知mqtts ca sender数据发送应用模块的task,MQTT连接已经断开
sys.sendMsg(mqtts_ca_sender.TASK_NAME, "MQTT_EVENT", "DISCONNECTED")
-- 如果存在mqtt client对象
if mqtt_client then
-- 关闭mqtt client,并且释放mqtt client对象
mqtt_client:close()
mqtt_client = nil
end
-- 5秒后跳转到循环体开始位置,自动发起重连
sys.wait(5000)
end
end
--创建并且启动一个task
--运行这个task的处理函数mqtts_ca_client_main_task_func
sys.taskInitEx(mqtts_ca_client_main_task_func, TASK_NAME)
@@ -0,0 +1,64 @@
--[[
@module mqtts_ca_receiver
@summary mqtts ca cient数据接收处理应用功能模块
@version 1.0
@date 2025.07.29
@author 马梦阳
@usage
本文件为mqtts ca client 数据接收应用功能模块,核心业务逻辑为:
处理接收到的publish数据,同时将数据发送给其他应用功能模块做进一步处理;
本文件的对外接口有2个:
1、mqtts_ca_receiver.proc(topic, payload, metas)publish数据处理入口,在mqtts_ca_main.lua中调用;
2、sys.publish("RECV_DATA_FROM_SERVER", "recv from mqtt ssl ca server: ", topic, payload)
将接收到的publish中的topic和payload数据通过消息"RECV_DATA_FROM_SERVER"发布出去;
需要处理数据的应用功能模块订阅处理此消息即可,本demo项目中uart_app.lua中订阅处理了本消息;
]]
local mqtts_ca_receiver = {}
--[[
处理接收到的publish数据
@api mqtts_ca_receiver.proc(topic, payload, metas)
@param1 topic string
表示publish主题
@param2 payload string
表示publish数据负载
@param2 payload string
表示publish数据负载
@param3 metas table
表示publish报文的一些参数;格式如下:
{
qos: number类型,取值范围0,1,2
retainnumber类型,取值范围0,1
dupnumber类型,取值范围0,1
message_id: number类型
}
@return1 result nil
@usage
mqtts_ca_receiver.proc(topic, payload, metas)
]]
function mqtts_ca_receiver.proc(topic, payload, metas)
log.info("mqtts_ca_receiver.proc", topic, payload:len(), json.encode(metas))
-- 接收到数据,通知网络环境检测看门狗功能模块进行喂狗
sys.publish("FEED_NETWORK_WATCHDOG")
-- 将topic和payload通过"RECV_DATA_FROM_SERVER"消息publish出去,给其他应用模块处理
sys.publish("RECV_DATA_FROM_SERVER", "recv from mqtt ssl ca server: ", topic, payload)
-- 也可以直接在此处编写代码,处理topic和payload
end
return mqtts_ca_receiver
@@ -0,0 +1,154 @@
--[[
@module mqtts_ca_sender
@summary mqtts ca client数据发送应用功能模块
@version 1.0
@date 2025.07.29
@author 马梦阳
@usage
本文件为mqtts ca client 数据发送应用功能模块,核心业务逻辑为:
1、sys.subscribe("SEND_DATA_REQ", send_data_req_proc_func)订阅"SEND_DATA_REQ"消息,将其他应用模块需要发送的数据存储到队列send_queue中;
2、mqtts ca sender task接收"CONNECT OK"、"PUBLISH_REQ"、"PUBLISH OK"三种类型的"MQTT_EVENT"消息,遍历队列send_queue,逐条发送数据到server
3、mqtts ca sender task接收"DISCONNECTED"类型的"MQTT_EVENT"消息,丢弃掉队列send_queue中未发送的数据;
4、任何一条数据无论发送成功还是失败,只要这条数据有回调函数,都会通过回调函数通知数据发送方;
本文件的对外接口有1个:
1、sys.subscribe("SEND_DATA_REQ", send_data_req_proc_func):订阅"SEND_DATA_REQ"消息;
其他应用模块如果需要发送数据,直接sys.publish这个消息即可,将需要发送的topic,payload和qos以及回调函数和回调参数一起publish出去;
本demo项目中uart_app.lua和timer_app.lua中publish了这个消息;
]]
local mqtts_ca_sender = {}
--[[
数据发送队列,数据结构为:
{
[1] = {topic="topic1", payload="payload1", qos=0, cb={func=callback_function1, para=callback_para1}},
[2] = {topic="topic2", payload="payload2", qos=1, cb={func=callback_function2, para=callback_para2}},
[3] = {topic="topic3", payload="payload3", qos=2, cb={func=callback_function3, para=callback_para3}},
}
topic的内容为publish的主题,string类型,必须存在;
payload的内容为publish的负载数据,string类型,必须存在;
qos的内容为publish的质量等级,number类型,取值范围0,1,2,可选,如果用户没有指定,默认为0;
cb.func的内容为数据发送结果的用户回调函数,可以不存在;
cb.para的内容为数据发送结果的用户回调函数的回调参数,可以不存在;
]]
local send_queue = {}
-- mqtts ca client的任务名前缀
mqtts_ca_sender.TASK_NAME_PREFIX = "mqtts_ca_"
-- mqtts_ca_client_sender的任务名
mqtts_ca_sender.TASK_NAME = mqtts_ca_sender.TASK_NAME_PREFIX.."sender"
-- "SEND_DATA_REQ"消息的处理函数
local function send_data_req_proc_func(tag, topic, payload, qos, cb)
-- 将原始数据增加前缀,然后插入到发送队列send_queue中
table.insert(send_queue, {topic=topic, payload="send from "..tag..": "..payload, qos=qos or 0, cb=cb})
-- 发送消息通知 mqtts ca sender task,有新数据等待发送
sys.sendMsg(mqtts_ca_sender.TASK_NAME, "MQTT_EVENT", "PUBLISH_REQ")
end
-- 按照顺序发送send_queue中的数据
-- 如果调用publish接口成功,则返回当前正在发送的数据项
-- 如果调用publish接口失败,通知回调函数发送失败后,继续发送下一条数据
local function publish_item(mqtt_client)
local item
-- 如果发送队列中有数据等待发送
while #send_queue>0 do
-- 取出来第一条数据赋值给item
-- 同时从队列send_queue中删除这一条数据
item = table.remove(send_queue, 1)
-- publish数据
-- result表示调用publish接口的同步结果,返回值有以下几种:
-- 如果失败,返回nil
-- 如果成功,number类型,qos为0时直接返回0qos为1或者2时返回publish报文的message id
result = mqtt_client:publish(item.topic, item.payload, item.qos)
-- publish接口调用成功
if result then
return item
-- publish接口调用失败
else
-- 如果当前发送的数据有用户回调函数,则执行用户回调函数
if item.cb and item.cb.func then
item.cb.func(false, item.cb.para)
end
end
end
end
local function publish_item_cbfunc(item, result)
if item then
-- 如果当前发送的数据有用户回调函数,则执行用户回调函数
if item.cb and item.cb.func then
item.cb.func(result, item.cb.para)
end
end
end
-- mqtts_ca client sender的任务处理函数
local function mqtts_ca_client_sender_task_func()
local mqtt_client
local send_item
local result, msg
while true do
-- 等待"MQTT_EVENT"消息
msg = sys.waitMsg(mqtts_ca_sender.TASK_NAME, "MQTT_EVENT")
-- mqtt连接成功
-- msg[3]表示mqtt client对象
if msg[2] == "CONNECT_OK" then
mqtt_client = msg[3]
-- 发送send_queue中的数据
send_item = publish_item(mqtt_client)
-- mqtt publish数据请求
elseif msg[2] == "PUBLISH_REQ" then
-- 如果mqtt client对象存在,并且没有正在等待发送结果的发送数据项
if mqtt_client and not send_item then
-- 发送send_queue中的数据
send_item = publish_item(mqtt_client)
end
-- mqtt publish数据成功
elseif msg[2] == "PUBLISH_OK" then
-- publish成功,执行回调函数通知发送方
publish_item_cbfunc(send_item, true)
-- publish成功,通知网络环境检测看门狗功能模块进行喂狗
sys.publish("FEED_NETWORK_WATCHDOG")
-- 发送send_queue中的数据
send_item = publish_item(mqtt_client)
-- mqtt断开连接
elseif msg[2] == "DISCONNECTED" then
-- 清空mqtt client对象
mqtt_client = nil
-- 如果存在正在等待发送结果的发送项,执行回调函数通知发送方失败
publish_item_cbfunc(send_item, false)
-- 如果发送队列中有数据等待发送
while #send_queue>0 do
-- 取出来第一条数据赋值给send_item
-- 同时从队列send_queue中删除这一条数据
send_item = table.remove(send_queue,1)
-- 执行回调函数通知发送方失败
publish_item_cbfunc(send_item, false)
end
-- 当前没有正在等待发送结果的发送项
send_item = nil
end
end
end
-- 订阅"SEND_DATA_REQ"消息;
-- 其他应用模块如果需要发送数据,直接sys.publish这个消息即可,将需要发送的数据以及回调函数和回调参数一起publish出去;
-- 本demo项目中uart_app.lua和timer_app.lua中publish了这个消息;
sys.subscribe("SEND_DATA_REQ", send_data_req_proc_func)
--创建并且启动一个task
--运行这个task的处理函数mqtts_ca_client_sender_task_func
sys.taskInitEx(mqtts_ca_client_sender_task_func, mqtts_ca_sender.TASK_NAME)
return mqtts_ca_sender
@@ -0,0 +1,31 @@
-----BEGIN CERTIFICATE-----
MIIFazCCA1OgAwIBAgIRAIIQz7DSQONZRGPgu2OCiwAwDQYJKoZIhvcNAQELBQAw
TzELMAkGA1UEBhMCVVMxKTAnBgNVBAoTIEludGVybmV0IFNlY3VyaXR5IFJlc2Vh
cmNoIEdyb3VwMRUwEwYDVQQDEwxJU1JHIFJvb3QgWDEwHhcNMTUwNjA0MTEwNDM4
WhcNMzUwNjA0MTEwNDM4WjBPMQswCQYDVQQGEwJVUzEpMCcGA1UEChMgSW50ZXJu
ZXQgU2VjdXJpdHkgUmVzZWFyY2ggR3JvdXAxFTATBgNVBAMTDElTUkcgUm9vdCBY
MTCCAiIwDQYJKoZIhvcNAQEBBQADggIPADCCAgoCggIBAK3oJHP0FDfzm54rVygc
h77ct984kIxuPOZXoHj3dcKi/vVqbvYATyjb3miGbESTtrFj/RQSa78f0uoxmyF+
0TM8ukj13Xnfs7j/EvEhmkvBioZxaUpmZmyPfjxwv60pIgbz5MDmgK7iS4+3mX6U
A5/TR5d8mUgjU+g4rk8Kb4Mu0UlXjIB0ttov0DiNewNwIRt18jA8+o+u3dpjq+sW
T8KOEUt+zwvo/7V3LvSye0rgTBIlDHCNAymg4VMk7BPZ7hm/ELNKjD+Jo2FR3qyH
B5T0Y3HsLuJvW5iB4YlcNHlsdu87kGJ55tukmi8mxdAQ4Q7e2RCOFvu396j3x+UC
B5iPNgiV5+I3lg02dZ77DnKxHZu8A/lJBdiB3QW0KtZB6awBdpUKD9jf1b0SHzUv
KBds0pjBqAlkd25HN7rOrFleaJ1/ctaJxQZBKT5ZPt0m9STJEadao0xAH0ahmbWn
OlFuhjuefXKnEgV4We0+UXgVCwOPjdAvBbI+e0ocS3MFEvzG6uBQE3xDk3SzynTn
jh8BCNAw1FtxNrQHusEwMFxIt4I7mKZ9YIqioymCzLq9gwQbooMDQaHWBfEbwrbw
qHyGO0aoSCqI3Haadr8faqU9GY/rOPNk3sgrDQoo//fb4hVC1CLQJ13hef4Y53CI
rU7m2Ys6xt0nUW7/vGT1M0NPAgMBAAGjQjBAMA4GA1UdDwEB/wQEAwIBBjAPBgNV
HRMBAf8EBTADAQH/MB0GA1UdDgQWBBR5tFnme7bl5AFzgAiIyBpY9umbbjANBgkq
hkiG9w0BAQsFAAOCAgEAVR9YqbyyqFDQDLHYGmkgJykIrGF1XIpu+ILlaS/V9lZL
ubhzEFnTIZd+50xx+7LSYK05qAvqFyFWhfFQDlnrzuBZ6brJFe+GnY+EgPbk6ZGQ
3BebYhtF8GaV0nxvwuo77x/Py9auJ/GpsMiu/X1+mvoiBOv/2X/qkSsisRcOj/KK
NFtY2PwByVS5uCbMiogziUwthDyC3+6WVwW6LLv3xLfHTjuCvjHIInNzktHCgKQ5
ORAzI4JMPJ+GslWYHb4phowim57iaztXOoJwTdwJx4nLCgdNbOhdjsnvzqvHu7Ur
TkXWStAmzOVyyghqpZXjFaH3pO3JLF+l+/+sKAIuvtd7u+Nxe5AW0wdeRlN8NwdC
jNPElpzVmbUq4JUagEiuTDkHzsxHpFKVK7q4+63SM1N95R1NbdWhscdCb+ZAJzVc
oyi3B43njTOQ5yOf+1CceWxG1bQVs5ZufpsMljq4Ui0/1lvh+wjChP4kqKOJ2qxq
4RgqsahDYVvTH9w7jXbyLeiNdd8XM2w9U/t7y0Ff/9yi0GE44Za4rF2LN9d11TPA
mRGunUHBcnWEvgJBQl9nJEiU0Zsnvgc/ubhPgXRR4Xq37Z0j4r7g1SgEEzwxA57d
emyPxgcYxn/eR44/KJ4EBs+lVDR3veyJm+kXQ99b21/+jh5Xos1AnX5iItreGCc=
-----END CERTIFICATE-----
@@ -0,0 +1,65 @@
--[[
@module sntp_app
@summary sntp时间同步应用功能模块
@version 1.0
@date 2025.07.01
@author 马梦阳
@usage
本文件为sntp时间同步应用功能模块,核心业务逻辑为:
1、连接ntp服务器进行时间同步;
2、如果同步成功,1小时之后重新发起同步动作;
3、如果同步失败,10秒钟之后重新发起同步动作;
本文件没有对外接口,直接在其他应用功能模块中require "sntp_app"就可以加载运行;
]]
-- sntp时间同步的任务处理函数
local function sntp_task_func()
while true do
-- 如果当前时间点设置的默认网卡还没有连接成功,一直在这里循环等待
while not socket.adapter(socket.dft()) do
log.warn("sntp_task_func", "wait IP_READY", socket.dft())
-- 在此处阻塞等待默认网卡连接成功的消息"IP_READY"
-- 或者等待1秒超时退出阻塞等待状态;
-- 注意:此处的1000毫秒超时不要修改的更长;
-- 因为当使用exnetif.set_priority_order配置多个网卡连接外网的优先级时,会隐式的修改默认使用的网卡
-- 当exnetif.set_priority_order的调用时序和此处的socket.adapter(socket.dft())判断时序有可能不匹配
-- 此处的1秒,能够保证,即使时序不匹配,也能1秒钟退出阻塞状态,再去判断socket.adapter(socket.dft())
sys.waitUntil("IP_READY", 1000)
end
-- 检测到了IP_READY消息
log.warn("sntp_task_func", "recv IP_READY")
-- 发起ntp时间同步动作
socket.sntp()
-- 等待ntp时间同步结果,30秒超时失败,通常只需要几百毫秒就能成功
local ret = sys.waitUntil("NTP_UPDATE", 30000)
--同步成功
if ret then
-- 以下是获取/打印时间的演示,注意时区问题
log.info("sntp_task_func", "时间同步成功", "本地时间", os.date())
log.info("sntp_task_func", "时间同步成功", "UTC时间", os.date("!%c"))
log.info("sntp_task_func", "时间同步成功", "RTC时钟(UTC时间)", json.encode(rtc.get()))
log.info("sntp_task_func", "时间同步成功", "本地时间戳", os.time())
local t = os.date("*t")
log.info("sntp_task_func", "时间同步成功", "本地时间os.date() json格式", json.encode(t))
log.info("sntp_task_func", "时间同步成功", "本地时间os.date(os.time())", os.time(t))
-- 正常使用, 一小时一次, 已经足够了, 甚至1天一次也可以
sys.wait(3600000)
--同步失败
else
log.info("sntp_task_func", "时间同步失败")
-- 10秒后重新发起同步动作
sys.wait(10000)
end
end
end
--创建并且启动一个task
--运行这个task的主函数sntp_task_func
sys.taskInit(sntp_task_func)
@@ -0,0 +1,22 @@
-----BEGIN CERTIFICATE-----
MIIDpTCCAY0CCQD1xL7tQqU/BzANBgkqhkiG9w0BAQsFADAXMRUwEwYDVQQDDAxI
WiBNUVRUIENBIDEwHhcNMjUwNzMxMTIzMDAwWhcNMjYwNzMxMTIzMDAwWjASMRAw
DgYDVQQDDAdjbGllbnQxMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA
oCzLLiXTnxQXdbQaOxFtdL+gM67SpFH9z9cmtks7b1c6gYUzFS9Npj2h0pW3Yb6r
D+JGHiF5HmW8vITRSnbWEQzcK9g5QW047LYVYiL8VO2OWjZb/CyWfHrY0aJC8nbl
YgT0P0KpGHPJeMN99gEProwVOC3tNKC2XAFxYcpBpiwMMFq9XdlcAfjMuQqUxbiR
PmWsIZsX/ee/ShUNw8wV2uoB72ScEitrhZllEj9O0oDPfHe9Zd7u1B7CtezGD/qm
pcU9swIJw/YH/kqEVfYybHdP/EG9uyFp/5bJ35tZNcfTyBDpKTERtvCthtZKzVZi
t5YUnuXCNpEUfKlI67i1bQIDAQABMA0GCSqGSIb3DQEBCwUAA4ICAQC1rQMDDVlc
5TluG5o9YBjUlK3Xatm9Rk1QGQj8BB/GN8u01yZ+YOJumxVuDFCIt2yzbZktzROF
oeKnjJ/jW6pHNoUMDK14UBSkLZi4ZDnyJKorWMahKOVz3QEve9Y++EpxfkdbrrUd
vUTlnePxEe0tMV43DpZh/GID3DUWpy+g/DvxiHhk4YP5r5I4na4oNoyQH8as7geS
2Z0/FSxT3a5J+c84Uona5g1FMGhriAWWH9dDDAs7z2wlsiBp7zuwSYOY2q7h+x6Z
na5Uy7HQcFcTFqoKMWkTHz55oTLXEhB0XHgIpY0gX+oFpE+dhVYBIVpo/fpHADNi
FQV3ME4z0X1w/i4Inq7C9HHt0IQ53Yjydkkmf/q7nv1fYN6PdmAXLYIUyVI/fafB
JHx474o+lY9a+NmBegkM84hjMwqWEmiqMcnYCmWMR69CpdSLJL3Wp6ip/6ixp0Vo
cOgYa+/CBGdhQICk1LHWKPHBHhWnhWLabyX1/TD7xoY6Lmqm3H/o+EwtLGGlnAi4
OwU4S5kXR5rqCaYUE0rUhd/7G7NLFbWiXR0dXpYys2IniTybkRgdVIaO3iDMPslY
4baRCZH5ELudBJaotmFOUkvEv91sza7F2SBk7DcWUJ9npEQsScDV/fsY1ronL5dl
4m4zKNEp83X4f1H0yBaazhfilCjMdjQKUg==
-----END CERTIFICATE-----
@@ -0,0 +1,27 @@
-----BEGIN RSA PRIVATE KEY-----
MIIEowIBAAKCAQEAoCzLLiXTnxQXdbQaOxFtdL+gM67SpFH9z9cmtks7b1c6gYUz
FS9Npj2h0pW3Yb6rD+JGHiF5HmW8vITRSnbWEQzcK9g5QW047LYVYiL8VO2OWjZb
/CyWfHrY0aJC8nblYgT0P0KpGHPJeMN99gEProwVOC3tNKC2XAFxYcpBpiwMMFq9
XdlcAfjMuQqUxbiRPmWsIZsX/ee/ShUNw8wV2uoB72ScEitrhZllEj9O0oDPfHe9
Zd7u1B7CtezGD/qmpcU9swIJw/YH/kqEVfYybHdP/EG9uyFp/5bJ35tZNcfTyBDp
KTERtvCthtZKzVZit5YUnuXCNpEUfKlI67i1bQIDAQABAoIBABWDzsrlE9bjWP9b
cMaMYbpz3u3yRqMBzGj1imKkGlirS9WH8cOftftlfe4OZDKB7PJWP3renilCGybr
ZD1vVvO1WYGeKpqqNCI/lwG2l1RFEID3mkV2K8SOr78A36M36znOVoyExLffHanW
AH5hyBx2/oasTJdsGvKi0vsCsuzB+Xwld3JCsjAZhGCeJfCrkiinDGpF9cqmoAyE
k1QC8QlyON577s6bR11hf3iK7F0FijgyiFoUR5YSTlFYn6zUgst9jhIUkemAPo62
kECE1crg+kfZAwm+zlB/rQoSB2ySULUW5mV4TVvkihAyHh8GcQWiUyKFJRgHjdTY
ziRwLDUCgYEAznqKTJQV71tStTdmfzAc4l2Ri2yqH9JDRpsdW/ud+P0E2fcuTrbr
1dJ1SJb7TZNjNLrG8J2dfEwK8FFAzuCAH5PwvoFjEV2pq76bjPhRqSM92w9S4LWt
nRTMuTui/g9Cou0zIGYAxMqQ266JOYyKixgzqa4Etc2SmXeifb6SWlcCgYEAxpdH
1HTmoOLpx0wO+Eaq3k2HAMQ6GzH/w/DgQnjPbHM/YqJtzy/5LjXdN1J0G4mQ2HHt
vNtuRVvKSzSwlr1cAOPQUmtxx1aAtFYZ9M3kkEYXIvOGo/E+3w4JJ7WId5L//eSt
lThdYYk9efDXE5rBqZ1AXuB4Iq5cy1RmLfyL29sCgYEAjIpcGEOzqng+k5ph6bCV
+H430u7j3UUBBc4O7c0WeaO+6Z2Vp35ah+/4ZLBf3S21GO8I+j5OM4yO5YRo2Gd2
zgeklgb7p5ZZJs1Wk11yN+4E7FZB0zHWOlUuA+bTypoYgU1yv1quq0Q11VG3lbRT
mNmgUJ5iFl0Wq98CexR8CYECgYB7PeDe0Vla2NcZznKv0zMw2E4CxMQGooXGNJMH
zW90LyVT8ddRr8Tkz5qkPXA+sqWIHVPbUxXgxzEIcP9x3Lve9YIP1Av9GAqxX4mm
KaxBeK3dC+IvlYRd6pj14qt5AWoKR0MOCrew0vmq1jKzQOQUfbk35/fGFg0SRiNP
QJ4O5wKBgAVPEHx+LPPnWWW1b/IhbB/u0++50iL9skWkonKUrw7w24Xu9UIDXjt4
OjrfgWpSuTyfuwMZbqyKgO9Rft+mZ//wTmhRz77s2bIRFCbnrvAwNgRo4HL2He81
FjJR+WtBndlT0NViIU9cOq6GP6sYo1t/Nd7uICXyvT/BFIrEdV6N
-----END RSA PRIVATE KEY-----
@@ -0,0 +1,303 @@
--[[
@module mqtts_m_ca_main
@summary ssl mutual ca client 主应用功能模块
@version 1.0
@date 2025.07.28
@author 马梦阳
@usage
本文件为mqtts mutual ca client 主应用功能模块,核心业务逻辑为:
1、创建一个mqtts mutual ca client,连接server
2、处理连接/订阅/取消订阅/异常逻辑,出现异常后执行重连动作;
3、调用mqtts_m_ca_receiver的外部接口mqtts_m_ca_receiver.proc,对接收到的publish数据进行处理;
4、调用sys.sendMsg接口,发送"CONNECT OK"、"PUBLISH OK"和"DISCONNECTED"三种类型的"MQTT_EVENT"消息到mqtts_m_ca_sender的task,控制publish数据发送逻辑;
5、收到MQTT心跳应答后,执行sys.publish("FEED_NETWORK_WATCHDOG") 对网络环境检测看门狗功能模块进行喂狗;
本文件没有对外接口,直接在main.lua中require "mqtts_m_ca_main"就可以加载运行;
]]
-- 加载sntp时间同步应用功能模块(证书校验的mqtt ssl需要时间同步功能)
require "sntp_app"
-- 加载mqtts mutual ca client数据接收功能模块
local mqtts_m_ca_receiver = require "mqtts_m_ca_receiver"
-- 加载mqtts mutual ca client数据发送功能模块
local mqtts_m_ca_sender = require "mqtts_m_ca_sender"
-- mqtts mutual ca服务器地址和端口
-- 这里使用的地址和端口,会不定期重启或维护,仅能用作测试用途,不可商用,说不定哪一天就关闭了
-- 用户开发项目时,替换为自己的商用服务器地址和端口
local SERVER_ADDR = "airtest.openluat.com"
local SERVER_PORT = 8886
-- mqtt连接时的用户名和密码
-- 必须填写,否则会连接失败
local USERNAME = "mqtt_hz_test_2"
local PASSWORD = "3bEKUb"
-- mqtts_m_ca_main的任务名
local TASK_NAME = mqtts_m_ca_sender.TASK_NAME_PREFIX.."main"
-- mqtt主题的前缀:IMEI号
local TOPIC_PREFIX = mobile.imei()
-- mqtts mutual ca client的事件回调函数
local function mqtts_m_ca_client_event_cbfunc(mqtt_client, event, data, payload, metas)
log.info("mqtts_m_ca_client_event_cbfunc", mqtt_client, event, data, payload, json.encode(metas))
-- mqtt连接成功
if event == "conack" then
sys.sendMsg(TASK_NAME, "MQTT_EVENT", "CONNECT", true)
-- 订阅单主题
-- 第二个参数表示qos,取值范围为0,1,2,如果不设置,默认为0
if not mqtt_client:subscribe(TOPIC_PREFIX .. "/down") then
sys.sendMsg(TASK_NAME, "MQTT_EVENT", "SUBSCRIBE", false, -1)
end
-- 订阅多主题,如果有需要,打开注释
-- 表中的每一个订阅主题的格式为[topic]=qos
-- if not mqtt_client:subscribe(
-- {
-- [(TOPIC_PREFIX .. "/data"]=0,
-- [(TOPIC_PREFIX .. "/cmd"]=1
-- }
-- ) then
-- sys.sendMsg(TASK_NAME, "MQTT_EVENT", "SUBSCRIBE", false, -1)
-- end
-- 订阅结果
-- data:订阅应答结果,true为成功,false为失败
-- payloadnumber类型;成功时表示qos,取值范围为0,1,2;失败时表示失败码,一般是0x80
elseif event == "suback" then
-- 发送消息通知 mqtts ca main task
sys.sendMsg(TASK_NAME, "MQTT_EVENT", "SUBSCRIBE", data, payload)
-- 取消订阅成功
elseif event == "unsuback" then
-- 发送消息通知 mqtts ca main task
sys.sendMsg(TASK_NAME, "MQTT_EVENT", "UNSUBSCRIBE", true)
-- 接收到服务器下发的publish数据
-- datastring类型,表示topic
-- payloadstring类型,表示payload
-- metastable类型,数据内容如下
-- {
-- qos: number类型,取值范围0,1,2
-- retainnumber类型,取值范围0,1
-- dupnumber类型,取值范围0,1
-- message_id: number类型
-- }
elseif event == "recv" then
-- 对接收到的publish数据处理
mqtts_m_ca_receiver.proc(data, payload, metas)
-- 发送成功publish数据
-- datanumber类型,表示message id
elseif event == "sent" then
-- 发送消息通知 mqtts ca sender task
sys.sendMsg(mqtts_m_ca_sender.TASK_NAME, "MQTT_EVENT", "PUBLISH_OK", data)
-- 服务器断开mqtt连接
elseif event == "disconnect" then
-- 发送消息通知 mqtts mutual ca main task
sys.sendMsg(TASK_NAME, "MQTT_EVENT", "DISCONNECTED", false)
-- 收到服务器的心跳应答
elseif event == "pong" then
-- 接收到数据,通知网络环境检测看门狗功能模块进行喂狗
sys.publish("FEED_NETWORK_WATCHDOG")
-- 严重异常,本地会主动断开连接
-- data:string类型,表示具体的异常,有以下几种:
-- "connect"tcp连接失败
-- "tx":数据发送失败
-- "conack"mqtt connect后,服务器应答CONNACK鉴权失败,失败码为payloadnumber类型)
-- "other":其他异常
elseif event == "error" then
if data == "connect" or data == "conack" then
-- 发送消息通知 mqtts mutual ca main task,连接失败
sys.sendMsg(TASK_NAME, "MQTT_EVENT", "CONNECT", false)
elseif data == "other" or data == "tx" then
-- 发送消息通知 mqtts mutual ca main task,出现异常
sys.sendMsg(TASK_NAME, "MQTT_EVENT", "ERROR")
end
end
end
-- mqtts mutual ca main task 的任务处理函数
local function mqtts_m_ca_client_main_task_func()
local mqtt_client
local result, msg
-- 用来验证server证书是否合法的ca证书文件为openluat_root_ca.crt,此证书永久有效
-- 将这个ca证书文件的内容读取出来,赋值给server_ca_cert
-- 注意:此处的ca证书文件仅用来验证airtest.openluat.com的server证书
-- 在有效期之前,airtest.openluat.com可能会更换server证书
-- 如果server证书更换后,此处验证使用的openluat_root_ca.crt也可能需要更换
-- 如果你使用的是自己的server,需要替换为自己server证书对应的ca证书文件
local server_ca_cert = io.readFile("/luadb/openluat_root_ca.crt")
-- 由airtest.openluat.com签发的client证书文件为airtest_client.crt
-- 此client证书的有效期截止到2026年7月31日
-- 将这个client证书文件的内容读出来,赋值给client_cert
-- 注意:此处的client证书仅在双向认证中用于向airtest.openluat.com验证client身份
-- 如果你使用的是自己的server,需要把client证书替换为你server签发的client证书
local client_cert = io.readFile("/luadb/airtest_client.crt")
-- 与上诉client证书对应的私钥文件是airtest_client.key
-- 私钥文件是纯数学参数,不会到期
-- 将这个私钥文件的内容读出来,赋值给client_key
-- 注意:此处的私钥文件与airtest_client.crt里的公钥成对,仅在双向认证中用于向airtest.openluat.com证明client身份
-- 如果你使用的是自己的server,需要把私钥文件替换为你server签发的client证书对应的私钥文件
local client_key = io.readFile("/luadb/airtest_client.key")
while true do
-- 如果当前时间点设置的默认网卡还没有连接成功,一直在这里循环等待
while not socket.adapter(socket.dft()) do
log.warn("mqtts_m_ca_client_main_task_func", "wait IP_READY", socket.dft())
-- 在此处阻塞等待默认网卡连接成功的消息"IP_READY"
-- 或者等待1秒超时退出阻塞等待状态;
-- 注意:此处的1000毫秒超时不要修改的更长;
-- 因为当使用exnetif.set_priority_order配置多个网卡连接外网的优先级时,会隐式的修改默认使用的网卡
-- 当exnetif.set_priority_order的调用时序和此处的socket.adapter(socket.dft())判断时序有可能不匹配
-- 此处的1秒,能够保证,即使时序不匹配,也能1秒钟退出阻塞状态,再去判断socket.adapter(socket.dft())
sys.waitUntil("IP_READY", 1000)
end
-- 检测到了IP_READY消息
log.info("mqtts_m_ca_client_main_task_func", "recv IP_READY", socket.dft())
-- 清空此task绑定的消息队列中的未处理的消息
sys.cleanMsg(TASK_NAME)
-- 创建mqtt client对象,ssl连接,双向证书校验
-- client校验server的证书,server校验client的证书和私钥文件
-- 如果做证书校验,需要特别注意以下几点:
-- 1、证书校验前,设备端必须同步为正确的时间,因为校验过程中会检查每个证书中的有效期是否合法;本demo中的sntp_app.lua会同步时间;
-- 2、任何证书都有有效期,无论是ca证书、server证书还是client证书,必须在有效期截止之前,及时更换证书,延长有效期,否则证书校验会失败;
-- 3、如果要更换证书,需要在设备端远程升级,必须保证证书失效之前升级成功,否则校验失败,就无法连接server;
-- 综上所述,证书校验虽然安全,可以验证身份,但是后续维护成本比较高;除非有需要,否则可以不配置证书校验功能;
mqtt_client = mqtt.create(nil, SERVER_ADDR, SERVER_PORT, {server_cert = server_ca_cert , client_cert = client_cert , client_key = client_key})
-- 如果创建mqtt client对象失败
if not mqtt_client then
log.error("mqtts_m_ca_client_main_task_func", "mqtt.create error")
goto EXCEPTION_PROC
end
-- 配置mqtt client对象的client idusernamepassword和clean session标志
result = mqtt_client:auth(TASK_NAME..mobile.imei(), USERNAME, PASSWORD, true)
-- 如果配置失败
if not result then
log.error("mqtts_m_ca_client_main_task_func", "mqtt_client:auth error")
goto EXCEPTION_PROC
end
-- 注册mqtt client对象的事件回调函数
mqtt_client:on(mqtts_m_ca_client_event_cbfunc)
-- 设置mqtt keepalive时间为120秒
-- 如果没有设置,内核固件中默认为180秒
-- 有需要的话,可以打开注释
-- mqtt_client:keepalive(120)
-- 设置遗嘱消息,有需要的话,可以打开注释
-- mqtt_client:will(TOPIC_PREFIX .. "/status", "offline")
-- 配置开启debug信息,有需要的话,可以打开注释
-- mqtt_client:debug(true)
-- socket.sslLog(3)
-- 连接server
result = mqtt_client:connect()
-- 如果连接server失败
if not result then
log.error("mqtts_m_ca_client_main_task_func", "mqtt_client:connect error")
goto EXCEPTION_PROC
end
-- 连接、断开连接、订阅、取消订阅、异常等各种事件的处理调度逻辑
while true do
-- 等待"MQTT_EVENT"消息
msg = sys.waitMsg(TASK_NAME, "MQTT_EVENT")
log.info("mqtts_m_ca_client_main_task_func waitMsg", msg[2], msg[3], msg[4])
-- connect连接结果
-- msg[3]表示连接结果,true为连接成功,false为连接失败
if msg[2] == "CONNECT" then
-- mqtt连接成功
if msg[3] then
log.info("mqtts_m_ca_client_main_task_func", "connect success")
-- 通知数据发送应用模块,MQTT连接成功
sys.sendMsg(mqtts_m_ca_sender.TASK_NAME, "MQTT_EVENT", "CONNECT_OK", mqtt_client)
-- mqtt连接失败
else
log.info("mqtts_m_ca_client_main_task_func", "connect error")
-- 退出循环,发起重连
break
end
-- subscribe订阅结果
-- msg[3]表示订阅结果,true为订阅成功,false为订阅失败
elseif msg[2] == "SUBSCRIBE" then
-- 订阅成功
if msg[3] then
log.info("mqtts_m_ca_client_main_task_func", "subscribe success", "qos: "..(msg[4] or "nil"))
-- 订阅失败
else
log.error("mqtts_m_ca_client_main_task_func", "subscribe error", "code", msg[4])
-- 主动断开mqtt client连接
mqtt_client:disconnect()
-- 发送disconnect之后,此处延时1秒,给数据发送预留一点儿时间,发送到服务器;
-- 即使1秒的时间不足以发送给服务器也没关系;对服务器来说,mqtt客户端只是没有优雅的断开,不影响什么实质功能;
sys.wait(1000)
break
end
-- unsubscribe取消订阅成功
elseif msg[2] == "UNSUBSCRIBE" then
log.info("mqtts_m_ca_client_main_task_func", "unsubscribe success")
-- 需要主动关闭mqtt连接
-- 用户需要主动关闭mqtt连接时,可以调用sys.sendMsg(TASK_NAME, "MQTT_EVENT", "CLOSE")
elseif msg[2] == "CLOSE" then
-- 主动断开mqtt client连接
mqtt_client:disconnect()
-- 发送disconnect之后,此处延时1秒,给数据发送预留一点儿时间,发送到服务器;
-- 即使1秒的时间不足以发送给服务器也没关系;对服务器来说,mqtt客户端只是没有优雅的断开,不影响什么实质功能;
sys.wait(1000)
break
-- 被动关闭了mqtt连接
-- 被网络或者服务器断开了连接
elseif msg[2] == "DISCONNECTED" then
break
-- 出现了其他异常
elseif msg[2] == "ERROR" then
break
end
end
-- 出现异常
::EXCEPTION_PROC::
-- 清空此task绑定的消息队列中的未处理的消息
sys.cleanMsg(TASK_NAME)
-- 通知mqtts mutual ca sender数据发送应用模块的task,MQTT连接已经断开
sys.sendMsg(mqtts_m_ca_sender.TASK_NAME, "MQTT_EVENT", "DISCONNECTED")
-- 如果存在mqtt client对象
if mqtt_client then
-- 关闭mqtt client,并且释放mqtt client对象
mqtt_client:close()
mqtt_client = nil
end
-- 5秒后跳转到循环体开始位置,自动发起重连
sys.wait(5000)
end
end
--创建并且启动一个task
--运行这个task的处理函数mqtts_m_ca_client_main_task_func
sys.taskInitEx(mqtts_m_ca_client_main_task_func, TASK_NAME)
@@ -0,0 +1,62 @@
--[[
@module mqtts_m_ca_receiver
@summary mqtts mutual ca cient数据接收处理应用功能模块
@version 1.0
@date 2025.07.29
@author 马梦阳
@usage
本文件为mqtts mutual ca client 数据接收应用功能模块,核心业务逻辑为:
处理接收到的publish数据,同时将数据发送给其他应用功能模块做进一步处理;
本文件的对外接口有2个:
1、mqtts_m_ca_receiver.proc(topic, payload, metas)publish数据处理入口,在mqtts_m_ca_main.lua中调用;
2、sys.publish("RECV_DATA_FROM_SERVER", "recv from mqtt ssl mutual ca server: ", topic, payload)
将接收到的publish中的topic和payload数据通过消息"RECV_DATA_FROM_SERVER"发布出去;
需要处理数据的应用功能模块订阅处理此消息即可,本demo项目中uart_app.lua中订阅处理了本消息;
]]
local mqtts_m_ca_receiver = {}
--[[
处理接收到的publish数据
@api mqtts_m_ca_receiver.proc(topic, payload, metas)
@param1 topic string
表示publish主题
@param2 payload string
表示publish数据负载
@param2 payload string
表示publish数据负载
@param3 metas table
表示publish报文的一些参数;格式如下:
{
qos: number类型,取值范围0,1,2
retainnumber类型,取值范围0,1
dupnumber类型,取值范围0,1
message_id: number类型
}
@return1 result nil
@usage
mqtts_m_ca_receiver.proc(topic, payload, metas)
]]
function mqtts_m_ca_receiver.proc(topic, payload, metas)
log.info("mqtts_m_ca_receiver.proc", topic, payload:len(), json.encode(metas))
-- 接收到数据,通知网络环境检测看门狗功能模块进行喂狗
sys.publish("FEED_NETWORK_WATCHDOG")
-- 将topic和payload通过"RECV_DATA_FROM_SERVER"消息publish出去,给其他应用模块处理
sys.publish("RECV_DATA_FROM_SERVER", "recv from mqtt mutual ssl ca server: ", topic, payload)
-- 也可以直接在此处编写代码,处理topic和payload
end
return mqtts_m_ca_receiver
@@ -0,0 +1,151 @@
--[[
@module mqtts_m_ca_sender
@summary mqtts mutual ca client数据发送应用功能模块
@version 1.0
@date 2025.07.29
@author 马梦阳
@usage
本文件为mqtts mutual ca client 数据发送应用功能模块,核心业务逻辑为:
1、sys.subscribe("SEND_DATA_REQ", send_data_req_proc_func)订阅"SEND_DATA_REQ"消息,将其他应用模块需要发送的数据存储到队列send_queue中;
2、mqtts mutual ca sender task接收"CONNECT OK"、"PUBLISH_REQ"、"PUBLISH OK"三种类型的"MQTT_EVENT"消息,遍历队列send_queue,逐条发送数据到server
3、mqtts mutual ca sender task接收"DISCONNECTED"类型的"MQTT_EVENT"消息,丢弃掉队列send_queue中未发送的数据;
4、任何一条数据无论发送成功还是失败,只要这条数据有回调函数,都会通过回调函数通知数据发送方;
本文件的对外接口有1个:
1、sys.subscribe("SEND_DATA_REQ", send_data_req_proc_func):订阅"SEND_DATA_REQ"消息;
其他应用模块如果需要发送数据,直接sys.publish这个消息即可,将需要发送的topic,payload和qos以及回调函数和回调参数一起publish出去;
本demo项目中uart_app.lua和timer_app.lua中publish了这个消息;
]]
local mqtts_m_ca_sender = {}
--[[
数据发送队列,数据结构为:
{
[1] = {topic="topic1", payload="payload1", qos=0, cb={func=callback_function1, para=callback_para1}},
[2] = {topic="topic2", payload="payload2", qos=1, cb={func=callback_function2, para=callback_para2}},
[3] = {topic="topic3", payload="payload3", qos=2, cb={func=callback_function3, para=callback_para3}},
}
topic的内容为publish的主题,string类型,必须存在;
payload的内容为publish的负载数据,string类型,必须存在;
qos的内容为publish的质量等级,number类型,取值范围0,1,2,可选,如果用户没有指定,默认为0;
cb.func的内容为数据发送结果的用户回调函数,可以不存在;
cb.para的内容为数据发送结果的用户回调函数的回调参数,可以不存在;
]]
local send_queue = {}
-- mqtts mutual ca client的任务名前缀
mqtts_m_ca_sender.TASK_NAME_PREFIX = "mqtts_m_ca_"
-- mqtts_mutual_ca_client_sender的任务名
mqtts_m_ca_sender.TASK_NAME = mqtts_m_ca_sender.TASK_NAME_PREFIX.."sender"
-- "SEND_DATA_REQ"消息的处理函数
local function send_data_req_proc_func(tag, topic, payload, qos, cb)
-- 将原始数据增加前缀,然后插入到发送队列send_queue中
table.insert(send_queue, {topic=topic, payload="send from "..tag..": "..payload, qos=qos or 0, cb=cb})
-- 发送消息通知 mqtts mutual ca sender task,有新数据等待发送
sys.sendMsg(mqtts_m_ca_sender.TASK_NAME, "MQTT_EVENT", "PUBLISH_REQ")
end
-- 按照顺序发送send_queue中的数据
-- 如果调用publish接口成功,则返回当前正在发送的数据项
-- 如果调用publish接口失败,通知回调函数发送失败后,继续发送下一条数据
local function publish_item(mqtt_client)
local item
-- 如果发送队列中有数据等待发送
while #send_queue>0 do
-- 取出来第一条数据赋值给item
-- 同时从队列send_queue中删除这一条数据
item = table.remove(send_queue, 1)
-- publish数据
-- result表示调用publish接口的同步结果,返回值有以下几种:
-- 如果失败,返回nil
-- 如果成功,number类型,qos为0时直接返回0qos为1或者2时返回publish报文的message id
result = mqtt_client:publish(item.topic, item.payload, item.qos)
-- publish接口调用成功
if result then
return item
-- publish接口调用失败
else
-- 如果当前发送的数据有用户回调函数,则执行用户回调函数
if item.cb and item.cb.func then
item.cb.func(false, item.cb.para)
end
end
end
end
local function publish_item_cbfunc(item, result)
if item then
-- 如果当前发送的数据有用户回调函数,则执行用户回调函数
if item.cb and item.cb.func then
item.cb.func(result, item.cb.para)
end
end
end
-- mqtts_m_ca client sender的任务处理函数
local function mqtts_m_ca_client_sender_task_func()
local mqtt_client
local send_item
local result, msg
while true do
-- 等待"MQTT_EVENT"消息
msg = sys.waitMsg(mqtts_m_ca_sender.TASK_NAME, "MQTT_EVENT")
-- mqtt连接成功
-- msg[3]表示mqtt client对象
if msg[2] == "CONNECT_OK" then
mqtt_client = msg[3]
-- 发送send_queue中的数据
send_item = publish_item(mqtt_client)
-- mqtt publish数据请求
elseif msg[2] == "PUBLISH_REQ" then
-- 如果mqtt client对象存在,并且没有正在等待发送结果的发送数据项
if mqtt_client and not send_item then
-- 发送send_queue中的数据
send_item = publish_item(mqtt_client)
end
-- mqtt publish数据成功
elseif msg[2] == "PUBLISH_OK" then
-- publish成功,执行回调函数通知发送方
publish_item_cbfunc(send_item, true)
-- publish成功,通知网络环境检测看门狗功能模块进行喂狗
sys.publish("FEED_NETWORK_WATCHDOG")
-- 发送send_queue中的数据
send_item = publish_item(mqtt_client)
-- mqtt断开连接
elseif msg[2] == "DISCONNECTED" then
-- 清空mqtt client对象
mqtt_client = nil
-- 如果存在正在等待发送结果的发送项,执行回调函数通知发送方失败
publish_item_cbfunc(send_item, false)
-- 如果发送队列中有数据等待发送
while #send_queue>0 do
-- 取出来第一条数据赋值给send_item
-- 同时从队列send_queue中删除这一条数据
send_item = table.remove(send_queue,1)
-- 执行回调函数通知发送方失败
publish_item_cbfunc(send_item, false)
end
-- 当前没有正在等待发送结果的发送项
send_item = nil
end
end
end
-- 订阅"SEND_DATA_REQ"消息;
-- 其他应用模块如果需要发送数据,直接sys.publish这个消息即可,将需要发送的数据以及回调函数和回调参数一起publish出去;
-- 本demo项目中uart_app.lua和timer_app.lua中publish了这个消息;
sys.subscribe("SEND_DATA_REQ", send_data_req_proc_func)
--创建并且启动一个task
--运行这个task的处理函数mqtts_m_ca_client_sender_task_func
sys.taskInitEx(mqtts_m_ca_client_sender_task_func, mqtts_m_ca_sender.TASK_NAME)
return mqtts_m_ca_sender
@@ -0,0 +1,31 @@
-----BEGIN CERTIFICATE-----
MIIFazCCA1OgAwIBAgIRAIIQz7DSQONZRGPgu2OCiwAwDQYJKoZIhvcNAQELBQAw
TzELMAkGA1UEBhMCVVMxKTAnBgNVBAoTIEludGVybmV0IFNlY3VyaXR5IFJlc2Vh
cmNoIEdyb3VwMRUwEwYDVQQDEwxJU1JHIFJvb3QgWDEwHhcNMTUwNjA0MTEwNDM4
WhcNMzUwNjA0MTEwNDM4WjBPMQswCQYDVQQGEwJVUzEpMCcGA1UEChMgSW50ZXJu
ZXQgU2VjdXJpdHkgUmVzZWFyY2ggR3JvdXAxFTATBgNVBAMTDElTUkcgUm9vdCBY
MTCCAiIwDQYJKoZIhvcNAQEBBQADggIPADCCAgoCggIBAK3oJHP0FDfzm54rVygc
h77ct984kIxuPOZXoHj3dcKi/vVqbvYATyjb3miGbESTtrFj/RQSa78f0uoxmyF+
0TM8ukj13Xnfs7j/EvEhmkvBioZxaUpmZmyPfjxwv60pIgbz5MDmgK7iS4+3mX6U
A5/TR5d8mUgjU+g4rk8Kb4Mu0UlXjIB0ttov0DiNewNwIRt18jA8+o+u3dpjq+sW
T8KOEUt+zwvo/7V3LvSye0rgTBIlDHCNAymg4VMk7BPZ7hm/ELNKjD+Jo2FR3qyH
B5T0Y3HsLuJvW5iB4YlcNHlsdu87kGJ55tukmi8mxdAQ4Q7e2RCOFvu396j3x+UC
B5iPNgiV5+I3lg02dZ77DnKxHZu8A/lJBdiB3QW0KtZB6awBdpUKD9jf1b0SHzUv
KBds0pjBqAlkd25HN7rOrFleaJ1/ctaJxQZBKT5ZPt0m9STJEadao0xAH0ahmbWn
OlFuhjuefXKnEgV4We0+UXgVCwOPjdAvBbI+e0ocS3MFEvzG6uBQE3xDk3SzynTn
jh8BCNAw1FtxNrQHusEwMFxIt4I7mKZ9YIqioymCzLq9gwQbooMDQaHWBfEbwrbw
qHyGO0aoSCqI3Haadr8faqU9GY/rOPNk3sgrDQoo//fb4hVC1CLQJ13hef4Y53CI
rU7m2Ys6xt0nUW7/vGT1M0NPAgMBAAGjQjBAMA4GA1UdDwEB/wQEAwIBBjAPBgNV
HRMBAf8EBTADAQH/MB0GA1UdDgQWBBR5tFnme7bl5AFzgAiIyBpY9umbbjANBgkq
hkiG9w0BAQsFAAOCAgEAVR9YqbyyqFDQDLHYGmkgJykIrGF1XIpu+ILlaS/V9lZL
ubhzEFnTIZd+50xx+7LSYK05qAvqFyFWhfFQDlnrzuBZ6brJFe+GnY+EgPbk6ZGQ
3BebYhtF8GaV0nxvwuo77x/Py9auJ/GpsMiu/X1+mvoiBOv/2X/qkSsisRcOj/KK
NFtY2PwByVS5uCbMiogziUwthDyC3+6WVwW6LLv3xLfHTjuCvjHIInNzktHCgKQ5
ORAzI4JMPJ+GslWYHb4phowim57iaztXOoJwTdwJx4nLCgdNbOhdjsnvzqvHu7Ur
TkXWStAmzOVyyghqpZXjFaH3pO3JLF+l+/+sKAIuvtd7u+Nxe5AW0wdeRlN8NwdC
jNPElpzVmbUq4JUagEiuTDkHzsxHpFKVK7q4+63SM1N95R1NbdWhscdCb+ZAJzVc
oyi3B43njTOQ5yOf+1CceWxG1bQVs5ZufpsMljq4Ui0/1lvh+wjChP4kqKOJ2qxq
4RgqsahDYVvTH9w7jXbyLeiNdd8XM2w9U/t7y0Ff/9yi0GE44Za4rF2LN9d11TPA
mRGunUHBcnWEvgJBQl9nJEiU0Zsnvgc/ubhPgXRR4Xq37Z0j4r7g1SgEEzwxA57d
emyPxgcYxn/eR44/KJ4EBs+lVDR3veyJm+kXQ99b21/+jh5Xos1AnX5iItreGCc=
-----END CERTIFICATE-----
@@ -0,0 +1,65 @@
--[[
@module sntp_app
@summary sntp时间同步应用功能模块
@version 1.0
@date 2025.07.01
@author 马梦阳
@usage
本文件为sntp时间同步应用功能模块,核心业务逻辑为:
1、连接ntp服务器进行时间同步;
2、如果同步成功,1小时之后重新发起同步动作;
3、如果同步失败,10秒钟之后重新发起同步动作;
本文件没有对外接口,直接在其他应用功能模块中require "sntp_app"就可以加载运行;
]]
-- sntp时间同步的任务处理函数
local function sntp_task_func()
while true do
-- 如果当前时间点设置的默认网卡还没有连接成功,一直在这里循环等待
while not socket.adapter(socket.dft()) do
log.warn("sntp_task_func", "wait IP_READY", socket.dft())
-- 在此处阻塞等待默认网卡连接成功的消息"IP_READY"
-- 或者等待1秒超时退出阻塞等待状态;
-- 注意:此处的1000毫秒超时不要修改的更长;
-- 因为当使用exnetif.set_priority_order配置多个网卡连接外网的优先级时,会隐式的修改默认使用的网卡
-- 当exnetif.set_priority_order的调用时序和此处的socket.adapter(socket.dft())判断时序有可能不匹配
-- 此处的1秒,能够保证,即使时序不匹配,也能1秒钟退出阻塞状态,再去判断socket.adapter(socket.dft())
sys.waitUntil("IP_READY", 1000)
end
-- 检测到了IP_READY消息
log.warn("sntp_task_func", "recv IP_READY")
-- 发起ntp时间同步动作
socket.sntp()
-- 等待ntp时间同步结果,30秒超时失败,通常只需要几百毫秒就能成功
local ret = sys.waitUntil("NTP_UPDATE", 30000)
--同步成功
if ret then
-- 以下是获取/打印时间的演示,注意时区问题
log.info("sntp_task_func", "时间同步成功", "本地时间", os.date())
log.info("sntp_task_func", "时间同步成功", "UTC时间", os.date("!%c"))
log.info("sntp_task_func", "时间同步成功", "RTC时钟(UTC时间)", json.encode(rtc.get()))
log.info("sntp_task_func", "时间同步成功", "本地时间戳", os.time())
local t = os.date("*t")
log.info("sntp_task_func", "时间同步成功", "本地时间os.date() json格式", json.encode(t))
log.info("sntp_task_func", "时间同步成功", "本地时间os.date(os.time())", os.time(t))
-- 正常使用, 一小时一次, 已经足够了, 甚至1天一次也可以
sys.wait(3600000)
--同步失败
else
log.info("sntp_task_func", "时间同步失败")
-- 10秒后重新发起同步动作
sys.wait(10000)
end
end
end
--创建并且启动一个task
--运行这个task的主函数sntp_task_func
sys.taskInit(sntp_task_func)
@@ -0,0 +1,43 @@
--[[
@module netdrv_4g
@summary “4G网卡”驱动模块
@version 1.0
@date 2025.07.01
@author 马梦阳
@usage
本文件为4G网卡驱动模块,核心业务逻辑为:
1、监听"IP_READY"和"IP_LOSE",在日志中进行打印;
本文件没有对外接口,直接在其他功能模块中require "netdrv_4g"就可以加载运行;
]]
local function ip_ready_func(ip, adapter)
if adapter == socket.LWIP_GP then
-- 在位置1和2设置自定义的DNS服务器ip地址:
-- "223.5.5.5",这个DNS服务器IP地址是阿里云提供的DNS服务器IP地址;
-- "114.114.114.114",这个DNS服务器IP地址是国内通用的DNS服务器IP地址;
-- 可以加上以下两行代码,在自动获取的DNS服务器工作不稳定的情况下,这两个新增的DNS服务器会使DNS服务更加稳定可靠;
-- 如果使用专网卡,不要使用这两行代码;
-- 如果使用国外的网络,不要使用这两行代码;
socket.setDNS(adapter, 1, "223.5.5.5")
socket.setDNS(adapter, 2, "114.114.114.114")
log.info("netdrv_4g.ip_ready_func", "IP_READY", socket.localIP(socket.LWIP_GP))
end
end
local function ip_lose_func(adapter)
if adapter == socket.LWIP_GP then
log.warn("netdrv_4g.ip_lose_func", "IP_LOSE")
end
end
-- 此处订阅"IP_READY"和"IP_LOSE"两种消息
-- 在消息的处理函数中,仅仅打印了一些信息,便于实时观察4G网络的连接状态
-- 也可以根据自己的项目需求,在消息处理函数中增加自己的业务逻辑控制,例如可以在连网状态发生改变时更新网络图标
sys.subscribe("IP_READY", ip_ready_func)
sys.subscribe("IP_LOSE", ip_lose_func)
-- 在Air780EPM上,内核固件运行起来之后,默认网卡就是socket.LWIP_GP
@@ -0,0 +1,77 @@
--[[
@module netdrv_eth_spi
@summary “通过SPI外挂CH390H芯片的以太网卡”驱动模块
@version 1.0
@date 2025.07.24
@author 马梦阳
@usage
本文件为“通过SPI外挂CH390H芯片的以太网卡”驱动模块,核心业务逻辑为:
1、打开CH390H芯片供电开关;
2、初始化spi0,初始化以太网卡,并且在以太网卡上开启DHCP(动态主机配置协议);
3、以太网卡的连接状态发生变化时,在日志中进行打印;
直接使用Air780EPM V1.3版本开发板硬件测试即可;
本文件没有对外接口,直接在其他功能模块中require "netdrv_eth_spi"就可以加载运行;
]]
local exnetif = require "exnetif"
local function ip_ready_func(ip, adapter)
if adapter == socket.LWIP_ETH then
-- 在位置1和2设置自定义的DNS服务器ip地址:
-- "223.5.5.5",这个DNS服务器IP地址是阿里云提供的DNS服务器IP地址;
-- "114.114.114.114",这个DNS服务器IP地址是国内通用的DNS服务器IP地址;
-- 可以加上以下两行代码,在自动获取的DNS服务器工作不稳定的情况下,这两个新增的DNS服务器会使DNS服务更加稳定可靠;
-- 如果使用专网卡,不要使用这两行代码;
-- 如果使用国外的网络,不要使用这两行代码;
socket.setDNS(adapter, 1, "223.5.5.5")
socket.setDNS(adapter, 2, "114.114.114.114")
log.info("netdrv_eth_spi.ip_ready_func", "IP_READY", socket.localIP(socket.LWIP_ETH))
end
end
local function ip_lose_func(adapter)
if adapter == socket.LWIP_ETH then
log.warn("netdrv_eth_spi.ip_lose_func", "IP_LOSE")
end
end
-- 以太网联网成功(成功连接路由器,并且获取到了IP地址)后,内核固件会产生一个"IP_READY"消息
-- 各个功能模块可以订阅"IP_READY"消息实时处理以太网联网成功的事件
-- 也可以在任何时刻调用socket.adapter(socket.LWIP_ETH)来获取以太网是否连接成功
-- 以太网断网后,内核固件会产生一个"IP_LOSE"消息
-- 各个功能模块可以订阅"IP_LOSE"消息实时处理以太网断网的事件
-- 也可以在任何时刻调用socket.adapter(socket.LWIP_ETH)来获取以太网是否连接成功
--此处订阅"IP_READY"和"IP_LOSE"两种消息
--在消息的处理函数中,仅仅打印了一些信息,便于实时观察“通过SPI外挂CH390H芯片的以太网卡”的连接状态
--也可以根据自己的项目需求,在消息处理函数中增加自己的业务逻辑控制,例如可以在连网状态发生改变时更新网络图标
sys.subscribe("IP_READY", ip_ready_func)
sys.subscribe("IP_LOSE", ip_lose_func)
local function netdrv_eth_spi_task_func()
-- 配置SPI外接以太网芯片CH390H的单网卡,exnetif.set_priority_order使用的网卡编号为socket.LWIP_ETH
-- 本demo使用Air780EPM V1.3版本开发板测试,开发板上的硬件配置为:
-- GPIO20为CH390H以太网芯片的供电使能控制引脚
-- 使用spi0,片选引脚使用GPIO8,中断引脚使用GPIO22
-- 如果使用的硬件和以上描述的环境不同,根据自己的硬件配置修改以下参数
exnetif.set_priority_order({
{
ETHERNET = {
pwrpin = 20,
tp = netdrv.CH390,
opts = {spi = 0, cs = 8, irq = 22}
}
}
})
end
-- 启动一个tasktask的处理函数为netdrv_eth_spi_task_func
-- 在处理函数中调用exnetif.set_priority_order设置网卡优先级
-- 因为exnetif.set_priority_order要求必须在task中被调用,所以此处启动一个task
sys.taskInit(netdrv_eth_spi_task_func)
@@ -0,0 +1,88 @@
--[[
@module netdrv_multiple
@summary 多网卡(4G网卡、通过SPI外挂CH390H芯片的以太网卡)驱动模块
@version 1.0
@date 2025.07.24
@author 马梦阳
@usage
本文件为多网卡驱动模块,核心业务逻辑为:
1、调用exnetif.set_priority_order配置多网卡的控制参数以及优先级;
直接使用Air780EPM V1.3版本开发板硬件测试即可;
本文件没有对外接口,直接在其他功能模块中require "netdrv_multiple"就可以加载运行;
]]
local exnetif = require "exnetif"
-- 网卡状态变化通知回调函数
-- 当exnetif中检测到网卡切换或者所有网卡都断网时,会触发调用此回调函数
-- 当网卡切换切换时:
-- net_typestring类型,表示当前使用的网卡字符串
-- adapternumber类型,表示当前使用的网卡id
-- 当所有网卡断网时:
-- net_type:为nil
-- adapternumber类型,为-1
local function netdrv_multiple_notify_cbfunc(net_type,adapter)
-- 在位置1和2设置自定义的DNS服务器ip地址:
-- "223.5.5.5",这个DNS服务器IP地址是阿里云提供的DNS服务器IP地址;
-- "114.114.114.114",这个DNS服务器IP地址是国内通用的DNS服务器IP地址;
-- 可以加上以下两行代码,在自动获取的DNS服务器工作不稳定的情况下,这两个新增的DNS服务器会使DNS服务更加稳定可靠;
-- 如果使用专网卡,不要使用这两行代码;
-- 如果使用国外的网络,不要使用这两行代码;
socket.setDNS(adapter, 1, "223.5.5.5")
socket.setDNS(adapter, 2, "114.114.114.114")
if type(net_type)=="string" then
log.info("netdrv_multiple_notify_cbfunc", "use new adapter", net_type, adapter)
elseif type(net_type)=="nil" then
log.warn("netdrv_multiple_notify_cbfunc", "no available adapter", net_type, adapter)
else
log.warn("netdrv_multiple_notify_cbfunc", "unknown status", net_type, adapter)
end
end
local function netdrv_multiple_task_func()
--设置网卡优先级
exnetif.set_priority_order(
{
-- “通过SPI外挂CH390H芯片”的以太网卡,使用Air780EPM V1.3版本开发板验证
{
ETHERNET = {
-- 供电使能GPIO
pwrpin = 20,
-- 设置的多个“已经IP READY,但是还没有ping通”网卡,循环执行ping动作的间隔(单位毫秒,可选)
-- 如果没有传入此参数,exnetif会使用默认值10秒
ping_time = 3000,
-- 连通性检测ip(选填参数)
-- 如果没有传入ip地址,exnetif中会默认使用httpdns能否成功获取baidu.com的ip作为是否连通的判断条件;
-- 如果传入,一定要传入可靠的并且可以ping通的ip地址;
-- ping_ip = "填入可靠的并且可以ping通的ip地址",
-- 网卡芯片型号(选填参数),仅spi方式外挂以太网时需要填写。
tp = netdrv.CH390,
opts = {spi=0, cs=8, irq=22}
}
},
-- 4G网卡
{
LWIP_GP = true
}
}
)
end
-- 设置网卡状态变化通知回调函数netdrv_multiple_notify_cbfunc
exnetif.notify_status(netdrv_multiple_notify_cbfunc)
-- 如果存在udp网络应用,并且udp网络应用中,根据应用层的心跳能够判断出来udp数据通信出现了异常;
-- 可以在判断出现异常的位置,调用一次exnetif.check_network_status()接口,强制对当前正式使用的网卡进行一次连通性检测;
-- 如果存在tcp网络应用,不需要用户调用exnetif.check_network_status()接口去控制,exnetif会在tcp网络应用通信异常时自动对当前使用的网卡进行连通性检测。
-- 启动一个tasktask的处理函数为netdrv_multiple_task_func
-- 在处理函数中调用exnetif.set_priority_order设置网卡优先级
-- 因为exnetif.set_priority_order要求必须在task中被调用,所以此处启动一个task
sys.taskInit(netdrv_multiple_task_func)
@@ -0,0 +1,57 @@
--[[
@module netdrv_pc
@summary “pc模拟器网卡”驱动模块
@version 1.0
@date 2025.07.01
@author 朱天华
@usage
本文件为pc模拟器网卡驱动模块,核心业务逻辑为:
1、监听"IP_READY"和"IP_LOSE",在日志中进行打印;
本文件没有对外接口,直接在其他功能模块中require "netdrv_pc"就可以加载运行;
]]
local function ip_ready_func(ip, adapter)
if adapter == socket.ETH0 then
-- 在位置1和2设置自定义的DNS服务器ip地址:
-- "223.5.5.5",这个DNS服务器IP地址是阿里云提供的DNS服务器IP地址;
-- "114.114.114.114",这个DNS服务器IP地址是国内通用的DNS服务器IP地址;
-- 可以加上以下两行代码,在自动获取的DNS服务器工作不稳定的情况下,这两个新增的DNS服务器会使DNS服务更加稳定可靠;
-- 如果使用专网卡,不要使用这两行代码;
-- 如果使用国外的网络,不要使用这两行代码;
socket.setDNS(adapter, 1, "223.5.5.5")
socket.setDNS(adapter, 2, "114.114.114.114")
log.info("netdrv_pc.ip_ready_func", "IP_READY", socket.localIP(socket.ETH0))
end
end
local function ip_lose_func(adapter)
if adapter == socket.ETH0 then
log.warn("netdrv_pc.ip_lose_func", "IP_LOSE")
end
end
--此处订阅"IP_READY"和"IP_LOSE"两种消息
--在消息的处理函数中,仅仅打印了一些信息,便于实时观察pc模拟器网络的连接状态
--也可以根据自己的项目需求,在消息处理函数中增加自己的业务逻辑控制,例如可以在连网状态发生改变时更新网络图标
sys.subscribe("IP_READY", ip_ready_func)
sys.subscribe("IP_LOSE", ip_lose_func)
-- 设置默认网卡为socket.ETH0
-- pc模拟器上的默认网卡仍然需要使用接口(socket.ETH0)来设置,因为exnetif扩展库当前还不支持模拟器
socket.dft(socket.ETH0)
-- 下面这段代码是在PC模拟器上构造一个唯一的ID;不同PC上运行模拟器,这个ID要不一样
-- 因为mqtt client使用的是这个ID做为client id,如果不同PC上的id一样,模拟器在不同PC上同时运行时,mqtt client就会频繁出现被踢下线的问题
-- 目前模拟器上还没有合适的接口获取唯一ID,所以此处先简单的构造一个ID,需要手动保证ID唯一,在此处我简单使用了zhutianhua1做为ID
_G.mobile = {}
function mobile.imei()
return "zhutianhua1"
-- log.info("mcu.unique_id()", mcu.unique_id())
-- return mcu.unique_id().."zhutianhua1"
end
@@ -0,0 +1,29 @@
--[[
@module netdrv_device
@summary 网络驱动设备功能模块
@version 1.0
@date 2025.07.24
@author 马梦阳
@usage
本文件为网络驱动设备功能模块,核心业务逻辑为:根据项目需求,选择并且配置合适的网卡(网络适配器)
1、netdrv_4gsocket.LWIP_GP4G网卡;
2、netdrv_eth_spisocket.LWIP_ETH,通过SPI外挂CH390H芯片的以太网卡;
3、netdrv_multiple:可以配置多种网卡的优先级,按照优先级配置,使用其中一种网卡连接外网;
根据自己的项目需求,只需要require以上三种中的一种即可;
本文件没有对外接口,直接在main.lua中require "netdrv_device"就可以加载运行;
]]
-- 根据自己的项目需求,只需要require以下三种中的一种即可;
-- 加载“4G网卡”驱动模块
require "netdrv_4g"
-- 加载“通过SPI外挂CH390H芯片的以太网卡”驱动模块
-- require "netdrv_eth_spi"
-- 加载“可以配置优先级的多种网卡”驱动模块
-- require "netdrv_multiple"
@@ -0,0 +1,63 @@
--[[
@module network_watchdog
@summary 网络环境检测看门狗功能模块
@version 1.0
@date 2025.07.23
@author 马梦阳
@usage
本文件为网络环境检测看门狗功能模块,监控网络环境是否工作正常(设备和服务器双向通信正常,或者至少单向通信正常),核心业务逻辑为:
1、启动一个网络环境检测看门狗task,等待其他mqtt网络应用功能模块来喂狗,如果喂狗超时,则控制软件重启;
2、如何确定“喂狗超时时间”,一般来说,有以下几个原则;
(1) 先确定一个最小基准值T1,2分钟或者5分钟或者10分钟,这个取值取决于具体项目需求,但是不能太短,因为开机后,在网络环境不太好的地方,网络初始化可能需要比较长的时间,一般推荐这个值不能小于2分钟;
(2) 再确定一个和产品业务逻辑有关的一个值T2,这个值和产品的应用业务逻辑息息相关,假设你的产品业务中:
<1> 服务器会定时下发数据给设备,例如设备连接上业务服务器之后,每隔3分钟,设备都会给服务器发送一次心跳,然后服务器都会立即回复一个心跳应答包;
这种情况下,可以取3分钟的大于等于1的倍数(例如1倍,1.5倍,2倍等等)+一段时间(例如10秒钟,如果前面是1倍,则此处必须加一段时间,给网络数据传输过程留够充足的时间);
<2> mqtt本身有keep alive的心跳机制,例如设备连接上业务服务器之后,默认每隔2分钟,设备都会给服务器发送一次心跳,服务器也会回复一个心跳应答数据;
这种情况下,可以取2分钟的大于等于1的倍数(例如1倍,1.5倍,2倍等等)+一段时间(例如10秒钟,如果前面是1倍,则此处必须加一段时间,给网络数据传输过程留够充足的时间);
(3) 取T1和T2的最大值,就是“喂狗超时时间”
3、其他mqtt网络业务功能模块的喂狗时机,和上面2.2的描述相对应,一般来说,可以在以下几种时间点执行喂狗动作:
(1) 设备收到服务器下发的数据时
(2) 设备收到服务器回复的mqtt心跳应答数据时
4、最重要的一点是:以上所说的原则,仅仅是建议,要根据自己的实际项目业务逻辑以及自己的需求最终确定看门狗方案
5、具体到本demo
(1) 产品业务逻辑为:
<1> 创建了一个mqtt连接,设备每隔5秒钟发送一次数据到服务器,服务器何时下发应用数据给设备不确定;
<2> 创建了一个mqtt ssl、不需要证书校验的连接,设备每隔5秒钟发送一次数据到服务器,服务器何时下发应用数据给设备不确定;
<3> 创建了一个mqtt ssl、client单向校验server证书的连接,设备每隔5秒钟发送一次数据到服务器,服务器何时下发应用数据给设备不确定;
<4> 每隔3分钟,这三路mqtt连接都会发送一次mqtt心跳给server,server收到心跳后回复心跳应答给client;
(2) 确定喂狗超时时间:
<1> 本demo支持单WIFI、单以太网、单4G网络连接外网,网络环境准备就绪预留2分钟的时间已经足够,所以最小基准值T1取值2分钟;
<2> 本demo中存在3路mqtt连接,但是这3路mqtt连接都没有定时或者至少一段时间,服务器下发应用数据给设备,所以无法基于服务器下发应用数据的业务逻辑来确定T2的值;
<3> 本demo中存在3路mqtt连接,每1路mqtt连接,设备都是3分钟发送一次mqtt心跳数据给服务器,服务器收到后会立即回复一个mqtt心跳应答数据给设备;
所以可以通过3分钟的大于等于1的倍数(例如1倍,1.5倍,2倍等等)+一段时间(例如10秒钟,如果前面是1倍,则此处必须加一段时间,给网络数据传输过程留够充足的时间)来确定T2的值;
在这个demo中,我能接受的网络连续异常时长是5分钟,所以,T2取值5分钟;
<4> 取T1 2分钟和T2 5分钟的最大值,最终的喂狗超时时间就是5分钟;
(3) 确定喂狗时机:
<1> 3路mqtt连接中,任何1路收到服务器的下发的应用数据时;
<2> 3路mqtt连接中,任何1路收到服务器的回复的心跳应答数据时;
6、本demo设计的网络环境检测看门狗功能模块,可以检测以下两种种的任意一种网络环境异常:
(1) 网络环境连续超过5分钟没有准备就绪
(2) mqtt、mqtt ssl、mqtt ssl单向校验证书3路连接中,连续5分钟没有收到服务器下发的应用数据或者服务器回复的心跳应答数据;
本文件没有对外接口,直接在main.lua中require "network_watchdog"就可以加载运行;
外部功能模块喂狗时,直接调用sys.publish("FEED_NETWORK_WATCHDOG")
]]
-- 网络环境检测看门狗task处理函数
local function network_watchdog_task_func()
while true do
--如果等待180秒没有等到"FEED_NETWORK_WATCHDOG"消息,则看门狗超时
if not sys.waitUntil("FEED_NETWORK_WATCHDOG", 300000) then
log.error("network_watchdog_task_func timeout")
-- 等待3秒钟,然后软件重启
sys.wait(3000)
rtos.reboot()
end
end
end
--创建并且启动一个task
--运行这个task的处理函数network_watchdog_task_func
sys.taskInit(network_watchdog_task_func)
@@ -0,0 +1,146 @@
## 功能模块介绍
1、main.lua:主程序入口;
2、netdrv_device.lua:网卡驱动设备,可以配置使用netdrv文件夹内的三种网卡(单4g网卡,单spi以太网卡,多网卡)中的任何一种网卡;
3、mqtt文件夹:mqtt client连接以及数据收发处理逻辑;
4、mqtts文件夹:mqtt ssl client(不支持证书校验)连接以及数据收发处理逻辑;
5、mqtts_ca文件夹:mqtt ssl client(仅单向校验server端证书)连接以及数据收发处理逻辑;
6、mqtts_mutual_ca文件夹:mqtt ssl client(双向校验证书)连接以及数据收发处理逻辑;
7、network_watchdog.lua:网络环境检测看门狗;
8、timer_app.lua:通知四个mqtt client定时发送数据到服务器;
9、uart_app.lua:在四个mqtt client和uart外设之间透传数据;
## 系统消息介绍
1、"IP_READY":某种网卡已经获取到ip信息,仅仅获取到了ip信息,能否和外网连通还不确认;
2、"IP_LOSE":某种网卡已经掉网;
## 用户消息介绍
1、"RECV_DATA_FROM_SERVER"mqtt client收到服务器下发的publish数据后,通过此消息发布出去,给其他应用模块(uart_app)处理;
2、"SEND_DATA_REQ":其他应用模块(uart_apptimer_app)发布此消息,通知mqtt client发送publish数据给服务器;
3、"FEED_NETWORK_WATCHDOG":网络环境检测看门狗的喂狗消息,在需要喂狗的地方发布此消息;
## 演示功能概述
1、创建四路mqtt连接,详情如下
> 注意:代码中的mqtt服务器地址和端口会不定期重启或维护,仅能用作测试用途,不可商用,说不定哪一天就关闭了。用户开发项目时,需要替换为自己的商用服务器地址和端口。
- 创建一个mqtt client,连接mqtt server
- 创建一个mqtt ssl client,连接mqtt ssl server,不做证书校验;
- 创建一个mqtt ssl client,连接mqtt ssl serverclient仅单向校验server的证书,server不校验client的证书和密钥文件;
- 创建一个mqtt ssl client,连接mqtt ssl serverclient校验server的证书,server校验client的证书和密钥文件;
2、每一路mqtt连接出现异常后,自动重连;
3、每一路mqtt连接,client按照以下几种逻辑发送数据给server
- 串口应用功能模块uart_app.lua,通过uart1接收到串口数据,将串口数据增加send from uart: 前缀后,使用mobile.imei().."/uart/up"主题,发送给server
- 定时器应用功能模块timer_app.lua,定时产生数据,将数据增加send from timer:前缀后,使用mobile.imei().."/timer/up"主题,发送给server
4、每一路mqtt连接,client收到server数据后,将数据增加recv from mqtt/mqtt ssl/mqtt ssl ca/mqtt ssl mutual ca(四选一)server: 前缀后,通过uart1发送出去;
5、启动一个网络业务逻辑看门狗task,用来监控网络环境,如果连续长时间工作不正常,重启整个软件系统;
6、netdrv_device:配置连接外网使用的网卡,目前支持以下三种选择(三选一)
(1) netdrv_4g4G网卡
(2) netdrv_eth_spi:通过SPI外挂CH390H芯片的以太网卡
(3) netdrv_multiple:支持以上两种网卡,可以配置两种网卡的优先级
## 演示硬件环境
![](https://docs.openluat.com/air780epm/luatos/app/driver/eth/image/RFSvb75NRoEWqYxfCRVcVrOKnsf.jpg)
1、Air780EPM V1.3版本开发板一块+可上网的sim卡一张+4g天线一根+网线一根:
- sim卡插入开发板的sim卡槽
- 天线装到开发板上
- 网线一端插入开发板网口,另外一端连接可以上外网的路由器网口
2、TYPE-C USB数据线一根 + USB转串口数据线一根,Air780EPM V1.3版本开发板和数据线的硬件接线方式为:
- Air780EPM V1.3版本开发板通过TYPE-C USB口供电;(外部供电/USB供电 拨动开关 拨到 USB供电一端)
- TYPE-C USB数据线直接插到核心板的TYPE-C USB座子,另外一端连接电脑USB口;
- USB转串口数据线,一般来说,白线连接开发板的UART1_TX,绿线连接开发板的UART1_RX,黑线连接核心板的GND,另外一端连接电脑USB口;
## 演示软件环境
1、Luatools下载调试工具
2、[Air780EPM V2012版本固件](https://docs.openluat.com/air780epm/luatos/firmware/version/)(理论上,2025年8月10日之后发布的固件都可以)
3、PC端的串口工具,例如SSCOM、LLCOM等都可以
4、[MQTT客户端软件MQTTX](https://docs.openluat.com/air780epm/luatos/common/swenv/#254-mqttmqttx)
## 演示核心步骤
1、搭建好硬件环境
2、demo脚本代码netdrv_device.lua中,按照自己的网卡需求启用对应的Lua文件
- 如果需要单4G网卡,打开require "netdrv_4g",其余注释掉
- 如果需要以太网卡,打开require "netdrv_eth_spi",其余注释掉
- 如果需要多网卡,打开require "netdrv_multiple",其余注释掉
3、Luatools烧录内核固件和修改后的demo脚本代码
4、烧录成功后,自动开机运行,如果出现以下日志,表示四路mqtt连接成功
``` lua
I/user.mqtt_client_main_task_func connect success
I/user.mqtts_client_main_task_func connect success
I/user.mqtts_ca_client_main_task_func connect success
I/user.mqtts_m_ca_client_main_task_func connect success
```
5、启动三个MQTTX工具,分别连接上mqtt client、mqtts client、mqtts mutual client对应的server,订阅$imei/timer/up$主题,imei表示设备的imei号;可以看到,每隔5秒钟,会接收到一段类似于 send from timer: 1 的数据,最后面的数字每次加1;
6、打开PC端的串口工具,选择对应的端口,配置波特率115200,数据位8,停止位1,无奇偶校验位;
7、PC端的串口工具输入一段数据,点击发送,在MQTTX工具订阅的$imei/uart/up$主题下可以接收到数据;
8、在MQTTX工具上,在主题$imei/down$下publish一段数据,点击发送,在PC端的串口工具上可以接收到主题和数据,并且也能看到是哪一个server发送的,类似于以下效果:
``` lua
recv from mqtt server: 864793080144269/down,123456798012345678901234567830
recv from mqtt ssl server: 864793080144269/down,123456798012345678901234567830
recv from mqtt ssl ca server: 864793080144269/down,123456798012345678901234567830
recv from mqtt ssl mutual ca server: 864793080144269/down,123456798012345678901234567830
```
@@ -0,0 +1,44 @@
--[[
@module timer_app
@summary 定时器应用功能模块
@version 1.0
@date 2025.07.01
@author 马梦阳
@usage
本文件为定时器应用功能模块,核心业务逻辑为:
创建一个5秒的循环定时器,每次产生一段数据,通知四个mqtt client进行处理;
本文件的对外接口有一个:
1、sys.publish("SEND_DATA_REQ", "timer", mobile.imei().."/timer/up", payload, 0, {func=send_data_cbfunc, para="timer"..payload})
通过publish通知四路mqtt client数据发送功能模块publish数据;
数据发送结果通过执行回调函数send_data_cbfunc通知本功能模块;
]]
local payload = 1
-- 数据发送结果回调函数
-- result:发送结果,true为发送成功,false为发送失败
-- para:回调参数,sys.publish("SEND_DATA_REQ", "timer", mobile.imei().."/timer/up", payload, 0, {func=send_data_cbfunc, para="timer"..payload})中携带的para
local function send_data_cbfunc(result, para)
log.info("send_data_cbfunc", result, para)
-- 无论上一次发送成功还是失败,启动一个5秒的定时器,5秒后发送下次数据
sys.timerStart(send_data_req_timer_cbfunc, 5000)
end
-- 定时器回调函数
function send_data_req_timer_cbfunc()
-- 发布消息"SEND_DATA_REQ"
-- 携带的第一个参数"timer"表示是定时器应用模块发布的消息
-- 携带的第二个参数mobile.imei().."/timer/up"为要publish的topic
-- 携带的第三个参数payload为要publish的payload
-- 携带的第四个参数0为publish的qos
-- 携带的第五个参数cb为发送结果回调(可以为空,如果为空,表示不关心mqtt client发送数据成功还是失败),其中:
-- cb.func为回调函数(可以为空,如果为空,表示不关心mqtt client发送数据成功还是失败)
-- cb.para为回调函数的第二个参数(可以为空),回调函数的第一个参数为发送结果(true表示成功,false表示失败)
sys.publish("SEND_DATA_REQ", "timer", mobile.imei().."/timer/up", payload, 0, {func=send_data_cbfunc, para="timer"..payload})
payload = payload+1
end
-- 启动一个5秒的单次定时器
-- 时间到达后,执行一次send_data_req_timer_cbfunc函数
sys.timerStart(send_data_req_timer_cbfunc, 5000)
@@ -0,0 +1,79 @@
--[[
@module uart_app
@summary 串口应用功能模块
@version 1.0
@date 2025.07.28
@author 马梦阳
@usage
本文件为串口应用功能模块,核心业务逻辑为:
1、打开uart1,波特率115200,数据位8,停止位1,无奇偶校验位;
2、uart1和pc端的串口工具相连;
3、从uart1接收到pc端串口工具发送的数据后,通知四个mqtt client进行处理;
4、收到四个mqtt client从socket server接收到的数据后,将数据通过uart1发送到pc端串口工具;
本文件的对外接口有两个:
1、sys.publish("SEND_DATA_REQ", "uart", mobile.imei().."/uart/up", read_buf, 1)
通知mqtt client数据发送模块,在mobile.imei().."/uart/up"的topic上publish数据read_buf,不关心数据发送成功还是失败;
2、sys.subscribe("RECV_DATA_FROM_SERVER", recv_data_from_server_proc),订阅RECV_DATA_FROM_SERVER消息,处理消息携带的数据;
]]
-- 使用UART1
local UART_ID = 1
-- 串口接收数据缓冲区
local read_buf = ""
-- 将前缀prefix和topicpayload数据拼接
-- 然后末尾增加回车换行两个字符,通过uart发送出去,方便在PC端换行显示查看
local function recv_data_from_server_proc(prefix, topic, payload)
uart.write(UART_ID, prefix..topic..","..payload.."\r\n")
end
local function concat_timeout_func()
-- 如果存在尚未处理的串口缓冲区数据;
-- 将数据通过publish通知其他应用功能模块处理;
-- 然后清空本文件的串口缓冲区数据
if read_buf:len() > 0 then
sys.publish("SEND_DATA_REQ", "uart", mobile.imei().."/uart/up", read_buf, 1)
read_buf = ""
end
end
-- UART1的数据接收中断处理函数,UART1接收到数据时,会执行此函数
local function read()
local s
while true do
-- 非阻塞读取UART1接收到的数据,最长读取1024字节
s = uart.read(UART_ID, 1024)
-- 如果从串口没有读到数据
if not s or s:len() == 0 then
-- 启动50毫秒的定时器,如果50毫秒内没收到新的数据,则处理当前收到的所有数据
-- 这样处理是为了防止将一大包数据拆分成多个小包来处理
-- 例如pc端串口工具下发1100字节的数据,可能会产生将近20次的中断进入到read函数,才能读取完整
-- 此处的50毫秒可以根据自己项目的需求做适当修改,在满足整包拼接完整的前提下,时间越短,处理越及时
sys.timerStart(concat_timeout_func, 50)
-- 跳出循环,退出本函数
break
end
log.info("uart_app.read len", s:len())
-- log.info("uart_app.read", s)
-- 将本次从串口读到的数据拼接到串口缓冲区read_buf中
read_buf = read_buf..s
end
end
-- 初始化UART1,波特率115200,数据位8,停止位1
uart.setup(UART_ID, 115200, 8, 1)
-- 注册UART1的数据接收中断处理函数,UART1接收到数据时,会执行read函数
uart.on(UART_ID, "receive", read)
-- 订阅"RECV_DATA_FROM_SERVER"消息的处理函数recv_data_from_server_proc
-- 收到"RECV_DATA_FROM_SERVER"消息后,会执行函数recv_data_from_server_proc
sys.subscribe("RECV_DATA_FROM_SERVER", recv_data_from_server_proc)