调整KFKA数据推送测试。

jiazx0107@163.com

2024-01-01 d454e51e06dbf7beb50551588fc58a5810295d7c

 src/main/java/com/fzzy/gateway/hx2023/service/DeviceReportServiceImpl.java

@@ -2,6 +2,7 @@

import com.alibaba.fastjson2.JSONObject;
import com.fzzy.api.data.PushProtocol;
import com.fzzy.data.ConfigData;
import com.fzzy.gateway.api.GatewayDeviceReportService;
import com.fzzy.gateway.data.BaseReqData;
import com.fzzy.gateway.data.BaseResp;
@@ -13,7 +14,9 @@
import com.fzzy.gateway.hx2023.data.WeightInfo;
import com.fzzy.gateway.hx2023.kafka.KafkaDeviceReportService;
import com.fzzy.mqtt.MqttGatewayService;
import jdk.nashorn.internal.runtime.regexp.joni.Config;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
@@ -26,6 +29,8 @@
    private KafkaDeviceReportService kafkaDeviceReportService;
    @Resource
    private MqttGatewayService publishService;
    @Resource
    private ConfigData configData;

    @Override
    public String getProtocol() {
@@ -37,6 +42,16 @@

        String topic = ScConstant.TOPIC_REPORT;
        topic = topic.replace("${productId}", reqData.getProductId()).replace("${deviceId}", reqData.getDeviceId());

        //如果是测试模式不执行推送
        if (configData.getActive().indexOf("dev") >= 0) {

            log.info("----------------------------推送MQTT粮情信息，注：调试模式不推送---------------------------");
            log.info("-----TOPIC-----{}", topic);
            log.info("-----Message-----{}", reqData.getData());

            return new BaseResp();
        }

        publishService.publishMqttWithTopic(reqData.getData(), topic);

@@ -97,7 +112,7 @@

        GatewayDevice device = reqData.getDevice();

        if (null == reqData.getData()) {
        if (StringUtils.isEmpty(reqData.getData())) {
            WebSocketPacket packet = new WebSocketPacket();
            WebSocketPacketHeader header = new WebSocketPacketHeader();
            header.setDeviceName(reqData.getDeviceName());
@@ -116,6 +131,8 @@
            jsonObject.put("position", device.getPosition());
            packet.setProperties(jsonObject);
            packet.setTimestamp(System.currentTimeMillis());

            reqData.setData(JSONObject.toJSONString(packet));
        }

        publishService.publishMqttWithTopic(reqData.getData(), topic);
@@ -129,6 +146,16 @@
    @Override
    public BaseResp reportGrainDataByKafka(BaseReqData reqData) {
        String topic = ScConstant.TOPIC_MESSAGE_REPORT;
        //如果是测试模式不执行推送
        if (configData.getActive().indexOf("dev") >= 0) {

            log.info("----------------------------推送KAFKA粮情信息，注：调试模式不推送---------------------------");
            log.info("-----TOPIC-----{}", topic);
            log.info("-----Message-----{}", reqData.getData());

            return new BaseResp();
        }

        kafkaDeviceReportService.publishWithTopic(reqData.getData(), topic);
        return new BaseResp();
    }