feat: IoT MQTT 客户端实现 — NET_SSC_ENABLE=0 时启用
新增文件:
- iot_mqtt_srv.h/c: MQTT 客户端 (TCP→CONNECT→SUBSCRIBE→READY)
支持: dev_info_query, pwd_verify (其他命令框架预留)
上报: loop_data (传感器), heartbeat (心跳 60s)
断线重连: 指数退避 5s~60s
协议: DLD960 IoT MQTT V1.00 (dld960/{sn}/...)
修改:
- net_config.h: 新增 NET_IOT_ENABLE = !NET_SSC_ENABLE,
IoT 模式分配 1 TCP socket (MQTT client)
- net_srv.c: IoT 初始化 + socket 中断路由
- peripheral_main.c: IoT poll + sensor publish 调用
模式:
NET_SSC_ENABLE=1 → SCC + TCP JSON Server (原有行为)
NET_SSC_ENABLE=0 → IoT MQTT Client (新增)
This commit is contained in:
wangfq
@@ -0,0 +1,55 @@
|
|||||||
|
/**
|
||||||
|
******************************************************************************
|
||||||
|
* @file iot_mqtt_srv.h
|
||||||
|
* @author wangfq
|
||||||
|
* @version V1.0
|
||||||
|
* @date 2026-07-03
|
||||||
|
* @brief IoT MQTT 客户端 — DLD960 IoT 协议 (MQTT + JSON)
|
||||||
|
* 当 NET_SSC_ENABLE=0 时启用, 替代 SSC 模式
|
||||||
|
* 协议参考: docs/DLD960_IoT_MQTT协议.md
|
||||||
|
******************************************************************************
|
||||||
|
*/
|
||||||
|
#ifndef __IOT_MQTT_SRV_H__
|
||||||
|
#define __IOT_MQTT_SRV_H__
|
||||||
|
|
||||||
|
#include <stdint.h>
|
||||||
|
|
||||||
|
/*===========================================================================
|
||||||
|
* Protocol Constants
|
||||||
|
*===========================================================================*/
|
||||||
|
#define IOT_MQTT_KEEPALIVE_SEC 60 // MQTT keepalive (s)
|
||||||
|
#define IOT_MQTT_RECONNECT_MIN_MS 5000 // 最小重连间隔
|
||||||
|
#define IOT_MQTT_RECONNECT_MAX_MS 60000 // 最大重连间隔 (指数退避上限)
|
||||||
|
#define IOT_MQTT_HEARTBEAT_MS 60000 // 心跳上报周期 (60s)
|
||||||
|
#define IOT_MQTT_RECV_BUF_LEN 2048 // 接收缓冲区
|
||||||
|
#define IOT_MQTT_SEND_BUF_LEN 2048 // 发送缓冲区
|
||||||
|
#define IOT_MQTT_TOPIC_MAX_LEN 128 // Topic 最大长度
|
||||||
|
|
||||||
|
/*===========================================================================
|
||||||
|
* MQTT 连接状态
|
||||||
|
*===========================================================================*/
|
||||||
|
typedef enum {
|
||||||
|
IOT_STATE_DISCONNECTED = 0, // 未连接
|
||||||
|
IOT_STATE_TCP_CONNECTING, // TCP 连接中
|
||||||
|
IOT_STATE_TCP_CONNECTED, // TCP 已连接, 等待 MQTT CONNECT
|
||||||
|
IOT_STATE_MQTT_CONNECTING, // MQTT CONNECT 已发送, 等待 CONNACK
|
||||||
|
IOT_STATE_MQTT_CONNECTED, // MQTT 已连接, 等待 SUBACK
|
||||||
|
IOT_STATE_READY, // 完全就绪 (已订阅)
|
||||||
|
} IotMqttState;
|
||||||
|
|
||||||
|
/*===========================================================================
|
||||||
|
* Externs
|
||||||
|
*===========================================================================*/
|
||||||
|
extern uint8_t g_iot_socket; // MQTT TCP socket ID
|
||||||
|
extern IotMqttState g_iot_state; // MQTT 连接状态
|
||||||
|
extern uint8_t g_iot_msg_id; // 递增消息 ID
|
||||||
|
|
||||||
|
/*===========================================================================
|
||||||
|
* API
|
||||||
|
*===========================================================================*/
|
||||||
|
void iot_mqtt_init(void); // 初始化 MQTT 连接 (TCP→CONNECT→SUBSCRIBE)
|
||||||
|
void iot_mqtt_poll(void); // 主循环轮询 (状态机 + 心跳 + 重连)
|
||||||
|
void iot_mqtt_handle_sock_int(uint8_t socketid, uint8_t intstat); // Socket 中断处理
|
||||||
|
void iot_mqtt_publish_sensor(void); // 推送传感器数据到 MQTT
|
||||||
|
|
||||||
|
#endif /* __IOT_MQTT_SRV_H__ */
|
||||||
@@ -19,9 +19,14 @@ extern "C" {
|
|||||||
|
|
||||||
/*********************************************************************
|
/*********************************************************************
|
||||||
* Feature toggles (0 = disabled, 1 = enabled)
|
* Feature toggles (0 = disabled, 1 = enabled)
|
||||||
|
*
|
||||||
|
* 模式互斥:
|
||||||
|
* NET_SSC_ENABLE=1 → SSC 模式 (UDP+TCP client) + TCP JSON server
|
||||||
|
* NET_SSC_ENABLE=0 → IoT MQTT 模式 (MQTT client)
|
||||||
*/
|
*/
|
||||||
#define NET_SSC_ENABLE 0 /* SSC (UDP + TCP client) protocol support */
|
#define NET_SSC_ENABLE 0 /* SSC (UDP + TCP client) protocol support */
|
||||||
#define NET_JSON_ENABLE 1 /* TCP JSON protocol server (port 5960) */
|
#define NET_JSON_ENABLE 1 /* TCP JSON protocol server (port 5960) */
|
||||||
|
#define NET_IOT_ENABLE (!NET_SSC_ENABLE) /* IoT MQTT client (auto-derived) */
|
||||||
|
|
||||||
/*********************************************************************
|
/*********************************************************************
|
||||||
* socket configuration, IPRAW + UDP + TCP + TCP_LISTEN = number of sockets
|
* socket configuration, IPRAW + UDP + TCP + TCP_LISTEN = number of sockets
|
||||||
@@ -29,17 +34,19 @@ extern "C" {
|
|||||||
#define WCHNET_NUM_IPRAW 0 /* Number of IPRAW connections */
|
#define WCHNET_NUM_IPRAW 0 /* Number of IPRAW connections */
|
||||||
|
|
||||||
#if NET_SSC_ENABLE
|
#if NET_SSC_ENABLE
|
||||||
#define WCHNET_NUM_UDP 1 /* The number of UDP connections */
|
#define WCHNET_NUM_UDP 1 /* SSC mode: 1 UDP */
|
||||||
#else
|
#else
|
||||||
#define WCHNET_NUM_UDP 0
|
#define WCHNET_NUM_UDP 0 /* IoT mode: 0 UDP */
|
||||||
#endif
|
#endif
|
||||||
|
|
||||||
#if NET_JSON_ENABLE && NET_SSC_ENABLE
|
#if NET_JSON_ENABLE && NET_SSC_ENABLE
|
||||||
#define WCHNET_NUM_TCP 2 /* JSON listen+data(1) + SSC client(1) */
|
#define WCHNET_NUM_TCP 2 /* SSC+JSON: JSON(1) + SSC client(1) */
|
||||||
#elif NET_JSON_ENABLE
|
#elif NET_JSON_ENABLE
|
||||||
#define WCHNET_NUM_TCP 1 /* JSON listen+data(1) */
|
#define WCHNET_NUM_TCP 1 /* JSON only: listen+data(1) */
|
||||||
#elif NET_SSC_ENABLE
|
#elif NET_SSC_ENABLE
|
||||||
#define WCHNET_NUM_TCP 1 /* SSC client(1) */
|
#define WCHNET_NUM_TCP 1 /* SSC only: TCP client(1) */
|
||||||
|
#elif NET_IOT_ENABLE
|
||||||
|
#define WCHNET_NUM_TCP 1 /* IoT MQTT client(1) */
|
||||||
#else
|
#else
|
||||||
#define WCHNET_NUM_TCP 0
|
#define WCHNET_NUM_TCP 0
|
||||||
#endif
|
#endif
|
||||||
|
|||||||
@@ -0,0 +1,630 @@
|
|||||||
|
/**
|
||||||
|
******************************************************************************
|
||||||
|
* @file iot_mqtt_srv.c
|
||||||
|
* @author wangfq
|
||||||
|
* @version V1.0
|
||||||
|
* @date 2026-07-03
|
||||||
|
* @brief IoT MQTT 客户端实现 — DLD960 IoT 协议
|
||||||
|
*
|
||||||
|
* 连接流程: TCP connect → MQTT CONNECT → CONNACK → SUBSCRIBE → SUBACK → READY
|
||||||
|
* 断线重连: 指数退避 5s~60s
|
||||||
|
* 数据上报: loop_data / event_report / heartbeat
|
||||||
|
******************************************************************************
|
||||||
|
*/
|
||||||
|
|
||||||
|
#include "CONFIG.h"
|
||||||
|
#include "iot_mqtt_srv.h"
|
||||||
|
#include "eth_driver.h"
|
||||||
|
#include "wchnet.h"
|
||||||
|
#include "net_srv.h"
|
||||||
|
#include "MQTTPacket.h"
|
||||||
|
#include "cmcng.h"
|
||||||
|
#include "loop_uart_proto.h"
|
||||||
|
#include "simple_json.h"
|
||||||
|
#include "storage.h"
|
||||||
|
#include <string.h>
|
||||||
|
#include <stdlib.h>
|
||||||
|
#include <stdio.h>
|
||||||
|
|
||||||
|
/*===========================================================================
|
||||||
|
* Global State
|
||||||
|
*===========================================================================*/
|
||||||
|
uint8_t g_iot_socket = 0xFF; // MQTT TCP socket ID
|
||||||
|
IotMqttState g_iot_state = IOT_STATE_DISCONNECTED;
|
||||||
|
uint8_t g_iot_msg_id = 0; // MQTT packet identifier
|
||||||
|
static uint32_t _iot_last_heartbeat = 0; // 上次心跳时刻 (ms)
|
||||||
|
static uint32_t _iot_reconnect_deadline = 0;
|
||||||
|
static uint32_t _iot_reconnect_backoff = 0;
|
||||||
|
static uint32_t _iot_connect_start = 0; // TCP connect 开始时刻
|
||||||
|
|
||||||
|
/*===========================================================================
|
||||||
|
* Buffers
|
||||||
|
*===========================================================================*/
|
||||||
|
static uint8_t _iot_recv_buf[IOT_MQTT_RECV_BUF_LEN];
|
||||||
|
static uint16_t _iot_recv_len = 0;
|
||||||
|
static uint8_t _iot_send_buf[IOT_MQTT_SEND_BUF_LEN];
|
||||||
|
static uint8_t _iot_wchnet_buf[RECE_BUF_LEN]; // WCHNET internal recv buffer
|
||||||
|
|
||||||
|
/*===========================================================================
|
||||||
|
* Topic 构建
|
||||||
|
*===========================================================================*/
|
||||||
|
static char g_iot_dev_serial[13]; // 设备序列码 (12 hex chars)
|
||||||
|
|
||||||
|
/* 构建 topic: dld960/{sn}/... */
|
||||||
|
static int iot_make_topic(char *out, uint16_t out_len, const char *direction,
|
||||||
|
const char *category, const char *sub) {
|
||||||
|
if (sub) {
|
||||||
|
return snprintf(out, out_len, "dld960/%s/%s/%s/%s",
|
||||||
|
g_iot_dev_serial, direction, category, sub);
|
||||||
|
}
|
||||||
|
return snprintf(out, out_len, "dld960/%s/%s/%s",
|
||||||
|
g_iot_dev_serial, direction, category);
|
||||||
|
}
|
||||||
|
|
||||||
|
/*===========================================================================
|
||||||
|
* MQTT Packet Helpers
|
||||||
|
*===========================================================================*/
|
||||||
|
|
||||||
|
/* 发送 MQTT 报文到 broker */
|
||||||
|
static int iot_mqtt_send(const uint8_t *buf, uint16_t len) {
|
||||||
|
uint16_t slen = len;
|
||||||
|
uint8_t ret = WCHNET_SocketSend(g_iot_socket, (uint8_t *)buf, &slen);
|
||||||
|
return (ret == WCHNET_ERR_SUCCESS) ? 0 : -1;
|
||||||
|
}
|
||||||
|
|
||||||
|
/* 发送 MQTT CONNECT */
|
||||||
|
static int iot_mqtt_send_connect(void) {
|
||||||
|
MQTTPacket_connectData opts = MQTTPacket_connectData_initializer;
|
||||||
|
uint8_t buf[256];
|
||||||
|
int len;
|
||||||
|
|
||||||
|
opts.MQTTVersion = 4; // MQTT 3.1.1
|
||||||
|
opts.keepAliveInterval = IOT_MQTT_KEEPALIVE_SEC;
|
||||||
|
opts.cleansession = 1;
|
||||||
|
|
||||||
|
// clientID
|
||||||
|
if (strlen((char *)iot_net_info.client_id) > 0) {
|
||||||
|
opts.clientID.cstring = (char *)iot_net_info.client_id;
|
||||||
|
} else {
|
||||||
|
opts.clientID.cstring = g_iot_dev_serial;
|
||||||
|
}
|
||||||
|
|
||||||
|
// username / password
|
||||||
|
if (strlen((char *)iot_net_info.username) > 0) {
|
||||||
|
opts.username.cstring = (char *)iot_net_info.username;
|
||||||
|
}
|
||||||
|
if (strlen((char *)iot_net_info.password) > 0) {
|
||||||
|
opts.password.cstring = (char *)iot_net_info.password;
|
||||||
|
}
|
||||||
|
|
||||||
|
len = MQTTSerialize_connect(buf, sizeof(buf), &opts);
|
||||||
|
if (len <= 0) {
|
||||||
|
PRINT("IOT: MQTTSerialize_connect failed\n");
|
||||||
|
return -1;
|
||||||
|
}
|
||||||
|
PRINT("IOT: → CONNECT clientId=%s keepAlive=%d\n",
|
||||||
|
opts.clientID.cstring, opts.keepAliveInterval);
|
||||||
|
return iot_mqtt_send(buf, (uint16_t)len);
|
||||||
|
}
|
||||||
|
|
||||||
|
/* 发送 MQTT SUBSCRIBE */
|
||||||
|
static int iot_mqtt_send_subscribe(void) {
|
||||||
|
uint8_t buf[256];
|
||||||
|
int len;
|
||||||
|
char topic[IOT_MQTT_TOPIC_MAX_LEN];
|
||||||
|
MQTTString topics[3];
|
||||||
|
int qos[3] = {1, 1, 1};
|
||||||
|
int count = 0;
|
||||||
|
|
||||||
|
// 订阅服务器下发消息
|
||||||
|
iot_make_topic(topic, sizeof(topic), "srv", "config", "set");
|
||||||
|
topics[count].cstring = topic; topics[count].lenstring.len = 0; count++;
|
||||||
|
|
||||||
|
iot_make_topic(topic, sizeof(topic), "srv", "config", "query");
|
||||||
|
topics[count].cstring = topic; topics[count].lenstring.len = 0; count++;
|
||||||
|
|
||||||
|
iot_make_topic(topic, sizeof(topic), "srv", "ctrl");
|
||||||
|
topics[count].cstring = topic; topics[count].lenstring.len = 0; count++;
|
||||||
|
|
||||||
|
len = MQTTSerialize_subscribe(buf, sizeof(buf), 0,
|
||||||
|
++g_iot_msg_id, count, topics, qos);
|
||||||
|
if (len <= 0) {
|
||||||
|
PRINT("IOT: MQTTSerialize_subscribe failed\n");
|
||||||
|
return -1;
|
||||||
|
}
|
||||||
|
PRINT("IOT: → SUBSCRIBE pktId=%d topics=%d\n", g_iot_msg_id, count);
|
||||||
|
return iot_mqtt_send(buf, (uint16_t)len);
|
||||||
|
}
|
||||||
|
|
||||||
|
/* 发送 MQTT PUBLISH */
|
||||||
|
static int iot_mqtt_publish(const char *topic, const char *payload,
|
||||||
|
uint16_t payload_len, uint8_t qos) {
|
||||||
|
uint8_t buf[IOT_MQTT_SEND_BUF_LEN];
|
||||||
|
int len;
|
||||||
|
MQTTString mqtt_topic = MQTTString_initializer;
|
||||||
|
mqtt_topic.cstring = (char *)topic;
|
||||||
|
|
||||||
|
len = MQTTSerialize_publish(buf, sizeof(buf), 0, qos, 0,
|
||||||
|
++g_iot_msg_id, mqtt_topic,
|
||||||
|
(unsigned char *)payload, payload_len);
|
||||||
|
if (len <= 0) {
|
||||||
|
PRINT("IOT: MQTTSerialize_publish failed\n");
|
||||||
|
return -1;
|
||||||
|
}
|
||||||
|
return iot_mqtt_send(buf, (uint16_t)len);
|
||||||
|
}
|
||||||
|
|
||||||
|
/*===========================================================================
|
||||||
|
* MQTT 接收处理
|
||||||
|
*===========================================================================*/
|
||||||
|
|
||||||
|
/* 处理一条收到的 MQTT PUBLISH 消息 */
|
||||||
|
static void iot_handle_publish(const char *topic, uint8_t *payload, int payload_len) {
|
||||||
|
char json[IOT_MQTT_RECV_BUF_LEN];
|
||||||
|
int copy_len = payload_len < (int)sizeof(json) - 1 ? payload_len : (int)sizeof(json) - 1;
|
||||||
|
memcpy(json, payload, copy_len);
|
||||||
|
json[copy_len] = '\0';
|
||||||
|
|
||||||
|
PRINT("IOT: PUBLISH topic=%s payload=%s\n", topic, json);
|
||||||
|
|
||||||
|
/* 提取 msg_id 和 cmd */
|
||||||
|
char tmp[256];
|
||||||
|
uint32_t msg_id = 0;
|
||||||
|
|
||||||
|
// 简单提取: 解析 JSON 中的 msg_id 和 cmd
|
||||||
|
memset(tmp, 0, sizeof(tmp));
|
||||||
|
simple_parse_json(json, "\"msg_id\"", tmp);
|
||||||
|
if (strlen(tmp) > 0) msg_id = (uint32_t)strtoul(tmp, NULL, 10);
|
||||||
|
|
||||||
|
memset(tmp, 0, sizeof(tmp));
|
||||||
|
simple_parse_json(json, "\"cmd\"", tmp);
|
||||||
|
// strip quotes
|
||||||
|
char *cmd_str = tmp;
|
||||||
|
if (cmd_str[0] == '"') cmd_str++;
|
||||||
|
int cmd_len = strlen(cmd_str);
|
||||||
|
if (cmd_len > 0 && cmd_str[cmd_len - 1] == '"') cmd_str[cmd_len - 1] = '\0';
|
||||||
|
|
||||||
|
if (strlen(cmd_str) == 0) {
|
||||||
|
PRINT("IOT: no cmd in PUBLISH\n");
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
/* 构建响应 topic */
|
||||||
|
char resp_topic[IOT_MQTT_TOPIC_MAX_LEN];
|
||||||
|
iot_make_topic(resp_topic, sizeof(resp_topic), "dev", "config", "resp");
|
||||||
|
|
||||||
|
/* 处理命令 — 复用 TCP JSON 的命令逻辑 */
|
||||||
|
/* 目前仅实现基本响应框架, 后续逐步对接 Loop MCU 命令 */
|
||||||
|
|
||||||
|
if (strcmp(cmd_str, "dev_info_query") == 0) {
|
||||||
|
char data_json[512];
|
||||||
|
snprintf(data_json, sizeof(data_json),
|
||||||
|
"{\"msg_id\":%lu,\"cmd\":\"dev_info_query\","
|
||||||
|
"\"ts\":%lu,\"code\":0,\"msg\":\"success\","
|
||||||
|
"\"data\":{"
|
||||||
|
"\"dev_serial\":\"%s\",\"hard_ver\":\"%s\",\"soft_ver\":\"%s\","
|
||||||
|
"\"model\":\"%s\",\"product_code\":\"960001\","
|
||||||
|
"\"sub_code\":{\"net\":%s,\"iot\":%s},"
|
||||||
|
"\"bus\":{\"bus1\":0,\"bus2\":0,\"bus3\":0,\"bus4\":0}}}",
|
||||||
|
msg_id, mstick() / 1000,
|
||||||
|
g_dev_number_str, HARDWARE_VER, FIRMWARE_VER, PRODUCT_MODEL,
|
||||||
|
g_sub_code_enable.net_enable ? "true" : "false",
|
||||||
|
g_sub_code_enable.iot_enable ? "true" : "false");
|
||||||
|
iot_mqtt_publish(resp_topic, data_json, strlen(data_json), 1);
|
||||||
|
|
||||||
|
} else if (strcmp(cmd_str, "pwd_verify") == 0) {
|
||||||
|
// 验证密码
|
||||||
|
char password[16] = {0};
|
||||||
|
char *data = (char *)malloc(512);
|
||||||
|
if (data) {
|
||||||
|
memset(data, 0, 512);
|
||||||
|
simple_parse_json(json, "\"data\"", data);
|
||||||
|
if (strlen(data) > 0) {
|
||||||
|
memset(tmp, 0, sizeof(tmp));
|
||||||
|
simple_parse_json(data, "\"password\"", tmp);
|
||||||
|
// strip quotes
|
||||||
|
char *pwd = tmp;
|
||||||
|
if (pwd[0] == '"') pwd++;
|
||||||
|
int plen = strlen(pwd);
|
||||||
|
if (plen > 0 && pwd[plen - 1] == '"') pwd[plen - 1] = '\0';
|
||||||
|
strncpy(password, pwd, 15);
|
||||||
|
}
|
||||||
|
free(data);
|
||||||
|
}
|
||||||
|
|
||||||
|
if (strlen(password) == 6 && memcmp(password, g_dev_password, 6) == 0) {
|
||||||
|
char resp[256];
|
||||||
|
snprintf(resp, sizeof(resp),
|
||||||
|
"{\"msg_id\":%lu,\"cmd\":\"pwd_verify\","
|
||||||
|
"\"ts\":%lu,\"code\":0,\"msg\":\"success\"}",
|
||||||
|
msg_id, mstick() / 1000);
|
||||||
|
iot_mqtt_publish(resp_topic, resp, strlen(resp), 1);
|
||||||
|
PRINT("IOT: Auth success via MQTT\n");
|
||||||
|
} else {
|
||||||
|
char resp[256];
|
||||||
|
snprintf(resp, sizeof(resp),
|
||||||
|
"{\"msg_id\":%lu,\"cmd\":\"pwd_verify\","
|
||||||
|
"\"ts\":%lu,\"code\":2,\"msg\":\"password incorrect\"}",
|
||||||
|
msg_id, mstick() / 1000);
|
||||||
|
iot_mqtt_publish(resp_topic, resp, strlen(resp), 1);
|
||||||
|
}
|
||||||
|
|
||||||
|
} else {
|
||||||
|
// 暂不支持的命令, 返回错误
|
||||||
|
char resp[256];
|
||||||
|
snprintf(resp, sizeof(resp),
|
||||||
|
"{\"msg_id\":%lu,\"cmd\":\"%s\","
|
||||||
|
"\"ts\":%lu,\"code\":4,\"msg\":\"unsupported command\"}",
|
||||||
|
msg_id, cmd_str, mstick() / 1000);
|
||||||
|
iot_mqtt_publish(resp_topic, resp, strlen(resp), 1);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/* 处理 MQTT SUBACK */
|
||||||
|
static void iot_handle_suback(void) {
|
||||||
|
PRINT("IOT: ← SUBACK, topics subscribed\n");
|
||||||
|
g_iot_state = IOT_STATE_READY;
|
||||||
|
_iot_reconnect_backoff = 0; // 连接成功, 重置退避
|
||||||
|
}
|
||||||
|
|
||||||
|
/* 处理收到的 MQTT 数据 */
|
||||||
|
static void iot_process_recv(void) {
|
||||||
|
uint8_t header;
|
||||||
|
int qos, payload_len;
|
||||||
|
unsigned char retained;
|
||||||
|
unsigned short packet_id;
|
||||||
|
MQTTString topic;
|
||||||
|
unsigned char *payload_ptr;
|
||||||
|
|
||||||
|
while (_iot_recv_len >= 2) {
|
||||||
|
header = _iot_recv_buf[0];
|
||||||
|
int msg_type = (header >> 4) & 0x0F;
|
||||||
|
|
||||||
|
/* 先尝试解析整个 MQTT 包 */
|
||||||
|
int mqtt_pkt_len = 0;
|
||||||
|
int multiplier = 1;
|
||||||
|
int rem_len = 0;
|
||||||
|
int i = 1;
|
||||||
|
while (i < (int)_iot_recv_len && i < 5) {
|
||||||
|
rem_len += (_iot_recv_buf[i] & 0x7F) * multiplier;
|
||||||
|
multiplier *= 128;
|
||||||
|
if ((_iot_recv_buf[i] & 0x80) == 0) {
|
||||||
|
mqtt_pkt_len = 1 + (i - 1 + 1) + rem_len; // header + rem_len_bytes + payload
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
i++;
|
||||||
|
}
|
||||||
|
if (mqtt_pkt_len == 0 || mqtt_pkt_len > (int)_iot_recv_len) {
|
||||||
|
return; // 包不完整, 等更多数据
|
||||||
|
}
|
||||||
|
|
||||||
|
PRINT("IOT: ← MQTT packet type=%d len=%d\n", msg_type, mqtt_pkt_len);
|
||||||
|
|
||||||
|
switch (msg_type) {
|
||||||
|
case CONNACK: {
|
||||||
|
unsigned char session_present, connack_rc;
|
||||||
|
if (MQTTDeserialize_connack(&session_present, &connack_rc,
|
||||||
|
_iot_recv_buf, mqtt_pkt_len) == 1) {
|
||||||
|
PRINT("IOT: ← CONNACK rc=%d\n", connack_rc);
|
||||||
|
if (connack_rc == 0) {
|
||||||
|
// 连接成功 → 订阅
|
||||||
|
g_iot_state = IOT_STATE_MQTT_CONNECTED;
|
||||||
|
iot_mqtt_send_subscribe();
|
||||||
|
} else {
|
||||||
|
PRINT("IOT: CONNACK rejected, rc=%d\n", connack_rc);
|
||||||
|
g_iot_state = IOT_STATE_DISCONNECTED;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
|
case SUBACK:
|
||||||
|
iot_handle_suback();
|
||||||
|
break;
|
||||||
|
|
||||||
|
case PUBLISH: {
|
||||||
|
unsigned char pub_dup;
|
||||||
|
if (MQTTDeserialize_publish(&pub_dup, &qos, &retained,
|
||||||
|
&packet_id, &topic,
|
||||||
|
&payload_ptr, &payload_len,
|
||||||
|
_iot_recv_buf, mqtt_pkt_len) == 1) {
|
||||||
|
char topic_str[IOT_MQTT_TOPIC_MAX_LEN];
|
||||||
|
memcpy(topic_str, topic.cstring ? topic.cstring : "",
|
||||||
|
topic.lenstring.len < (int)sizeof(topic_str) - 1
|
||||||
|
? topic.lenstring.len : (int)sizeof(topic_str) - 1);
|
||||||
|
topic_str[topic.lenstring.len] = '\0';
|
||||||
|
|
||||||
|
iot_handle_publish(topic_str, payload_ptr, payload_len);
|
||||||
|
|
||||||
|
// 如果 QoS > 0, 发送 PUBACK
|
||||||
|
if (qos > 0) {
|
||||||
|
uint8_t ack_buf[4];
|
||||||
|
int ack_len = MQTTSerialize_ack(ack_buf, sizeof(ack_buf),
|
||||||
|
PUBACK, 0, packet_id);
|
||||||
|
iot_mqtt_send(ack_buf, (uint16_t)ack_len);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
break;
|
||||||
|
|
||||||
|
case PINGRESP:
|
||||||
|
PRINT("IOT: ← PINGRESP\n");
|
||||||
|
break;
|
||||||
|
|
||||||
|
default:
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
|
/* 移除已处理的包 */
|
||||||
|
memmove(_iot_recv_buf, _iot_recv_buf + mqtt_pkt_len,
|
||||||
|
_iot_recv_len - mqtt_pkt_len);
|
||||||
|
_iot_recv_len -= mqtt_pkt_len;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/*===========================================================================
|
||||||
|
* 传感器数据上报
|
||||||
|
*===========================================================================*/
|
||||||
|
void iot_mqtt_publish_sensor(void) {
|
||||||
|
if (g_iot_state != IOT_STATE_READY) return;
|
||||||
|
if (!g_report_active) return;
|
||||||
|
|
||||||
|
/* 复用 g_pkg_uart_2 中的传感器帧 */
|
||||||
|
if (g_pkg_uart_2.flag == 0) return;
|
||||||
|
if (g_pkg_uart_2.pkg[0] != 0x7F) return;
|
||||||
|
if (g_pkg_uart_2.pkg[3] != 0xC0) return;
|
||||||
|
|
||||||
|
LUP_SensorReport sr;
|
||||||
|
memset(&sr, 0, sizeof(sr));
|
||||||
|
int ret = lup_parse_sensor_report(g_pkg_uart_2.pkg, g_pkg_uart_2.offset, &sr);
|
||||||
|
if (ret != 0) {
|
||||||
|
PRINT("IOT: sensor parse failed (%d)\n", ret);
|
||||||
|
InitPkgUart(&g_pkg_uart_2);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
/* 构建 loop_data JSON */
|
||||||
|
char data_json[2048];
|
||||||
|
char *p = data_json;
|
||||||
|
int remaining = sizeof(data_json);
|
||||||
|
int written;
|
||||||
|
uint8_t i;
|
||||||
|
|
||||||
|
written = snprintf(p, remaining,
|
||||||
|
"{\"channels\":[");
|
||||||
|
if (written < 0 || written >= remaining) goto done;
|
||||||
|
p += written; remaining -= written;
|
||||||
|
|
||||||
|
for (i = 0; i < sr.coil_count; i++) {
|
||||||
|
const LUP_CoilSensor *cs = &sr.coils[i];
|
||||||
|
const char *misc_type_str = "time";
|
||||||
|
uint32_t misc_val = 0;
|
||||||
|
|
||||||
|
if (cs->misc_type == 0) { misc_type_str = "time"; misc_val = cs->misc.passtime_ms; }
|
||||||
|
else if (cs->misc_type == 1) { misc_type_str = "cut_count"; misc_val = cs->misc.cut_amount; }
|
||||||
|
else if (cs->misc_type == 2) { misc_type_str = "flow_count"; misc_val = cs->misc.flow_amount; }
|
||||||
|
else if (cs->misc_type == 3) { misc_type_str = "relay_count"; misc_val = cs->misc.relay_count; }
|
||||||
|
|
||||||
|
const char *freq_level_names[] = {"high", "mid_high", "mid_low", "low"};
|
||||||
|
|
||||||
|
written = snprintf(p, remaining,
|
||||||
|
"%s{\"ch\":%d,\"freq_level\":\"%s\",\"has_car\":%s,"
|
||||||
|
"\"loop_ok\":%s,\"freq_current\":%lu,\"freq_diff\":%d,"
|
||||||
|
"\"sensitivity\":%d,\"condition\":%d,"
|
||||||
|
"\"misc\":{\"type\":\"%s\",\"value\":%lu}}",
|
||||||
|
(i > 0) ? "," : "",
|
||||||
|
i + 1,
|
||||||
|
freq_level_names[cs->freq_level],
|
||||||
|
cs->car_state ? "true" : "false",
|
||||||
|
cs->loop_state ? "false" : "true",
|
||||||
|
cs->freq, cs->variation,
|
||||||
|
cs->sensitivity, cs->condition,
|
||||||
|
misc_type_str, misc_val);
|
||||||
|
if (written < 0 || written >= remaining) goto done;
|
||||||
|
p += written; remaining -= written;
|
||||||
|
}
|
||||||
|
|
||||||
|
snprintf(p, remaining, "]}");
|
||||||
|
|
||||||
|
/* 包装为通用消息格式 */
|
||||||
|
char payload[IOT_MQTT_SEND_BUF_LEN];
|
||||||
|
snprintf(payload, sizeof(payload),
|
||||||
|
"{\"msg_id\":%d,\"cmd\":\"loop_data\",\"ts\":%lu,\"data\":%s}",
|
||||||
|
++g_iot_msg_id, mstick() / 1000, data_json);
|
||||||
|
|
||||||
|
char topic[IOT_MQTT_TOPIC_MAX_LEN];
|
||||||
|
iot_make_topic(topic, sizeof(topic), "dev", "data", "loop");
|
||||||
|
iot_mqtt_publish(topic, payload, strlen(payload), 0);
|
||||||
|
|
||||||
|
done:
|
||||||
|
InitPkgUart(&g_pkg_uart_2);
|
||||||
|
}
|
||||||
|
|
||||||
|
/*===========================================================================
|
||||||
|
* 心跳上报
|
||||||
|
*===========================================================================*/
|
||||||
|
static void iot_send_heartbeat(void) {
|
||||||
|
if (g_iot_state != IOT_STATE_READY) return;
|
||||||
|
|
||||||
|
char payload[512];
|
||||||
|
uint8_t loop_ok[4] = {1, 1, 1, 1};
|
||||||
|
uint8_t i;
|
||||||
|
for (i = 0; i < 4; i++) {
|
||||||
|
/* 简化: 读取线圈状态 */
|
||||||
|
loop_ok[i] = 1; // TODO: 从 Loop MCU 获取实际状态
|
||||||
|
}
|
||||||
|
|
||||||
|
snprintf(payload, sizeof(payload),
|
||||||
|
"{\"msg_id\":%d,\"cmd\":\"heartbeat\","
|
||||||
|
"\"ts\":%lu,\"data\":{"
|
||||||
|
"\"uptime\":%lu,\"loop_status\":[%s,%s,%s,%s],"
|
||||||
|
"\"net_status\":true,\"iot_status\":true}}",
|
||||||
|
++g_iot_msg_id, mstick() / 1000,
|
||||||
|
mstick() / 1000,
|
||||||
|
loop_ok[0] ? "true" : "false", loop_ok[1] ? "true" : "false",
|
||||||
|
loop_ok[2] ? "true" : "false", loop_ok[3] ? "true" : "false");
|
||||||
|
|
||||||
|
char topic[IOT_MQTT_TOPIC_MAX_LEN];
|
||||||
|
iot_make_topic(topic, sizeof(topic), "dev", "status");
|
||||||
|
iot_mqtt_publish(topic, payload, strlen(payload), 0);
|
||||||
|
}
|
||||||
|
|
||||||
|
/*===========================================================================
|
||||||
|
* 连接管理
|
||||||
|
*===========================================================================*/
|
||||||
|
|
||||||
|
/* 连接到 MQTT Broker */
|
||||||
|
static void iot_connect_broker(void) {
|
||||||
|
struct _WCHNET_IPAddr broker_ip;
|
||||||
|
SOCK_INF sock_inf;
|
||||||
|
uint8_t ret;
|
||||||
|
|
||||||
|
// 解析 broker 地址
|
||||||
|
if (get_ipstr_to_array((char *)iot_net_info.remote_addr, broker_ip.a) == 0) {
|
||||||
|
// 是 IP 地址
|
||||||
|
memcpy(broker_ip.a, iot_net_info.remote_addr, 4);
|
||||||
|
PRINT("IOT: broker IP %d.%d.%d.%d:%d\n",
|
||||||
|
broker_ip.a[0], broker_ip.a[1], broker_ip.a[2], broker_ip.a[3],
|
||||||
|
iot_net_info.mqtt_port);
|
||||||
|
} else {
|
||||||
|
PRINT("IOT: broker hostname=%s (DNS not implemented yet)\n",
|
||||||
|
iot_net_info.remote_addr);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
memset(&sock_inf, 0, sizeof(SOCK_INF));
|
||||||
|
sock_inf.DestPort = iot_net_info.mqtt_port;
|
||||||
|
sock_inf.ProtoType = PROTO_TYPE_TCP;
|
||||||
|
sock_inf.RecvBufLen = RECE_BUF_LEN;
|
||||||
|
memcpy(sock_inf.DestAddr, broker_ip.a, 4);
|
||||||
|
|
||||||
|
ret = WCHNET_SocketCreat(&g_iot_socket, &sock_inf);
|
||||||
|
if (ret != WCHNET_ERR_SUCCESS) {
|
||||||
|
PRINT("IOT: SocketCreat failed: 0x%02X\n", ret);
|
||||||
|
g_iot_socket = 0xFF;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
ret = WCHNET_SocketConnect(g_iot_socket);
|
||||||
|
if (ret != WCHNET_ERR_SUCCESS) {
|
||||||
|
PRINT("IOT: SocketConnect failed: 0x%02X\n", ret);
|
||||||
|
WCHNET_SocketClose(g_iot_socket, TCP_CLOSE_NORMAL);
|
||||||
|
g_iot_socket = 0xFF;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
g_iot_state = IOT_STATE_TCP_CONNECTING;
|
||||||
|
_iot_connect_start = mstick();
|
||||||
|
PRINT("IOT: TCP connecting to broker (sock=%d)...\n", g_iot_socket);
|
||||||
|
}
|
||||||
|
|
||||||
|
/*===========================================================================
|
||||||
|
* Socket 中断处理
|
||||||
|
*===========================================================================*/
|
||||||
|
void iot_mqtt_handle_sock_int(uint8_t socketid, uint8_t intstat) {
|
||||||
|
if (socketid != g_iot_socket) return;
|
||||||
|
|
||||||
|
/* CONNECT 成功 */
|
||||||
|
if (intstat & SINT_STAT_CONNECT) {
|
||||||
|
PRINT("IOT: TCP connected (sock=%d)\n", socketid);
|
||||||
|
WCHNET_ModifyRecvBuf(socketid, (uint32_t)_iot_wchnet_buf, RECE_BUF_LEN);
|
||||||
|
g_iot_state = IOT_STATE_TCP_CONNECTED;
|
||||||
|
_iot_recv_len = 0;
|
||||||
|
// 发送 MQTT CONNECT
|
||||||
|
iot_mqtt_send_connect();
|
||||||
|
g_iot_state = IOT_STATE_MQTT_CONNECTING;
|
||||||
|
}
|
||||||
|
|
||||||
|
/* 收到数据 */
|
||||||
|
if (intstat & SINT_STAT_RECV) {
|
||||||
|
uint32_t recv_len = WCHNET_SocketRecvLen(socketid, NULL);
|
||||||
|
if (recv_len > 0) {
|
||||||
|
uint16_t space = IOT_MQTT_RECV_BUF_LEN - _iot_recv_len;
|
||||||
|
if (recv_len > space) recv_len = space;
|
||||||
|
uint32_t rd_len = recv_len;
|
||||||
|
uint8_t tmp[RECE_BUF_LEN];
|
||||||
|
WCHNET_SocketRecv(socketid, tmp, &rd_len);
|
||||||
|
memcpy(_iot_recv_buf + _iot_recv_len, tmp, (uint16_t)rd_len);
|
||||||
|
_iot_recv_len += (uint16_t)rd_len;
|
||||||
|
iot_process_recv();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/* 断开 / 超时 */
|
||||||
|
if (intstat & (SINT_STAT_DISCONNECT | SINT_STAT_TIM_OUT)) {
|
||||||
|
PRINT("IOT: socket disconnect/timeout (sock=%d)\n", socketid);
|
||||||
|
g_iot_state = IOT_STATE_DISCONNECTED;
|
||||||
|
g_iot_socket = 0xFF;
|
||||||
|
_iot_recv_len = 0;
|
||||||
|
_iot_reconnect_backoff = IOT_MQTT_RECONNECT_MIN_MS;
|
||||||
|
_iot_reconnect_deadline = mstick() + _iot_reconnect_backoff;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/*===========================================================================
|
||||||
|
* 主循环轮询
|
||||||
|
*===========================================================================*/
|
||||||
|
void iot_mqtt_poll(void) {
|
||||||
|
/* 断线重连 */
|
||||||
|
if (g_iot_state == IOT_STATE_DISCONNECTED && g_iot_socket == 0xFF) {
|
||||||
|
if (_iot_reconnect_deadline == 0 || mstick() > _iot_reconnect_deadline) {
|
||||||
|
iot_connect_broker();
|
||||||
|
if (g_iot_socket == 0xFF) {
|
||||||
|
// 连接失败, 退避
|
||||||
|
if (_iot_reconnect_backoff == 0) {
|
||||||
|
_iot_reconnect_backoff = IOT_MQTT_RECONNECT_MIN_MS;
|
||||||
|
} else {
|
||||||
|
_iot_reconnect_backoff *= 2;
|
||||||
|
if (_iot_reconnect_backoff > IOT_MQTT_RECONNECT_MAX_MS)
|
||||||
|
_iot_reconnect_backoff = IOT_MQTT_RECONNECT_MAX_MS;
|
||||||
|
}
|
||||||
|
_iot_reconnect_deadline = mstick() + _iot_reconnect_backoff;
|
||||||
|
PRINT("IOT: reconnect in %lu ms\n", _iot_reconnect_backoff);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
/* TCP 连接超时 (10s) */
|
||||||
|
if (g_iot_state == IOT_STATE_TCP_CONNECTING) {
|
||||||
|
if (mstick() - _iot_connect_start > 10000) {
|
||||||
|
PRINT("IOT: TCP connect timeout\n");
|
||||||
|
WCHNET_SocketClose(g_iot_socket, TCP_CLOSE_NORMAL);
|
||||||
|
g_iot_socket = 0xFF;
|
||||||
|
g_iot_state = IOT_STATE_DISCONNECTED;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/* MQTT Keepalive — PINGREQ */
|
||||||
|
if (g_iot_state == IOT_STATE_READY) {
|
||||||
|
uint32_t now = mstick();
|
||||||
|
if (now - _iot_last_heartbeat > IOT_MQTT_HEARTBEAT_MS) {
|
||||||
|
uint8_t ping_buf[2];
|
||||||
|
int ping_len = MQTTSerialize_pingreq(ping_buf, sizeof(ping_buf));
|
||||||
|
iot_mqtt_send(ping_buf, (uint16_t)ping_len);
|
||||||
|
_iot_last_heartbeat = now;
|
||||||
|
// 心跳上报
|
||||||
|
iot_send_heartbeat();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/*===========================================================================
|
||||||
|
* 初始化
|
||||||
|
*===========================================================================*/
|
||||||
|
void iot_mqtt_init(void) {
|
||||||
|
/* 初始化设备序列码 (12 hex chars) */
|
||||||
|
snprintf(g_iot_dev_serial, sizeof(g_iot_dev_serial),
|
||||||
|
"%02X%02X%02X%02X%02X%02X",
|
||||||
|
g_dev_number[0], g_dev_number[1], g_dev_number[2],
|
||||||
|
g_dev_number[3], g_dev_number[4], g_dev_number[5]);
|
||||||
|
|
||||||
|
PRINT("IOT: dev_serial=%s\n", g_iot_dev_serial);
|
||||||
|
|
||||||
|
/* 初始化传感器回调 — 用于 MQTT 上报 */
|
||||||
|
lup_set_sensor_callback(NULL); // MQTT 模式不需要 JSON sensor_cb
|
||||||
|
|
||||||
|
_iot_reconnect_backoff = 0;
|
||||||
|
_iot_reconnect_deadline = mstick() + 2000; // 启动后 2s 开始首次连接
|
||||||
|
g_iot_state = IOT_STATE_DISCONNECTED;
|
||||||
|
}
|
||||||
@@ -18,6 +18,9 @@
|
|||||||
#include "cmcng.h"
|
#include "cmcng.h"
|
||||||
#include "simple_json.h"
|
#include "simple_json.h"
|
||||||
#include "tcp_json_srv.h"
|
#include "tcp_json_srv.h"
|
||||||
|
#if NET_IOT_ENABLE
|
||||||
|
#include "iot_mqtt_srv.h"
|
||||||
|
#endif
|
||||||
|
|
||||||
|
|
||||||
uint32_t slen;
|
uint32_t slen;
|
||||||
@@ -539,9 +542,14 @@ void WCHNET_HandleSockInt(uint8_t socketid, uint8_t intstat)
|
|||||||
g_net_state.flag = 1;
|
g_net_state.flag = 1;
|
||||||
// WCHNET_CreateTcpSocket();
|
// WCHNET_CreateTcpSocket();
|
||||||
}
|
}
|
||||||
#else /* !NET_SSC_ENABLE — all sockets belong to JSON handler */
|
#else /* !NET_SSC_ENABLE */
|
||||||
|
#if NET_IOT_ENABLE
|
||||||
|
/* IoT MQTT mode — route MQTT socket, JSON server not active */
|
||||||
|
iot_mqtt_handle_sock_int(socketid, intstat);
|
||||||
|
#else
|
||||||
tcp_json_handle_sock_int(socketid, intstat);
|
tcp_json_handle_sock_int(socketid, intstat);
|
||||||
#endif
|
#endif
|
||||||
|
#endif
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
@@ -1123,7 +1131,11 @@ void net_srv_init(void)
|
|||||||
#if NET_SSC_ENABLE
|
#if NET_SSC_ENABLE
|
||||||
WCHNET_CreateUdpSocket();
|
WCHNET_CreateUdpSocket();
|
||||||
#endif
|
#endif
|
||||||
|
#if NET_IOT_ENABLE
|
||||||
|
iot_mqtt_init(); // Init IoT MQTT client (connect to broker)
|
||||||
|
#else
|
||||||
tcp_json_srv_init(); // Start JSON protocol TCP listener on port 5960
|
tcp_json_srv_init(); // Start JSON protocol TCP listener on port 5960
|
||||||
|
#endif
|
||||||
#if !NET_SSC_ENABLE
|
#if !NET_SSC_ENABLE
|
||||||
g_net_state.flag = 2; // Skip SSC init, advance state directly
|
g_net_state.flag = 2; // Skip SSC init, advance state directly
|
||||||
#endif
|
#endif
|
||||||
|
|||||||
@@ -24,6 +24,9 @@
|
|||||||
#include "storage.h"
|
#include "storage.h"
|
||||||
#include "tcp_json_srv.h"
|
#include "tcp_json_srv.h"
|
||||||
#include "loop_uart_proto.h"
|
#include "loop_uart_proto.h"
|
||||||
|
#if NET_IOT_ENABLE
|
||||||
|
#include "iot_mqtt_srv.h"
|
||||||
|
#endif
|
||||||
|
|
||||||
/*********************************************************************
|
/*********************************************************************
|
||||||
* GLOBAL TYPEDEFS
|
* GLOBAL TYPEDEFS
|
||||||
@@ -270,9 +273,13 @@ void Main_Circulation(void)
|
|||||||
|
|
||||||
poll_dbn_ble();
|
poll_dbn_ble();
|
||||||
|
|
||||||
|
#if NET_IOT_ENABLE
|
||||||
|
iot_mqtt_publish_sensor(); // Push 0xC0 sensor data to MQTT broker
|
||||||
|
iot_mqtt_poll(); // IoT MQTT state machine + heartbeat
|
||||||
|
#else
|
||||||
tcp_json_push_sensor(); // Push 0xC0 sensor data to TCP JSON client
|
tcp_json_push_sensor(); // Push 0xC0 sensor data to TCP JSON client
|
||||||
|
|
||||||
tcp_json_poll();
|
tcp_json_poll();
|
||||||
|
#endif
|
||||||
|
|
||||||
key_event_srv();
|
key_event_srv();
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user