第5篇_PUBLISH 不是收到就转发：Broker 怎么处理 QoS、PacketId 和多客户端 fanout

PUBLISH PacketId = 10

// 发布者 PacketId 只用于发布者连接的 ACK。
uiSourcePacketId := stPublishIn.uiPacketId;

// fanout 到订阅者时，必须按目标连接重新分配 PacketId。
uiTargetPacketId := aConnections[uiTargetSlot].M_AssignPacketId();

aConnections[uiTargetSlot].M_EnqueueDelivery(
    sTopicName := stPublishIn.sTopicName,
    pPayload := stPublishIn.pPayload,
    uiPayloadLen := stPublishIn.uiPayloadLen,
    eQoS := eTargetQoS,
    uiPacketId := uiTargetPacketId);

IF stPublish.eQoS = E_MqttQoS.byQoS1 THEN
    IF fbRxScheduler.M_RecordPublish(stPublish := stPublish, udiNowMs := udiNowMs) THEN
        IF (uiSourceSlot >= 1) AND (uiSourceSlot <= GVL_MqttBroker.cnMaxClientSlots) THEN
            aConnections[uiSourceSlot].M_EnqueueProtocolAck(
                ePacketType := E_MqttPacketType.byPubAck,
                uiPacketId := stPublish.uiPacketId,
                byReturnCode := 0);
            fbRxScheduler.M_CompletePublish(uiSlot := uiSourceSlot, uiPacketId := stPublish.uiPacketId);
            stMetrics.udiPubAckSent := stMetrics.udiPubAckSent + 1;
        END_IF
    END_IF
END_IF

IF stPublish.eQoS = E_MqttQoS.byQoS2 THEN
    IF fbRxScheduler.M_RecordPublish(stPublish := stPublish, udiNowMs := udiNowMs) THEN
        IF (uiSourceSlot >= 1) AND (uiSourceSlot <= GVL_MqttBroker.cnMaxClientSlots) THEN
            aConnections[uiSourceSlot].M_EnqueueProtocolAck(
                ePacketType := E_MqttPacketType.byPubRec,
                uiPacketId := stPublish.uiPacketId,
                byReturnCode := 0);
            stMetrics.udiPubRecSent := stMetrics.udiPubRecSent + 1;
        END_IF
    END_IF
    M_HandlePublish := TRUE;
    RETURN;
END_IF

M_RoutePublishNow(stPublish := stPublish);

IF fbRouter.M_FindNextRoute(
    stDelivery := stRouteFrame,
    stPublish := stPublish,
    xRestart := TRUE,
    xFound => xRouteFound) THEN
    WHILE xRouteFound AND (uiRouteCount < GVL_MqttBroker.cnMaxRouteFanoutPerScan) DO
        IF (stRouteFrame.uiTargetSlot >= 1) AND (stRouteFrame.uiTargetSlot <= GVL_MqttBroker.cnMaxClientSlots) THEN
            IF aConnections[stRouteFrame.uiTargetSlot].M_EnqueueDelivery(stDelivery := stRouteFrame) THEN
                stMetrics.udiPublishDelivered := stMetrics.udiPublishDelivered + 1;
                uiRouteCount := uiRouteCount + 1;

                IF stRouteFrame.eQoS <> E_MqttQoS.byQoS0 THEN
                    // M_EnqueueDelivery 会为首次 QoS>0 投递分配“目标连接内”的 PacketId。
                    // 事务表必须登记这个已分配后的 PacketId，后续订阅者 PUBACK 才能正确清理事务。
                    fbTxScheduler.M_RegisterPublish(stPublish := stRouteFrame, udiNowMs := udiNowMs);
                END_IF
            ELSE
                stMetrics.udiPublishDropped := stMetrics.udiPublishDropped + 1;
            END_IF
        END_IF

        fbRouter.M_FindNextRoute(
            stDelivery := stRouteFrame,
            stPublish := stPublish,
            xRestart := FALSE,
            xFound => xRouteFound);
    END_WHILE
END_IF

IF stDelivery.eQoS = E_MqttQoS.byQoS0 THEN
    stDelivery.uiPacketId := 0;
ELSIF stDelivery.uiPacketId = 0 THEN
    // Packet Identifier 的作用域是单条 MQTT 客户端连接。
    // 入站 PUBLISH 的 PacketId 属于发布者连接，不能直接转发给订阅者；
    // 首次投递给订阅者时必须在自己的连接内重新分配 PacketId，避免 QoS1/QoS2 确认链路冲突。
    M_AssignPacketId(stPublish := stDelivery);
ELSE
    // QoS 重发必须沿用首次投递的 PacketId。
    // 如果重发时重新分配 PacketId，客户端会 PUBACK 新编号，而事务表仍等待旧编号，
    // 结果就是每 2 秒重复投递一次，达到最大重试次数后 Broker 主动断开客户端。
END_IF