#

vincentlu

2026-03-25 40bd839e334399764c9a822373cc685c6bd005c2

#

 zy-acs-common/src/main/java/com/zy/acs/common/hk/HkConnectionMessage.java

New file
@@ -0,0 +1,9 @@
package com.zy.acs.common.hk;

import lombok.Data;

@Data
public class HkConnectionMessage extends HkMessageHeader {

    private HkConnectionStateType connectionState;
}

 zy-acs-common/src/main/java/com/zy/acs/common/hk/HkConnectionStateType.java

New file
@@ -0,0 +1,10 @@
package com.zy.acs.common.hk;

public enum HkConnectionStateType {

    ONLINE,
    OFFLINE,
    CONNECTIONBROKEN,
    ;

}

 zy-acs-common/src/main/java/com/zy/acs/common/hk/HkMessageHeader.java

New file
@@ -0,0 +1,17 @@
package com.zy.acs.common.hk;

import lombok.Data;

@Data
public class HkMessageHeader {

    private Long headerId;

    private String timestamp;

    private String version;

    private String manufacturer;

    private String serialNumber;
}

 zy-acs-hk/zy-acs-hk-latent/src/main/java/com/zy/acs/hk/latent/mqtt/handler/ConnectionMessageHandler.java

@@ -1,12 +1,16 @@
package com.zy.acs.hk.latent.mqtt.handler;

import com.zy.acs.common.hk.HkConnectionMessage;
import com.zy.acs.common.hk.HkConnectionStateType;
import com.zy.acs.common.constant.RedisConstant;
import com.zy.acs.common.utils.GsonUtils;
import com.zy.acs.common.utils.RedisSupport;
import com.zy.acs.hk.latent.mqtt.HkMessageHandler;
import com.zy.acs.hk.latent.mqtt.type.HkSubTopicType;
import com.zy.acs.hk.latent.mqtt.HkTopicInfo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

@Slf4j
@Component
@@ -21,10 +25,39 @@

    @Override
    public void handle(HkTopicInfo topicInfo, String payload) {
        log.info("handle connection message, agvNo={}, topic={}, payload={}",
                topicInfo.getAgvNo(), topicInfo.getRawTopic(), payload);
        HkConnectionMessage connectionMessage;
        try {
            connectionMessage = GsonUtils.fromJson(payload, HkConnectionMessage.class);
        } catch (Exception e) {
            log.error("parse connection message failed, agvNo={}, topic={}, payload={}",
                    topicInfo.getAgvNo(), topicInfo.getRawTopic(), payload, e);
            return;
        }

        redis.setObject(RedisConstant.HK_AGV_ONLINE_FLAG, topicInfo.getAgvNo(), 1, 30);
        if (connectionMessage == null || connectionMessage.getConnectionState() == null) {
            log.warn("ignore connection message, connectionState is empty, agvNo={}, topic={}, payload={}",
                    topicInfo.getAgvNo(), topicInfo.getRawTopic(), payload);
            return;
        }

        if (StringUtils.hasText(connectionMessage.getSerialNumber())
                && !topicInfo.getAgvNo().equals(connectionMessage.getSerialNumber())) {
            log.warn("connection message serialNumber mismatch, topicAgvNo={}, bodySerialNumber={}, topic={}",
                    topicInfo.getAgvNo(), connectionMessage.getSerialNumber(), topicInfo.getRawTopic());
        }

        if (HkConnectionStateType.ONLINE == connectionMessage.getConnectionState()) {
            redis.setObject(RedisConstant.HK_AGV_ONLINE_FLAG, topicInfo.getAgvNo(), 1, 30);
        } else {
            redis.deleteObject(RedisConstant.HK_AGV_ONLINE_FLAG, topicInfo.getAgvNo());
        }

        log.info("handle connection message, agvNo={}, connectionState={}, headerId={}, timestamp={}, topic={}",
                topicInfo.getAgvNo(),
                connectionMessage.getConnectionState(),
                connectionMessage.getHeaderId(),
                connectionMessage.getTimestamp(),
                topicInfo.getRawTopic());
    }

}