乐于分享
好东西不私藏

源码加更04_CONNECT、PING、DISCONNECT 会话生命周期

源码加更04_CONNECT、PING、DISCONNECT 会话生命周期

源码加更04_CONNECT、PING、DISCONNECT 会话生命周期

[!abstract] 这一篇完整公开 MQTT 会话建立、CONNACK 处理、心跳和断开的代码。它对应工业现场最常见的连不上、心跳超时和断线重连问题。

适合谁收藏

  • 正在对照 CONNECT/CONNACK 十六进制报文和 ST 代码的读者。
  • 需要排查 KeepAlive、PINGRESP 和 DISCONNECT 收口的工程师。
  • 想看 MQTT 3.1.1/5.0 版本边界如何落到变量的人。

本篇核心图

读图重点:先看源码对象之间的职责边界,再看数据、状态和错误如何沿着调用链流动。源码加更不是把文件名列出来,而是把完整代码、工程意图和验证口径一起讲清楚。

先给结论

会话生命周期只有三件事:正确建立、持续保活、明确断开。任何一步含糊,现场就会把错误统一叫成“掉线”。

从理论到代码实现链路

MQTT 标准给的是报文类型、固定头、可变头、载荷、QoS 交互和会话语义;PLC 工程真正要解决的是周期扫描、缓冲区长度、错误锁存、在线变量、连接重入和现场可诊断性。

所以这套开源实现不能只按协议章节拆,也不能只按文件名拆。正确读法是把标准约束翻译成程序对象:入口程序负责给命令和观测点,GVL 和 DUT 定义容量与数据模型,主功能块负责调度状态机,构建方法负责出站报文,处理方法负责入站报文,辅助方法负责长度、队列、事务、主题和诊断边界。

本篇完整公开 CONNECT、PINGREQ、DISCONNECT 构建和 CONNACK/PINGRESP 处理方法。

再往下一层看,这里其实有两条线同时存在。第一条是协议线:固定头、Remaining Length、PacketId、QoS、Topic、Payload 和 Reason Code 必须能按 MQTT 规则组合起来。第二条是 PLC 工程线:每个周期只能推进有限步骤,所有中间状态都要能被在线变量观察,所有错误都要能被锁存并归类,所有缓冲区长度都要在写入前被检查。

这就是源码加更必须完整公开的原因。只给几段核心片段,读者最多能看懂某个判断;把完整对象放出来,读者才能看到对象之间如何传递状态、长度、错误和诊断信息。完整源码讲解不是为了堆代码,而是为了让读者能从标准约束一路追到可运行的 ST 对象,再从现场现象反向定位到具体边界。

本篇公开的完整源码范围

序号
源码对象
讲解重点
1
M_BuildConnectPacket.st
出站报文构建,把引脚命令翻译成 MQTT 字节流
2
M_BuildPingReqPacket.st
出站报文构建,把引脚命令翻译成 MQTT 字节流
3
M_BuildDisconnectPacket.st
出站报文构建,把引脚命令翻译成 MQTT 字节流
4
M_HandleConnAck.st
入站报文处理,把 MQTT 响应落到状态和诊断
5
M_HandlePingResp.st
入站报文处理,把 MQTT 响应落到状态和诊断

怎么读这些源码

第一遍只看对象职责:这个文件解决哪一层问题,是入口、模型、状态、构建、接收、事务,还是诊断。

第二遍看边界变量:长度、索引、PacketId、QoS、状态枚举、错误码、缓冲区水位和在线观测量。PLC 通信代码最怕的是“能跑但不可诊断”,所以每个关键对象都要问一句:现场出问题时,我能不能从它留下的变量看出原因。

第三遍再看具体语句。源码全部公开,不等于读者要从第一行顺序读到最后一行。更稳的方式是用图和表先建立地图,再回到完整代码里确认每个边界确实落地。

工程验证路径

验证时依次看 CONNECT 报文长度、CONNACK 返回码、心跳周期和断开状态是否收口。

本篇完整开源代码

完整代码 1:M_BuildConnectPacket.st

这一段完整公开 M_BuildConnectPacket.st。读代码时先看对象职责,再看状态、长度、错误和返回值,不要只抄几行赋值。

/// =======================================================================/// 名称      : M_BuildConnectPacket/// 功能      : 构建 CONNECT 发送报文/// 说明      : 按 MQTT 3.1.1/5.0 规则组装 CONNECT 报文并写入发送缓冲区。/// 编程人员  : ControlRookie/// 时间      : 2026-05-05/// 版本      : V1.0/// ======================================================================={attribute 'hide_all_locals'}METHOD M_BuildConnectPacket : BOOLVAR    uiPos               : UINT := 0;                                        // 当前写入发送缓冲区的位置偏移[byte]    uiVarHeaderLen      : UINT;                                             // CONNECT 可变报头总长度[byte]    uiPayloadLen        : UINT;                                             // CONNECT 载荷总长度[byte]    uiRemainingLen      : UINT;                                             // 写入固定报头中的 Remaining Length 值[byte]    uiPropsLen          : UINT;                                             // MQTT 5.0 CONNECT 属性区总长度[byte]    i                   : DINT;                                             // 清空发送缓冲区时使用的循环索引    byConnectFlags      : BYTE;                                             // 即将写入 CONNECT 报文的连接标志字节    uiTopicAliasMax     : UINT;                                             // 客户端告诉 Broker“我最多能接收多少个主题别名编号”END_VAR// === IMPLEMENTATION ===// BUG-10: 缓冲区溢出保护IF SIZEOF(aTxBuf) < 256 THEN    M_BuildConnectPacket := FALSE;    RETURN;END_IFIF (sPassword <> '') AND (sUsername = '') THEN    M_SetError(        uiErrorCode := TO_UINT(E_ReasonCode.uiErrInvalidParameter),        sMessage := 'Password requires username');    M_BuildConnectPacket := FALSE;    RETURN;END_IFIF (NOT bWillFlag) AND ((sWillTopic <> '') OR (sWillMessage <> '')) THEN    M_SetError(        uiErrorCode := TO_UINT(E_ReasonCode.uiErrInvalidParameter),        sMessage := 'Will fields require Will Flag');    M_BuildConnectPacket := FALSE;    RETURN;END_IFIF (bWillRetain OR (eWillQoS <> E_MqttQoS.byQoS0)) AND (NOT bWillFlag) THEN    M_SetError(        uiErrorCode := TO_UINT(E_ReasonCode.uiErrInvalidParameter),        sMessage := 'Invalid will flag combination');    M_BuildConnectPacket := FALSE;    RETURN;END_IFIF bWillFlag AND (sWillTopic = '') THEN    M_SetError(        uiErrorCode := TO_UINT(E_ReasonCode.uiErrInvalidParameter),        sMessage := 'Will topic is required');    M_BuildConnectPacket := FALSE;    RETURN;END_IF// 清空发送缓冲区,避免数据干扰FOR i := LOWER_BOUND(aTxBuf, 1) TO UPPER_BOUND(aTxBuf, 1) DO    aTxBuf[i] := 0;END_FOR/// =======================================================================/// 长度计算/// =======================================================================// 可变报头长度: 协议名(6) + 协议版本(1) + 连接标志(1) + 保持连接(2)uiVarHeaderLen := 10;// MQTT 5.0: 连接属性长度uiPropsLen := 0;IF eVersion = E_MqttVersion.byMqttVersion50 THEN    // Session Expiry Interval (4字节值 + 1字节标识符)    IF udiSessionExpiry > 0 THEN        uiPropsLen := uiPropsLen + 5;    END_IF    // Receive Maximum (2字节值 + 1字节标识符)    uiPropsLen := uiPropsLen + 3;    // Maximum Packet Size (4字节值 + 1字节标识符)    uiPropsLen := uiPropsLen + 5;    // Topic Alias Maximum (2字节值 + 1字节标识符)    uiPropsLen := uiPropsLen + 3;    // Request Problem Information (1字节值 + 1字节标识符)    uiPropsLen := uiPropsLen + 2;    // Request Response Information (1字节值 + 1字节标识符)    IF bRequestResponseInfo THEN        uiPropsLen := uiPropsLen + 2;    END_IF    // 属性长度字段的编码长度    IF uiPropsLen < 128 THEN        uiVarHeaderLen := uiVarHeaderLen + 1 + uiPropsLen;                  //1字节属性长度    ELSE        uiVarHeaderLen := uiVarHeaderLen + 2 + uiPropsLen;                  //2字节属性长度    END_IFEND_IF// 计算 CONNECT 载荷总长度uiPayloadLen := TO_UINT(2 + LEN(sClientID));                                    //Client IDIF sUsername <> '' THEN    uiPayloadLen := uiPayloadLen + TO_UINT(2 + LEN(sUsername));END_IFIF sPassword <> '' THEN    uiPayloadLen := uiPayloadLen + TO_UINT(2 + LEN(sPassword));END_IFIF bWillFlag THEN    IF eVersion = E_MqttVersion.byMqttVersion50 THEN        uiPayloadLen := uiPayloadLen + 1;    END_IF    uiPayloadLen := uiPayloadLen + TO_UINT(2 + LEN(sWillTopic) + 2 + LEN(sWillMessage));END_IFIF uiReceiveMax = 0 THEN    M_SetError(        uiErrorCode := TO_UINT(E_ReasonCode.uiErrInvalidParameter),        sMessage := 'Receive Maximum must not be zero');    M_BuildConnectPacket := FALSE;    RETURN;END_IFIF udMaxPacketSize = 0 THEN    M_SetError(        uiErrorCode := TO_UINT(E_ReasonCode.uiErrInvalidParameter),        sMessage := 'Maximum Packet Size must not be zero');    M_BuildConnectPacket := FALSE;    RETURN;END_IF// 计算 Remaining Length 值uiRemainingLen := uiVarHeaderLen + uiPayloadLen;// 缓冲区长度检查IF uiRemainingLen + 5 > SIZEOF(aTxBuf) THEN    M_BuildConnectPacket := FALSE;    RETURN;END_IF/// =======================================================================/// 创建报文/// =======================================================================uiPos := 0;// ****************** 固定报文头 = 报文类型 + Remaining Length ******************// 报文类型aTxBuf[0] := E_MqttPacketType.byConnect;uiPos := uiPos + 1;// Remaining Length 值uiPos := uiPos + M_EncodeRemainingLength(uiRemainingLen, ADR(aTxBuf[uiPos]));// ****************** 可变报文头 = 协议名(6) + 协议版本(1) + 连接标志(1) + 保持连接(2) ******************// 协议名:"MQTT"aTxBuf[uiPos] := 0;                                                         //长度高字节MSBuiPos := uiPos + 1;aTxBuf[uiPos] := 4;                                                         //长度低字节LSBuiPos := uiPos + 1;aTxBuf[uiPos] := 77;                                                            // 'M'uiPos := uiPos + 1;aTxBuf[uiPos] := 81;                                                            // 'Q'uiPos := uiPos + 1;aTxBuf[uiPos] := 84;                                                            // 'T'uiPos := uiPos + 1;aTxBuf[uiPos] := 84;                                                            // 'T'uiPos := uiPos + 1;// 协议版本(根据eVersion选择)IF eVersion = E_MqttVersion.byMqttVersion50 THEN    aTxBuf[uiPos] := 5;                                                     //MQTT 5.0ELSE    aTxBuf[uiPos] := 4;                                                     //MQTT 3.1.1END_IFuiPos := uiPos + 1;// 连接标志byConnectFlags :=        (BOOL_TO_BYTE(FALSE) * 1) OR                                            //Bit 0: Reserved。服务端必须验证保留标志位是否为0        (BOOL_TO_BYTE(bCleanSession) * 2) OR                                    //Bit 1: Clean Session        (BOOL_TO_BYTE(bWillFlag AND sWillTopic <> '') * 4) OR                   //Bit 2: Will Flag        (BOOL_TO_BYTE(eWillQoS = 1 OR eWillQoS = 3) * 8) OR                 //Bit 3: Will QoS低位        (BOOL_TO_BYTE(eWillQoS = 2 OR eWillQoS = 3) * 16) OR                    //Bit 4: Will QoS高位        (BOOL_TO_BYTE(bWillFlag AND bWillRetain) * 32) OR                       //Bit 5: Will Retain        (BOOL_TO_BYTE(sPassword <> '') * 64) OR                             //Bit 6: Password Flag        (BOOL_TO_BYTE(sUsername <> '') * 128);                                  //Bit 7: Username FlagaTxBuf[uiPos] := byConnectFlags;uiPos := uiPos + 1;// 保持连接aTxBuf[uiPos] := UINT_TO_BYTE(SHR(uiKeepAlive, 8));                         //KeepAlive高字节uiPos := uiPos + 1;aTxBuf[uiPos] := UINT_TO_BYTE(uiKeepAlive AND 16#FF);                           //KeepAlive低字节uiPos := uiPos + 1;// ****************** MQTT 5.0: 连接属性 ******************IF eVersion = E_MqttVersion.byMqttVersion50 THEN    // 属性长度(可变字节编码)    IF uiPropsLen < 128 THEN        aTxBuf[uiPos] := TO_BYTE(uiPropsLen);        uiPos := uiPos + 1;    ELSE        aTxBuf[uiPos] := TO_BYTE((uiPropsLen / 128) OR 16#80);        uiPos := uiPos + 1;        aTxBuf[uiPos] := TO_BYTE(uiPropsLen MOD 128);        uiPos := uiPos + 1;    END_IF    // Session Expiry Interval    IF udiSessionExpiry > 0 THEN        aTxBuf[uiPos] := GVL_Mqtt.cnPropSessionExpiry;  uiPos := uiPos + 1;        aTxBuf[uiPos] := UDINT_TO_BYTE(SHR(udiSessionExpiry, 24) AND 16#FF);    uiPos := uiPos + 1;        aTxBuf[uiPos] := UDINT_TO_BYTE(SHR(udiSessionExpiry, 16) AND 16#FF);    uiPos := uiPos + 1;        aTxBuf[uiPos] := UDINT_TO_BYTE(SHR(udiSessionExpiry, 8) AND 16#FF); uiPos := uiPos + 1;        aTxBuf[uiPos] := UDINT_TO_BYTE(udiSessionExpiry AND 16#FF);         uiPos := uiPos + 1;    END_IF    // Receive Maximum    aTxBuf[uiPos] := GVL_Mqtt.cnPropReceiveMaximum; uiPos := uiPos + 1;    aTxBuf[uiPos] := UINT_TO_BYTE(SHR(uiReceiveMax, 8));    uiPos := uiPos + 1;    aTxBuf[uiPos] := UINT_TO_BYTE(uiReceiveMax AND 16#FF);  uiPos := uiPos + 1;    // Maximum Packet Size    aTxBuf[uiPos] := GVL_Mqtt.cnPropMaxPacketSize;  uiPos := uiPos + 1;    aTxBuf[uiPos] := UDINT_TO_BYTE(SHR(udMaxPacketSize, 24) AND 16#FF); uiPos := uiPos + 1;    aTxBuf[uiPos] := UDINT_TO_BYTE(SHR(udMaxPacketSize, 16) AND 16#FF); uiPos := uiPos + 1;    aTxBuf[uiPos] := UDINT_TO_BYTE(SHR(udMaxPacketSize, 8) AND 16#FF);  uiPos := uiPos + 1;    aTxBuf[uiPos] := UDINT_TO_BYTE(udMaxPacketSize AND 16#FF);          uiPos := uiPos + 1;    // Topic Alias Maximum    uiTopicAliasMax := GVL_Mqtt.cnMaxTopicAlias;    aTxBuf[uiPos] := GVL_Mqtt.cnPropTopicAliasMax;  uiPos := uiPos + 1;    aTxBuf[uiPos] := UINT_TO_BYTE(SHR(uiTopicAliasMax, 8)); uiPos := uiPos + 1;    aTxBuf[uiPos] := UINT_TO_BYTE(uiTopicAliasMax AND 16#FF); uiPos := uiPos + 1;    // Request Problem Information    aTxBuf[uiPos] := GVL_Mqtt.cnPropRequestProblemInfo; uiPos := uiPos + 1;    aTxBuf[uiPos] := BOOL_TO_BYTE(bRequestProblemInfo);             uiPos := uiPos + 1;    // Request Response Information    IF bRequestResponseInfo THEN        aTxBuf[uiPos] := GVL_Mqtt.cnPropRequestResponseInfo; uiPos := uiPos + 1;        aTxBuf[uiPos] := 1; uiPos := uiPos + 1;    END_IFEND_IF// ****************** 载荷 = 客户端标识符 + 遗嘱主题 + 遗嘱消息 + 用户名 + 密码 ******************// 客户端标识符uiPos := uiPos + M_AppendString(sClientID, ADR(aTxBuf[uiPos]));// 遗嘱主题 + 遗嘱消息IF bWillFlag THEN    IF eVersion = E_MqttVersion.byMqttVersion50 THEN        aTxBuf[uiPos] := 0;        uiPos := uiPos + 1;    END_IF    uiPos := uiPos + M_AppendString(sWillTopic, ADR(aTxBuf[uiPos]));    uiPos := uiPos + M_AppendString(sWillMessage, ADR(aTxBuf[uiPos]));END_IF// 用户名IF sUsername <> '' THEN    uiPos := uiPos + M_AppendString(sUsername, ADR(aTxBuf[uiPos]));END_IF// 密码IF sPassword <> '' THEN    uiPos := uiPos + M_AppendString(sPassword, ADR(aTxBuf[uiPos]));END_IFuiTxLength := uiPos;                                                            //需要发送的报文总字节数M_BuildConnectPacket := TRUE;

完整代码 2:M_BuildPingReqPacket.st

这一段完整公开 M_BuildPingReqPacket.st。读代码时先看对象职责,再看状态、长度、错误和返回值,不要只抄几行赋值。

/// =======================================================================/// 名称      : M_BuildPingReqPacket/// 功能      : 构建 PINGREQ 发送报文/// 说明      : 生成 MQTT 心跳请求固定报文头并更新发送长度。/// 编程人员  : ControlRookie/// 时间      : 2026-05-05/// 版本      : V1.0/// ======================================================================={attribute 'hide_all_locals'}METHOD M_BuildPingReqPacket : BOOLVAR    uiPos               : UINT := 0;                                        // 当前写入发送缓冲区的位置偏移[byte]    i                   : DINT;                                             // 清空发送缓冲区时使用的循环索引END_VAR// === IMPLEMENTATION ===// BUG-10: 缓冲区溢出保护IF SIZEOF(aTxBuf) < 2 THEN    M_BuildPingReqPacket := FALSE;    RETURN;END_IF// 清空发送缓冲区,避免数据干扰FOR i := LOWER_BOUND(aTxBuf, 1) TO UPPER_BOUND(aTxBuf, 1) DO    aTxBuf[i] := 0;END_FOR/// =======================================================================/// 创建报文/// =======================================================================uiPos := 0;// ****************** 固定报文头 = 报文类型 + Remaining Length ******************// 报文类型aTxBuf[0] := E_MqttPacketType.byPingReq;uiPos := uiPos + 1;// Remaining Length: 0aTxBuf[uiPos] := 0;uiPos := uiPos + 1;uiTxLength := uiPos;                                                            //需要发送的报文总字节数M_BuildPingReqPacket := TRUE;;

完整代码 3:M_BuildDisconnectPacket.st

这一段完整公开 M_BuildDisconnectPacket.st。读代码时先看对象职责,再看状态、长度、错误和返回值,不要只抄几行赋值。

/// =======================================================================/// 名称      : M_BuildDisconnectPacket/// 功能      : 构建 DISCONNECT 发送报文/// 说明      : 根据 MQTT 版本生成断开连接报文并更新发送长度。/// 编程人员  : ControlRookie/// 时间      : 2026-05-05/// 版本      : V1.0/// ======================================================================={attribute 'hide_all_locals'}METHOD M_BuildDisconnectPacket : BOOLVAR    uiPos               : UINT := 0;                                        // 当前写入发送缓冲区的位置偏移[byte]    uiRemainingLen      : UINT;                                             // 写入固定报头中的 Remaining Length 值[byte]    i                   : DINT;                                             // 清空发送缓冲区时使用的循环索引END_VAR// === IMPLEMENTATION ===// BUG-10: 缓冲区溢出保护IF SIZEOF(aTxBuf) < 4 THEN    M_BuildDisconnectPacket := FALSE;    RETURN;END_IF// 清空发送缓冲区,避免数据干扰FOR i := LOWER_BOUND(aTxBuf, 1) TO UPPER_BOUND(aTxBuf, 1) DO    aTxBuf[i] := 0;END_FOR/// =======================================================================/// 创建报文/// =======================================================================uiPos := 0;IF eVersion = E_MqttVersion.byMqttVersion50 THEN    // MQTT 5.0: DISCONNECT包含Reason Code和属性    uiRemainingLen := 2;                                                        //Reason Code(1) + 属性长度(1)    // 报文类型    aTxBuf[0] := E_MqttPacketType.byDisconnect;    uiPos := 1;    // Remaining Length 值    uiPos := uiPos + M_EncodeRemainingLength(uiRemainingLen, ADR(aTxBuf[uiPos]));    // Reason Code: 0x00 = Normal disconnection    aTxBuf[uiPos] := 16#00;    uiPos := uiPos + 1;    // 属性长度: 0(无属性)    aTxBuf[uiPos] := 0;    uiPos := uiPos + 1;ELSE    // MQTT 3.1.1: 固定报文头(2字节)    aTxBuf[0] := E_MqttPacketType.byDisconnect;    // BUG-02修复: 原代码 aTxBuf[0] := 0 覆盖了报文类型,改为 aTxBuf[1] := 0    aTxBuf[1] := 0;    uiPos := 2;END_IFuiTxLength := uiPos;                                                            //需要发送的报文总字节数M_BuildDisconnectPacket := TRUE;;

完整代码 4:M_HandleConnAck.st

这一段完整公开 M_HandleConnAck.st。读代码时先看对象职责,再看状态、长度、错误和返回值,不要只抄几行赋值。

/// =======================================================================/// 名称      : M_HandleConnAck/// 功能      : 处理 CONNACK 接收报文/// 说明      : 解析连接确认结果与服务器属性,成功后清除等待确认状态。/// 编程人员  : ControlRookie/// 时间      : 2026-05-05/// 版本      : V1.1/// ======================================================================={attribute 'hide_all_locals'}METHOD M_HandleConnAck : BOOLVAR    byReasonCode            : BYTE;  // CONNACK 原因码    byConnAckFlags          : BYTE;  // CONNACK 标志位    uiPropsLen              : UINT;  // 属性总长度    uiPropertyHeaderBytes   : UINT;  // 解析 CONNACK 属性长度这个 VBI 字段实际消耗的字节数[byte]    uiPropsStart            : UINT;  // 属性起始偏移    uiPropsEnd              : UINT;  // 属性结束偏移    uiPos                   : UINT;  // 当前解析偏移    byPropId                : BYTE;  // 当前属性标识符    uiLoopGuard             : UINT;  // 解析属性列表时的循环保护计数器,防止异常报文卡死状态机END_VAR// === IMPLEMENTATION ===/// 先检查当前缓冲区里是否至少有一帧最小 CONNACK,/// 并确认我们现在确实处在“等待 CONNACK”这个会话阶段。IF uiRxLength < 4 THEN    M_HandleConnAck := FALSE;    RETURN;END_IFIF NOT xWaitingForAck OR aRxBuf[0] <> byExpectedMsgType THEN    M_HandleConnAck := FALSE;    RETURN;END_IFbyConnAckFlags := aRxBuf[2];/// CONNACK 标志位除 Session Present 外其余位都必须为 0。IF (byConnAckFlags AND 16#FE) <> 0 THEN    sDiagMsg := 'Invalid CONNACK flags';    M_HandleConnAck := FALSE;    RETURN;END_IFIF eVersion = E_MqttVersion.byMqttVersion50 THEN    byReasonCode := aRxBuf[3];ELSE    byReasonCode := aRxBuf[3];END_IF/// 原因码非 0 代表建连被拒绝。/// 这里不仅要翻译成可读诊断,还要消费掉当前失败 CONNACK,/// 避免下一次重新连接时又把旧失败报文重复判一次。IF byReasonCode <> 0 THEN    IF eVersion = E_MqttVersion.byMqttVersion50 THEN        CASE byReasonCode OF            E_MqttReasonCode.byUnsupportedProtocolVersion: sDiagMsg := 'Protocol version not supported';            E_MqttReasonCode.byClientIdentifierInvalid: sDiagMsg := 'Client identifier invalid';            E_MqttReasonCode.byBadUserNamePassword: sDiagMsg := 'Bad username or password';            E_MqttReasonCode.byNotAuthorized: sDiagMsg := 'Not authorized';            E_MqttReasonCode.byServerUnavailable: sDiagMsg := 'Server unavailable';            E_MqttReasonCode.byServerBusy: sDiagMsg := 'Server busy';            E_MqttReasonCode.byBanned: sDiagMsg := 'Banned';            E_MqttReasonCode.byServerShuttingDown: sDiagMsg := 'Server shutting down';            E_MqttReasonCode.byBadAuthMethod: sDiagMsg := 'Bad authentication method';            E_MqttReasonCode.byQuotaExceeded: sDiagMsg := 'Quota exceeded';            E_MqttReasonCode.byRetainNotSupported: sDiagMsg := 'Retain not supported';            E_MqttReasonCode.byQosNotSupported: sDiagMsg := 'QoS not supported';            E_MqttReasonCode.byUseAnotherServer: sDiagMsg := 'Use another server';            E_MqttReasonCode.byServerMoved: sDiagMsg := 'Server moved';        ELSE            sDiagMsg := CONCAT('CONNACK reason code: ', BYTE_TO_STRING(byReasonCode));        END_CASE    ELSE        CASE byReasonCode OF            1: sDiagMsg := 'Unacceptable protocol version';            2: sDiagMsg := 'Identifier rejected';            3: sDiagMsg := 'Server unavailable';            4: sDiagMsg := 'Bad username or password';            5: sDiagMsg := 'Not authorized';        ELSE            sDiagMsg := CONCAT('CONNACK return code: ', BYTE_TO_STRING(byReasonCode));        END_CASE    END_IF    uiRxLength := 0;    xWaitingForAck := FALSE;    M_HandleConnAck := FALSE;    RETURN;END_IF/// 先把服务端能力恢复到“默认全开”假设,/// 再根据 MQTT 5.0 属性逐项覆盖,这样缺省属性也能得到合理值。uiServerReceiveMax := GVL_Mqtt.cnDefaultReceiveMax;byServerMaxQoS := TO_BYTE(E_MqttQoS.byQoS2);bServerRetainAvailable := TRUE;udServerMaxPacketSize := GVL_Mqtt.cnMaxPacketSize;uiServerTopicAliasMax := 0;uiSendQuota := GVL_Mqtt.cnDefaultReceiveMax;bServerWildcardSubAvail := TRUE;bServerSubIdAvail := TRUE;bServerSharedSubAvail := TRUE;IF eVersion = E_MqttVersion.byMqttVersion50 AND uiRxLength >= 5 THEN    uiPos := 4;    uiPropertyHeaderBytes := M_DecodeRemainingLength(pBuffer := ADR(aRxBuf[uiPos]), uiLength => uiPropsLen);    IF uiPropertyHeaderBytes = 0 THEN        M_HandleConnAck := FALSE;        RETURN;    END_IF    uiPos := uiPos + uiPropertyHeaderBytes;    uiPropsStart := uiPos;    uiPropsEnd := uiPropsStart + uiPropsLen;    IF uiPropsEnd > uiRxLength THEN        M_HandleConnAck := FALSE;        RETURN;    END_IF    uiLoopGuard := 0;    /// 这里只解析当前客户端已经真正使用的 CONNACK 属性,    /// 例如 Receive Maximum、Max QoS、Retain Available、Topic Alias 等。    WHILE uiPos < uiPropsEnd AND uiPos < uiRxLength AND uiLoopGuard < GVL_Mqtt.cnMaxPropertyLoop DO        byPropId := aRxBuf[uiPos];        uiPos := uiPos + 1;        CASE byPropId OF            GVL_Mqtt.cnPropReceiveMaximum:                IF uiPos + 1 >= uiPropsEnd THEN                    M_HandleConnAck := FALSE;                    RETURN;                END_IF                uiServerReceiveMax := SHL(BYTE_TO_UINT(aRxBuf[uiPos]), 8) OR BYTE_TO_UINT(aRxBuf[uiPos + 1]);                IF uiServerReceiveMax = 0 THEN                    sDiagMsg := 'Receive Maximum must not be zero';                    M_HandleConnAck := FALSE;                    RETURN;                END_IF                uiPos := uiPos + 2;            GVL_Mqtt.cnPropMaxQoS:                IF uiPos >= uiPropsEnd THEN                    M_HandleConnAck := FALSE;                    RETURN;                END_IF                byServerMaxQoS := aRxBuf[uiPos];                IF byServerMaxQoS > TO_BYTE(E_MqttQoS.byQoS2) THEN                    sDiagMsg := 'Server Max QoS is invalid';                    M_HandleConnAck := FALSE;                    RETURN;                END_IF                uiPos := uiPos + 1;            GVL_Mqtt.cnPropRetainAvailable:                IF uiPos >= uiPropsEnd THEN                    M_HandleConnAck := FALSE;                    RETURN;                END_IF                bServerRetainAvailable := BYTE_TO_BOOL(aRxBuf[uiPos]);                uiPos := uiPos + 1;            GVL_Mqtt.cnPropMaxPacketSize:                IF uiPos + 3 >= uiPropsEnd THEN                    M_HandleConnAck := FALSE;                    RETURN;                END_IF                udServerMaxPacketSize := SHL(BYTE_TO_UDINT(aRxBuf[uiPos]), 24) OR SHL(BYTE_TO_UDINT(aRxBuf[uiPos + 1]), 16) OR SHL(BYTE_TO_UDINT(aRxBuf[uiPos + 2]), 8) OR BYTE_TO_UDINT(aRxBuf[uiPos + 3]);                IF udServerMaxPacketSize = 0 THEN                    sDiagMsg := 'Server Maximum Packet Size must not be zero';                    M_HandleConnAck := FALSE;                    RETURN;                END_IF                uiPos := uiPos + 4;            GVL_Mqtt.cnPropTopicAliasMax:                IF uiPos + 1 >= uiPropsEnd THEN                    M_HandleConnAck := FALSE;                    RETURN;                END_IF                uiServerTopicAliasMax := SHL(BYTE_TO_UINT(aRxBuf[uiPos]), 8) OR BYTE_TO_UINT(aRxBuf[uiPos + 1]);                uiPos := uiPos + 2;            GVL_Mqtt.cnPropReasonString:                IF uiPos + 1 >= uiPropsEnd THEN                    M_HandleConnAck := FALSE;                    RETURN;                END_IF                uiPos := uiPos + 2 + (SHL(BYTE_TO_UINT(aRxBuf[uiPos]), 8) OR BYTE_TO_UINT(aRxBuf[uiPos + 1]));            GVL_Mqtt.cnPropServerKeepAlive:                IF uiPos + 1 >= uiPropsEnd THEN                    M_HandleConnAck := FALSE;                    RETURN;                END_IF                uiKeepAlive := SHL(BYTE_TO_UINT(aRxBuf[uiPos]), 8) OR BYTE_TO_UINT(aRxBuf[uiPos + 1]);                uiPos := uiPos + 2;            GVL_Mqtt.cnPropAssignedClientId:                IF uiPos + 1 >= uiPropsEnd THEN                    M_HandleConnAck := FALSE;                    RETURN;                END_IF                uiPos := uiPos + 2 + (SHL(BYTE_TO_UINT(aRxBuf[uiPos]), 8) OR BYTE_TO_UINT(aRxBuf[uiPos + 1]));            GVL_Mqtt.cnPropAuthMethod,            GVL_Mqtt.cnPropAuthData,            GVL_Mqtt.cnPropServerReference:                IF uiPos + 1 >= uiPropsEnd THEN                    M_HandleConnAck := FALSE;                    RETURN;                END_IF                uiPos := uiPos + 2 + (SHL(BYTE_TO_UINT(aRxBuf[uiPos]), 8) OR BYTE_TO_UINT(aRxBuf[uiPos + 1]));            GVL_Mqtt.cnPropResponseInfo:                IF uiPos + 1 >= uiPropsEnd THEN                    M_HandleConnAck := FALSE;                    RETURN;                END_IF                uiPos := uiPos + 2 + (SHL(BYTE_TO_UINT(aRxBuf[uiPos]), 8) OR BYTE_TO_UINT(aRxBuf[uiPos + 1]));            GVL_Mqtt.cnPropUserProperty:                IF uiPos + 1 >= uiPropsEnd THEN                    M_HandleConnAck := FALSE;                    RETURN;                END_IF                uiPos := uiPos + 2 + (SHL(BYTE_TO_UINT(aRxBuf[uiPos]), 8) OR BYTE_TO_UINT(aRxBuf[uiPos + 1]));                IF uiPos + 1 >= uiPropsEnd THEN                    M_HandleConnAck := FALSE;                    RETURN;                END_IF                uiPos := uiPos + 2 + (SHL(BYTE_TO_UINT(aRxBuf[uiPos]), 8) OR BYTE_TO_UINT(aRxBuf[uiPos + 1]));            GVL_Mqtt.cnPropWildcardSubAvail,            GVL_Mqtt.cnPropSubIdAvail,            GVL_Mqtt.cnPropSharedSubAvail:                IF uiPos >= uiPropsEnd THEN                    M_HandleConnAck := FALSE;                    RETURN;                END_IF                CASE byPropId OF                    GVL_Mqtt.cnPropWildcardSubAvail:                        bServerWildcardSubAvail := BYTE_TO_BOOL(aRxBuf[uiPos]);                    GVL_Mqtt.cnPropSubIdAvail:                        bServerSubIdAvail := BYTE_TO_BOOL(aRxBuf[uiPos]);                    GVL_Mqtt.cnPropSharedSubAvail:                        bServerSharedSubAvail := BYTE_TO_BOOL(aRxBuf[uiPos]);                END_CASE                uiPos := uiPos + 1;            GVL_Mqtt.cnPropRequestProblemInfo,            GVL_Mqtt.cnPropRequestResponseInfo:                IF uiPos >= uiPropsEnd THEN                    M_HandleConnAck := FALSE;                    RETURN;                END_IF                uiPos := uiPos + 1;        ELSE            M_HandleConnAck := FALSE;            RETURN;        END_CASE        uiLoopGuard := uiLoopGuard + 1;    END_WHILEEND_IF/// 到这里说明本帧 CONNACK 已完整消费,可以清掉等待态和接收缓存。uiRxLength := 0;xWaitingForAck := FALSE;/// MQTT 5.0 下,服务端的 Receive Maximum 决定了我们允许并发挂起多少条出站 QoS>0 消息。IF eVersion = E_MqttVersion.byMqttVersion50 THEN    uiSendQuota := uiServerReceiveMax;END_IF/// Session Present / Clean Session 共同决定“这次连上后是否需要自动补订本地订阅意图”。IF NOT bCleanSession THEN    IF (byConnAckFlags AND GVL_Mqtt.cnConnAckSessionPresent) = 0 THEN        M_InflightClear();        IF uiSubscriptionCount > 0 THEN            bRestoreSubscriptions := TRUE;            uiRestoreSubscriptionIndex := 0;        ELSE            bRestoreSubscriptions := FALSE;        END_IF        sActiveSubTopic := '';        eActiveSubQoS := E_MqttQoS.byQoS0;        udiActiveSubscriptionId := 0;        bActiveSubscribeRestore := FALSE;    ELSE        bRestoreSubscriptions := FALSE;        uiRestoreSubscriptionIndex := 0;    END_IFELSE    M_InflightClear();    IF uiSubscriptionCount > 0 THEN        bRestoreSubscriptions := TRUE;        uiRestoreSubscriptionIndex := 0;    ELSE        bRestoreSubscriptions := FALSE;        uiRestoreSubscriptionIndex := 0;    END_IF    sActiveSubTopic := '';    eActiveSubQoS := E_MqttQoS.byQoS0;    udiActiveSubscriptionId := 0;    bActiveSubscribeRestore := FALSE;END_IF/// Topic Alias 映射属于会话态缓存,连上新会话后统一从空表重新建立。M_TopicAliasClear();M_HandleConnAck := TRUE;

完整代码 5:M_HandlePingResp.st

这一段完整公开 M_HandlePingResp.st。读代码时先看对象职责,再看状态、长度、错误和返回值,不要只抄几行赋值。

/// =======================================================================/// 名称      : M_HandlePingResp/// 功能      : 兼容桩方法,保留 PINGRESP 旧接口/// 说明      : 该方法已废弃,实际处理逻辑已迁移至 M_ProcessReceive,仅为兼容旧调用保留。/// 编程人员  : ControlRookie/// 时间      : 2026-05-05/// 版本      : V1.0/// ======================================================================={attribute 'hide_all_locals'}METHOD M_HandlePingResp : BOOL// === IMPLEMENTATION ===// 逻辑已移至M_ProcessReceiveM_HandlePingResp := FALSE;

这一篇你最该记住的几句话

  1. 源码加更不是片段展示,而是完整源码对象公开讲解。
  2. 先建立对象地图,再读状态、报文和事务,现场调试才不会迷路。
  3. 判断源码成熟度,不只看功能是否实现,还要看边界、错误和在线观测量是否闭环。

系列导航

  • 系列定位:MqttClient 系列教程,源码加更阶段,第 14 篇 / 共 16 篇
  • 上一篇:源码加更03
  • 下一篇:源码加更05