/** ****************************************************************************** * @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 "tcp_json_srv.h" #include "storage.h" #include #include #include /*=========================================================================== * Global State *===========================================================================*/ extern ReportConfig g_report_cfg; // from tcp_json_srv.c extern uint8_t SocketId_TCP; // from net_srv.c, MQTT socket created by WCHNET_CreateTcpMqttSocket 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; PRINT("IOT: SocketSend sock=%d len=%d\n", g_iot_socket, len); uint8_t ret = WCHNET_SocketSend(g_iot_socket, (uint8_t *)buf, &slen); PRINT("IOT: SocketSend ret=0x%02X sent=%d\n", ret, slen); return (ret == WCHNET_ERR_SUCCESS) ? 0 : -1; } /* 发送 MQTT CONNECT — 参考 net_srv.c mqtt_connect 实现 */ static int iot_mqtt_send_connect(void) { static MQTTPacket_connectData opts; // static: 避免栈开销 static uint8_t buf[256]; int len; // 手动初始化（参考 net_srv.c 模式，不依赖 brace-initializer 赋值） memset(&opts, 0, sizeof(opts)); opts.struct_id[0] = 'M'; opts.struct_id[1] = 'Q'; opts.struct_id[2] = 'T'; opts.struct_id[3] = 'C'; opts.MQTTVersion = 4; opts.keepAliveInterval = IOT_MQTT_KEEPALIVE_SEC; opts.cleansession = 1; // clientID — 参考 net_srv.c 直接使用 g_dev_number_str opts.clientID.cstring = g_dev_number_str; // username / password — 参考 net_srv.c 直接赋值 opts.username.cstring = (char *)iot_net_info.username; 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) { static uint8_t buf[256]; // static: 避免栈溢出 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", NULL); 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) { static uint8_t buf[IOT_MQTT_SEND_BUF_LEN]; // static: 避免 2KB 栈溢出 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) { static char json[IOT_MQTT_RECV_BUF_LEN]; // static: 避免 1KB 栈开销 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_cfg.enable) return; /* Reuse g_pkg_uart_2 sensor frame from Loop MCU */ 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; } /* Batch-publish: 2 channels per message to keep each under 512 bytes. Full-field JSON (freq_current, freq_diff, condition, misc) per the protocol doc. */ static char data_json[1024]; // single buffer for 2-channel JSON static char payload[1024]; char *pub_topic = (char *)g_iot_topic.topic_pub; uint8_t batch_start = 0; uint8_t i; while (batch_start < sr.coil_count) { char *p = data_json; int remaining = sizeof(data_json); int written; uint8_t batch_end = batch_start + 2; if (batch_end > sr.coil_count) batch_end = sr.coil_count; written = snprintf(p, remaining, "{\"channels\":["); if (written < 0 || written >= remaining) goto done; p += written; remaining -= written; for (i = batch_start; i < batch_end; 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 > batch_start) ? "," : "", 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, "]}"); /* Wrap in outer message and publish */ snprintf(payload, sizeof(payload), "{\"msg_id\":%d,\"cmd\":\"loop_data\",\"ts\":%lu,\"data\":%s}", ++g_iot_msg_id, mstick() / 1000, data_json); uint16_t plen = (uint16_t)strlen(payload); mqtt_publish(pub_topic, payload, 0); batch_start = batch_end; } done: InitPkgUart(&g_pkg_uart_2); } 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", NULL); iot_mqtt_publish(topic, payload, strlen(payload), 0); } /*=========================================================================== * 连接管理 *===========================================================================*/ /* 连接到 MQTT Broker — 复用 SocketId_TCP（已由 WCHNET_CreateTcpMqttSocket 创建并连接） */ static void iot_connect_broker(void) { g_iot_socket = SocketId_TCP; // 复用已创建的 MQTT socket 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 成功 — 仅标记状态，由 poll 延迟发送 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; // 不在此发送 CONNECT，等下一轮 poll 处理 } /* 收到数据 */ 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; } } /* TCP 已连接 — 发送 MQTT CONNECT（延迟到 poll 而非中断内） */ if (g_iot_state == IOT_STATE_TCP_CONNECTED) { iot_mqtt_send_connect(); g_iot_state = IOT_STATE_MQTT_CONNECTING; } /* 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) { /* 复用 net_srv.c 创建的 MQTT socket (SocketId_TCP) */ g_iot_socket = SocketId_TCP; /* 初始化设备序列码 (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; }