重新调整代码结构和实现方式

jiazx0107@163.com

2023-11-10 d52795fc5de0b6ed748cd2ef217dcd1371e4b8e9

重新调整代码结构和实现方式

 src/main/java/com/fzzy/gateway/api/DeviceReportService.java

ÎļþÒÑɾ³ý

 src/main/java/com/fzzy/gateway/api/GatewayDeviceReportService.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,42 @@
package com.fzzy.gateway.api;

import com.fzzy.gateway.data.BaseReqData;
import com.fzzy.gateway.data.BaseResp;
import com.fzzy.gateway.entity.GatewayDevice;
import com.fzzy.gateway.hx2023.data.KafaGrainData;


/**
 * ä¸ŠæŠ¥æŽ¥å£
 */
public interface GatewayDeviceReportService {

    /**
     * åè®®å®šä¹‰
     *
     * @return
     */
    String getProtocol();


    /**
     * é€šè¿‡è®¾å¤‡åºåˆ—号返回结果信息
     *
     * @param weigh
     * @param device è®¾å¤‡åºåˆ—号
     */
    String report2GatewayBySn(double weigh, GatewayDevice device);


    /**
     * æŽ¨é€ç²®æƒ…信息
     * @param reqData
     */
    BaseResp reportGrainData(BaseReqData reqData);

    BaseResp reportWeightData(BaseReqData reqData);

    BaseResp reportLprData(BaseReqData reqData);

    BaseResp reportGrainDataByKafka(BaseReqData reqData);
}

 src/main/java/com/fzzy/gateway/api/GatewayDeviceTestService.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,26 @@
package com.fzzy.gateway.api;

import com.fzzy.gateway.data.BaseResp;
import com.fzzy.gateway.data.BaseReqData;


/**
 * ç½‘关设备测试接口
 */
public interface GatewayDeviceTestService {

    /**
     * åè®®å®šä¹‰
     *
     * @return
     */
    String getProtocol();

    BaseResp testGrain(BaseReqData reqData);

    BaseResp testGrainKafka(BaseReqData reqData);

    BaseResp testWeight(BaseReqData reqData);

    BaseResp testLpr(BaseReqData reqData);
}

 src/main/java/com/fzzy/gateway/api/GatewayRemoteManager.java

@@ -17,7 +17,7 @@

    public static Map<String, GatewayRemoteService> remoteMap = new HashMap<>();

    public static Map<String, DeviceReportService> reportMap = new HashMap<>();
    public static Map<String, GatewayDeviceReportService> reportMap = new HashMap<>();

    public static Map<String, GatewaySyncGranService> syncGrain = new HashMap<>();

@@ -28,6 +28,8 @@
    public static Map<String, GatewaySyncLprService> syncLpr = new HashMap<>();

    public static Map<String, GatewaySyncWeightService> syncWeight = new HashMap<>();

    public static Map<String, GatewayDeviceTestService> testMap = new HashMap<>();

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
@@ -62,9 +64,14 @@
            syncWeight.put(serviceMap6.get(key).getWeightProtocol(), serviceMap6.get(key));
        }

        Map<String, DeviceReportService> serviceMap7 = applicationContext.getBeansOfType(DeviceReportService.class);
        Map<String, GatewayDeviceReportService> serviceMap7 = applicationContext.getBeansOfType(GatewayDeviceReportService.class);
        for (String key : serviceMap7.keySet()) {
            reportMap.put(serviceMap7.get(key).getProvinceProtocol(), serviceMap7.get(key));
            reportMap.put(serviceMap7.get(key).getProtocol(), serviceMap7.get(key));
        }

        Map<String, GatewayDeviceTestService> serviceMap8 = applicationContext.getBeansOfType(GatewayDeviceTestService.class);
        for (String key : serviceMap8.keySet()) {
            testMap.put(serviceMap8.get(key).getProtocol(), serviceMap8.get(key));
        }
    }

@@ -136,8 +143,17 @@
     * @param protocol
     * @return
     */
    public DeviceReportService getDeviceReportService(String protocol) {
    public GatewayDeviceReportService getDeviceReportService(String protocol) {
        return reportMap.get(protocol);
    }


    /**
     * @param protocol
     * @return
     */
    public GatewayDeviceTestService getGatewayTestService(String protocol) {
        return testMap.get(protocol);
    }

}

 src/main/java/com/fzzy/gateway/api/GatewaySyncGranService.java

@@ -2,8 +2,8 @@


import com.fzzy.gateway.data.BaseResp;
import com.fzzy.gateway.data.BaseReqData;
import com.fzzy.gateway.hx2023.data.KafaGrainData;
import com.fzzy.gateway.hx2023.data.*;

/**
 * ç½‘关设备同步接口，根据业务需求定义接口
@@ -17,14 +17,6 @@
     */
    String getGrainProtocol();

    /**
     * åŒæ­¥ç²®æƒ…信息
     *
     * @param reqData
     * @return
     */
    public KafaGrainData syncGrain(SyncReqData reqData);


    /**
     * åŒæ­¥ç²®æƒ…信息返回JSON报文
@@ -32,5 +24,5 @@
     * @param reqData
     * @return
     */
    public BaseResp syncGrain2(SyncReqData reqData);
    public BaseResp syncGrain(BaseReqData reqData);
}

 src/main/java/com/fzzy/gateway/api/GatewaySyncIdCardService.java

@@ -1,8 +1,8 @@
package com.fzzy.gateway.api;


import com.fzzy.gateway.data.BaseReqData;
import com.fzzy.gateway.hx2023.data.IdCardData;
import com.fzzy.gateway.hx2023.data.*;

/**
 * ç½‘关设备同步接口，根据业务需求定义接口
@@ -21,7 +21,7 @@
     *
     * @return
     */
    public IdCardData syncIdCard(SyncReqData reqData);
    public IdCardData syncIdCard(BaseReqData reqData);




 src/main/java/com/fzzy/gateway/api/GatewaySyncLedService.java

@@ -1,7 +1,7 @@
package com.fzzy.gateway.api;


import com.fzzy.gateway.hx2023.data.*;
import com.fzzy.gateway.data.BaseReqData;

/**
 * ç½‘关设备同步接口，根据业务需求定义接口
@@ -26,7 +26,7 @@
     *
     * @return
     */
    public String push2Led(SyncReqData reqData);
    public String push2Led(BaseReqData reqData);




 src/main/java/com/fzzy/gateway/api/GatewaySyncLprService.java

@@ -1,6 +1,7 @@
package com.fzzy.gateway.api;


import com.fzzy.gateway.data.BaseReqData;
import com.fzzy.gateway.hx2023.data.*;

/**
@@ -21,6 +22,6 @@
     *
     * @return
     */
    public LprData syncLpr(SyncReqData reqData);
    public LprData syncLpr(BaseReqData reqData);

}

 src/main/java/com/fzzy/gateway/api/GatewaySyncWeightService.java

@@ -1,6 +1,7 @@
package com.fzzy.gateway.api;


import com.fzzy.gateway.data.BaseReqData;
import com.fzzy.gateway.hx2023.data.*;

/**
@@ -21,5 +22,5 @@
     *
     * @return
     */
    public WeightInfo syncWeightInfo(SyncReqData reqData);
    public WeightInfo syncWeightInfo(BaseReqData reqData);
}

 src/main/java/com/fzzy/gateway/controller/GatewayDeviceTestController.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,226 @@
package com.fzzy.gateway.controller;

import com.alibaba.fastjson2.JSONObject;
import com.bstek.dorado.annotation.Expose;
import com.fzzy.api.data.GatewayDeviceType;
import com.fzzy.api.utils.DateUtil;
import com.fzzy.async.fzzy40.Fzzy40CommonService;
import com.fzzy.async.fzzy40.entity.Fz40Grain;
import com.fzzy.gateway.GatewayUtils;
import com.fzzy.gateway.api.GatewayDeviceReportService;
import com.fzzy.gateway.api.GatewayRemoteManager;
import com.fzzy.gateway.data.BaseResp;
import com.fzzy.gateway.data.BaseReqData;
import com.fzzy.gateway.entity.GateWayTestParam;
import com.fzzy.gateway.entity.GatewayDevice;
import com.fzzy.gateway.hx2023.ScConstant;
import com.fzzy.gateway.hx2023.data.*;
import com.fzzy.gateway.service.GatewayDeviceService;
import com.fzzy.mqtt.MqttGatewayService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.time.DateFormatUtils;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

import javax.annotation.Resource;
import java.util.List;
import java.util.stream.Collectors;


/**
 *
 */
@Slf4j
@Controller
@RequestMapping("/gateway/test")
public class GatewayDeviceTestController {

    @Resource
    private GatewayDeviceService gatewayDeviceService;

    @Resource
    private GatewayRemoteManager gatewayRemoteManager;


    /**
     * æµ‹è¯•å…¥å£
     *
     * @param param
     * @return
     */
    @PostMapping("/deviceTest")
    public @ResponseBody
    String deviceTest(@RequestBody GateWayTestParam param) throws Exception {

        if (StringUtils.isEmpty(param.getBizType())) {
            return "ERROR:没有获取到业务类型，执行失败";
        }

        if ("testGrain".equals(param.getBizType())) {
            return testGrain(param);
        }

        if ("ajaxTestKafkaGrain".equals(param.getBizType())) {
            return ajaxTestKafkaGrain(param);
        }

        if ("ajaxTestWeight".equals(param.getBizType())) {
            return ajaxTestWeight(param);
        }

        if ("ajaxTestLpr".equals(param.getBizType())) {
            return ajaxTestLpr(param);
        }

        return "SUCCESS";
    }

    private String testGrain(GateWayTestParam param) {
        String deviceId = param.getDeviceId();

        GatewayDevice device = GatewayUtils.getCacheByDeviceId(deviceId);

        BaseReqData reqData = new BaseReqData();
        reqData.setDeviceId(device.getDeviceId());
        reqData.setProductId(device.getProductId());
        reqData.setDeviceName(device.getDeviceName());
        reqData.setDevice(device);
        reqData.setAutoReplay(true);

        if (!GatewayDeviceType.TYPE_07.getCode().equals(device.getType())) {
            return "ERROR：当前设备非粮情设备不支持当前操作";
        }

        if (StringUtils.isEmpty(device.getCableRule())) {
            return "ERROR：当前设备没有配置布线规则，无法执行";
        }

        BaseResp resp = gatewayRemoteManager.getGatewayTestService(device.getPushProtocol()).testGrain(reqData);
        //自动推送
        if (BaseResp.CODE_200 == resp.getCode() && reqData.isAutoReplay()) {
            reqData.setData(resp.getData());
            gatewayRemoteManager.getDeviceReportService(device.getPushProtocol()).reportGrainData(reqData);
        }
        return "SUCCESS：执行完成";
    }


    /**
     * ç²®æƒ…推送测试KAFKA方式
     *
     * @param param
     * @return
     */
    public String ajaxTestKafkaGrain(GateWayTestParam param) throws Exception {
        List<GatewayDevice> list = gatewayDeviceService.listAll();
        if (null == list || list.isEmpty()) {
            return "ERROR：为获取到系统中设备配置，取消执行";
        }

        BaseReqData reqData;
        BaseResp resp;
        for (GatewayDevice device : list) {
            reqData = new BaseReqData();
            reqData.setDeviceId(param.getDeviceId());
            reqData.setDayTime(param.getDayTime());
            reqData.setDevice(device);

            resp = gatewayRemoteManager.getGatewayTestService(device.getPushProtocol()).testGrainKafka(reqData);

            //自动推送
            if (BaseResp.CODE_200 == resp.getCode()) {
                reqData.setData(resp.getData());
                gatewayRemoteManager.getDeviceReportService(device.getPushProtocol()).reportGrainDataByKafka(reqData);
            }
        }

        return "SUCCESS";
    }

    /**
     * åœ°ç£…推送测试
     *
     * @param param
     * @return
     */
    public String ajaxTestWeight(GateWayTestParam param) throws Exception {

        double weight = param.getWeight();
        String deviceId = param.getDeviceId();

        List<GatewayDevice> list = gatewayDeviceService.listAll();
        if (list == null || list.size() <= 0) {
            return "ERROR：没有配置设备信息，执行失败";
        }

        List<GatewayDevice> weights = list.stream().filter(s -> (GatewayDeviceType.TYPE_01.getCode().equals(s.getType()))).collect(Collectors.toList());
        if (weights == null || weights.size() <= 0) {
            return "ERROR：ERROR：没有配置设备信息，执行失败";
        }

        BaseReqData reqData;
        BaseResp resp;
        for (GatewayDevice device : weights) {
            reqData = new BaseReqData();
            reqData.setDeviceId(device.getDeviceId());
            reqData.setProductId(device.getProductId());
            reqData.setDeviceName(device.getDeviceName());
            reqData.setDevice(device);
            reqData.setAutoReplay(true);
            reqData.setWeight(weight);
            resp = gatewayRemoteManager.getGatewayTestService(device.getPushProtocol()).testWeight(reqData);

            //自动推送
            if (BaseResp.CODE_200 == resp.getCode() && reqData.isAutoReplay()) {
                reqData.setData(resp.getData());
                gatewayRemoteManager.getDeviceReportService(device.getPushProtocol()).reportWeightData(reqData);
            }
        }
        return "SUCCESS";
    }

    /**
     * åœ°ç£…推送测试
     *
     * @return
     */
    public String ajaxTestLpr(GateWayTestParam param) throws Exception {

        String carNumber = param.getCarNumber();

        List<GatewayDevice> list = gatewayDeviceService.listAll();
        if (list == null || list.size() <= 0) {
            log.error("ERROR：没有配置设备信息，执行失败");
            return "ERROR：没有配置设备信息，执行失败";
        }
        List<GatewayDevice> weights = list.stream().filter(s -> (GatewayDeviceType.TYPE_02.getCode().equals(s.getType()))).collect(Collectors.toList());
        if (weights == null || weights.size() <= 0) {
            log.error("ERROR：没有配置设备信息，执行失败");
            return "ERROR：没有配置设备信息，执行失败";
        }

        BaseReqData reqData;
        BaseResp resp;
        for (GatewayDevice device : weights) {
            reqData = new BaseReqData();
            reqData.setDeviceId(device.getDeviceId());
            reqData.setProductId(device.getProductId());
            reqData.setDeviceName(device.getDeviceName());
            reqData.setDevice(device);
            reqData.setAutoReplay(true);
            reqData.setCarNumber(carNumber);
            resp = gatewayRemoteManager.getGatewayTestService(device.getPushProtocol()).testLpr(reqData);

            //自动推送
            if (BaseResp.CODE_200 == resp.getCode() && reqData.isAutoReplay()) {
                reqData.setData(resp.getData());
                gatewayRemoteManager.getDeviceReportService(device.getPushProtocol()).reportLprData(reqData);
            }
        }
        return "SUCCESS";
    }
}

 src/main/java/com/fzzy/gateway/data/BaseReqData.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,34 @@
package com.fzzy.gateway.data;

import com.fzzy.gateway.entity.GatewayDevice;
import lombok.Data;

import java.util.Date;

@Data
public class BaseReqData {

    private String productId;

    private String deviceId;

    private String deviceName;

    private String messageType;

    private String messageId;

    private String functionId;

    private GatewayDevice device;

    private boolean autoReplay;

    private double weight;

    private String carNumber;

    private Date dayTime;

    private String data;
}

 src/main/java/com/fzzy/gateway/data/BaseResp.java

@@ -5,10 +5,25 @@
@Data
public class BaseResp {

    private int code = 200;
    public static int CODE_200 = 200;
    public static int CODE_50 = 500;

    private int code = CODE_200;

    private String msg = "成功";


    private String data;


    public BaseResp() {
    }

    public BaseResp(String data) {
        this.data = data;
    }

    public BaseResp(int code,String msg) {
        this.msg = msg;
        this.code = code;
    }
}

 src/main/java/com/fzzy/gateway/entity/GatewayDevice.java

@@ -85,9 +85,9 @@
    @Column(name = "PUSH_PROTOCOL_", length = 20)
    private String pushProtocol;

    @PropertyDef(label = "省平台协议")
    @Column(name = "PROVINCE_PROTOCOL_", length = 20)
    private String provinceProtocol;
//    @PropertyDef(label = "省平台协议")
//    @Column(name = "PROVINCE_PROTOCOL_", length = 20)
//    private String provinceProtocol;

    @PropertyDef(label = "同步协议")
    @Column(name = "SYNC_PROTOCOL_", length = 20)

 src/main/java/com/fzzy/gateway/hx2023/ScConstant.java

@@ -25,8 +25,6 @@
    }


    public static int CODE_200 = 200;

    /**
     * ä¸‹å‘指令回复报文topic
     */

 src/main/java/com/fzzy/gateway/hx2023/controller/GatewayDeviceTestController.java

ÎļþÒÑɾ³ý

 src/main/java/com/fzzy/gateway/hx2023/data/SyncReqData.java

ÎļþÒÑɾ³ý

 src/main/java/com/fzzy/gateway/hx2023/kafka/KafkaDeviceReport.java

ÎļþÒÑɾ³ý

 src/main/java/com/fzzy/gateway/hx2023/kafka/KafkaDeviceReportService.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,64 @@
package com.fzzy.gateway.hx2023.kafka;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;


/**
 * ä½¿ç”¨KAFKA推动返回信息实现，针对当前网关
 */
@Slf4j
@Component
public class KafkaDeviceReportService {


    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;


    public String publishWithTopic(String sendData, String topic) {
        //推送信息
        kafkaTemplate.send(topic, sendData).addCallback(success -> {
            // æ¶ˆæ¯å‘送到的topic
            // String topic = success.getRecordMetadata().topic();
            // æ¶ˆæ¯å‘送到的分区
            int partition = success.getRecordMetadata().partition();
            // æ¶ˆæ¯åœ¨åˆ†åŒºå†…çš„offset
            long offset = success.getRecordMetadata().offset();

            log.info("---推送至KAFKA成功--:{}-{}-{}-{}", topic, partition, offset, sendData);
        }, failure -> {
            log.info("---推送至KAFKA失败--:{}-{}-{}-{}", topic, sendData);
            rePublishWithTopic(sendData, topic);
        });
        return "SUCCESS";
    }

    private void rePublishWithTopic(String sendData, String topic) {
        //推送信息
        kafkaTemplate.send(topic, sendData).addCallback(success -> {
            // æ¶ˆæ¯å‘送到的topic
            // String topic = success.getRecordMetadata().topic();
            // æ¶ˆæ¯å‘送到的分区
            int partition = success.getRecordMetadata().partition();
            // æ¶ˆæ¯åœ¨åˆ†åŒºå†…çš„offset
            long offset = success.getRecordMetadata().offset();

            log.info("---推送至KAFKA成功--:{}-{}-{}-{}", topic, partition, offset, sendData);
        }, failure -> {
            log.info("---推送至KAFKA失败--:{}-{}-{}-{}", topic, sendData);
        });
    }


    @KafkaListener(topics = {"TOPIC_ZLJ_GRAIN_TEMPERATURE"})
    public void onMessage1(ConsumerRecord<?, ?> record) {
        //  log.debug("获取消费信息：" + record.topic() + "-" + record.partition() + "-" + record.value());
    }

}

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

@@ -1,17 +1,13 @@
package com.fzzy.gateway.hx2023.service;

import com.alibaba.fastjson2.JSONObject;
import com.fzzy.api.data.GatewayProtocol;
import com.fzzy.api.data.PushProtocol;
import com.fzzy.gateway.api.DeviceReportService;
import com.fzzy.gateway.api.GatewayDeviceReportService;
import com.fzzy.gateway.data.BaseReqData;
import com.fzzy.gateway.data.BaseResp;
import com.fzzy.gateway.entity.GatewayDevice;
import com.fzzy.gateway.hx2023.ScConstant;
import com.fzzy.gateway.hx2023.data.KafaGrainData;
import com.fzzy.gateway.hx2023.data.WebSocketPacket;
import com.fzzy.gateway.hx2023.data.WebSocketPacketHeader;
import com.fzzy.gateway.hx2023.data.WeightInfo;
import com.fzzy.gateway.hx2023.kafka.KafkaDeviceReport;
import com.fzzy.gateway.hx2023.websocket.WebSocketDeviceReport;
import com.fzzy.gateway.hx2023.kafka.KafkaDeviceReportService;
import com.fzzy.mqtt.MqttGatewayService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

@@ -19,16 +15,15 @@

@Slf4j
@Component
public class DeviceReportServiceImpl implements DeviceReportService {

public class DeviceReportServiceImpl implements GatewayDeviceReportService {

    @Resource
    private WebSocketDeviceReport webSocketDeviceReport;
    private KafkaDeviceReportService kafkaDeviceReportService;
    @Resource
    private KafkaDeviceReport kafkaDeviceReport;
    private MqttGatewayService publishService;

    @Override
    public String getProvinceProtocol() {
    public String getProtocol() {
        return PushProtocol.GATEWAY_SC_2023.getCode();
    }

@@ -40,47 +35,57 @@
            return "ERROR:没有获取到设备配置信息";
        }

        //使用WEBSOCKET返回
        if (GatewayProtocol.GATE_WEBSOCKET.equals(device.getPushProtocol())) {

            WebSocketPacket packet = new WebSocketPacket();

            //productId的值在webSocket执行发送方法中补充
            WebSocketPacketHeader header = new WebSocketPacketHeader();
            header.setDeviceName(device.getDeviceName());
            header.setProductId("hx-weigh-big-01");
            //header.set
            packet.setHeaders(header);
            packet.setMessageType(ScConstant.MESSAGE_TYPE_REPORT_PROPERTY);
            packet.setDeviceId(device.getDeviceId());

            //设置信息主体
            WeightInfo weightInfo = new WeightInfo();
            weightInfo.setGrossWeight(weigh);
            weightInfo.setNetWeight(weigh);
            weightInfo.setNetWeight(weigh);
            weightInfo.setWeightUnit("KG");
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("weightInfo", JSONObject.toJSONString(weightInfo));

            packet.setProperties(jsonObject);

            packet.setTimestamp(System.currentTimeMillis());


            webSocketDeviceReport.sendByPacket(packet);
        }

        if (GatewayProtocol.GATE_MQTT.equals(device.getPushProtocol())) {
            //TODO ---->>>后期添加支持
        }

        return null;
    }

    @Override
    public String pushGrainData2Cloud(KafaGrainData data) {
    public BaseResp reportGrainData(BaseReqData reqData) {

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

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

        log.info("----------------------------推送MQTT粮情信息---------------------------");
        log.info("-----TOPIC-----{}", topic);
        log.info("-----Message-----{}", reqData.getData());

        return new BaseResp();
    }

    @Override
    public BaseResp reportWeightData(BaseReqData reqData) {

        String topic = ScConstant.TOPIC_MESSAGE_REPORT;

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

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

        log.info("----------------------------推送MQTT地磅信息---------------------------");
        log.info("-----TOPIC-----{}", topic);
        log.info("-----Message-----{}", reqData.getData());
        return new BaseResp();
    }

    @Override
    public BaseResp reportLprData(BaseReqData reqData) {
        String topic = ScConstant.TOPIC_MESSAGE_REPORT;
        topic = topic.replace("${productId}", reqData.getProductId()).replace("{deviceId}", reqData.getDeviceId());

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

        log.info("----------------------------推送MQTT车牌识别信息---------------------------");
        log.info("-----TOPIC-----{}", topic);
        log.info("-----Message-----{}", reqData.getData());
        return new BaseResp();
    }

    @Override
    public BaseResp reportGrainDataByKafka(BaseReqData reqData) {
        String topic = ScConstant.TOPIC_MESSAGE_REPORT;
        kafkaDeviceReportService.publishWithTopic(reqData.getData(), topic);
        return new BaseResp();
    }
}

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

@@ -3,7 +3,7 @@
import com.fzzy.api.data.GatewayDeviceProtocol;
import com.fzzy.gateway.api.GatewaySyncLprService;
import com.fzzy.gateway.hx2023.data.LprData;
import com.fzzy.gateway.hx2023.data.SyncReqData;
import com.fzzy.gateway.data.BaseReqData;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
@@ -23,7 +23,7 @@
    }

    @Override
    public LprData syncLpr(SyncReqData reqData) {
    public LprData syncLpr(BaseReqData reqData) {




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

@@ -4,6 +4,7 @@
import com.fzzy.api.data.PushProtocol;
import com.fzzy.api.utils.*;
import com.fzzy.api.view.repository.ApiLogRep;
import com.fzzy.gateway.data.BaseResp;
import com.fzzy.gateway.service.GatewayConfService;
import com.fzzy.gateway.util.GatewayHttpUtil;
import com.fzzy.gateway.api.GatewayRemoteService;
@@ -61,7 +62,7 @@

            CloudResp respKey = JSONObject.parseObject(jsonStr, CloudResp.class);

            if (ScConstant.CODE_200 == respKey.getCode()) {
            if (BaseResp.CODE_200 == respKey.getCode()) {
                JSONObject object = respKey.getData();
                String pubKey = (String) object.get("pubKey");
                String priKey = (String) object.get("priKey");
@@ -79,7 +80,7 @@
            log.info("---获取AccessToken接口-返回---{}", jsonStr);
            CloudResp respToken = JSONObject.parseObject(jsonStr, CloudResp.class);

            if (ScConstant.CODE_200 == respToken.getCode()) {
            if (BaseResp.CODE_200 == respToken.getCode()) {
                JSONObject object = respKey.getData();
                if (null != object) {
                    String token = (String) object.get("token");

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

@@ -1,22 +1,15 @@
package com.fzzy.gateway.hx2023.service;

import com.alibaba.fastjson2.JSONObject;
import com.fzzy.api.data.GatewayDeviceProtocol;
import com.fzzy.api.utils.NumberUtil;
import com.fzzy.gateway.api.GatewaySyncGranService;
import com.fzzy.gateway.data.BaseResp;
import com.fzzy.gateway.data.WeatherWebDto;
import com.fzzy.gateway.data.BaseReqData;
import com.fzzy.gateway.entity.GatewayDevice;
import com.fzzy.gateway.hx2023.ScConstant;
import com.fzzy.gateway.hx2023.data.*;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;

/**
 * ç½‘关与粮情分机通讯和解析
@@ -32,136 +25,13 @@
        return GatewayDeviceProtocol.DEVICE_TEST.getCode();
    }

    @Override
    public KafaGrainData syncGrain(SyncReqData reqData) {


        return null;
    }


    @Override
    public BaseResp syncGrain2(SyncReqData reqData) {
    public BaseResp syncGrain(BaseReqData reqData) {

        GatewayDevice device = reqData.getDevice();


        if (StringUtils.isEmpty(device.getCableCir())) {
            return getGrainTest1(reqData, device);
       return new BaseResp();
        }


        BaseResp resp = new BaseResp();
        resp.setCode(500);
        resp.setMsg("没有匹配到规则");
        return resp;
    }

    private BaseResp getGrainTest1(SyncReqData reqData, GatewayDevice device) {

        String[] cableRule = device.getCableRule().split("-");

        int cableZ = Integer.valueOf(cableRule[0]);
        int cableY = Integer.valueOf(cableRule[1]);
        int cableX = Integer.valueOf(cableRule[2]);
        int sumNum = cableZ * cableY * cableX;


        WeatherWebDto weather = WeatherWebDto.contextMap.get("default");
        double tMIn = 20, tMax = 25;
        if (null != weather) {
            double tOut = Double.valueOf(weather.getTem());
            tMIn = tOut - 4;
            if(tMIn < 4)tMIn = 5;
            tMax = tOut + 0;
            if(tMax< 15)tMax = 15;
        }


        //数据封装
        GrainData grain = new GrainData();
        grain.setMessageId(ScConstant.getMessageId());
        grain.setDeviceId(device.getDeviceId());
        grain.setTimestamp(System.currentTimeMillis() + "");

        ClientHeaders headers = new ClientHeaders();
        headers.setDeviceName(device.getDeviceName());
        headers.setProductId(device.getProductId());
        headers.setOrgId(device.getOrgId());
        headers.setMsgId(ScConstant.getMessageId());
        grain.setHeaders(headers);

        GrainOutPut outPut = new GrainOutPut();

        outPut.setAvgTemperature(NumberUtil.keepPrecision((tMax + tMIn) / 2, 1) + "");
        outPut.setMinTemperature(tMax + "");
        outPut.setMaxTemperature(tMIn + "");


        List<GrainTemp> temperature = new ArrayList<>();
        //根号
        int cableNum = 1, position = 0;

        double curTemp = tMIn;
        double randomNumber = tMIn;
        int x = 0, y = 0, z = 0;
        for (int i = 0; i < sumNum; i++) {
            if((i) % Integer.valueOf(cableRule[0]) == 0){
                randomNumber = Math.random() * (tMax - tMIn + 1) + tMIn;

            } else if ((i) % Integer.valueOf(cableRule[0]) == 1) {
                randomNumber = Math.random() * (tMax - tMIn + 1) + tMIn;

            }else if ((i) % Integer.valueOf(cableRule[0]) == 2) {
                randomNumber = Math.random() * (tMax - tMIn + 1) + tMIn;

            }else if ((i) % Integer.valueOf(cableRule[0]) == 3) {
                randomNumber = Math.random() * (tMax - tMIn + 1) + tMIn;

            }else if ((i) % Integer.valueOf(cableRule[0]) == 4) {
                randomNumber = Math.random() * (tMax - tMIn + 1) + tMIn;

            }else {
                randomNumber = Math.random() * (tMax - tMIn + 1) + tMIn;

            }
           // randomNumber = Math.random() * (tMax - tMIn + 1) + tMIn;
            curTemp = NumberUtil.keepPrecision(randomNumber, 1);
            position = i;
            z = i % cableZ + 1;
            x = i / (cableZ * cableY);
            y = x * (cableZ * cableY);
            y = (i - y) / cableZ;

            //根号
            cableNum = (i / cableZ) + 1;

            temperature.add(new GrainTemp(cableNum + "", z + "", curTemp + "", position + ""));
        }

        outPut.setTemperature(temperature);

        grain.setOutPut(outPut);

        //气象信息
        GrainWeather weatherStation = new GrainWeather();
        weatherStation.setMessageId(ScConstant.getMessageId());
        weatherStation.setId(device.getDeviceId());
        weatherStation.setAirPressure(weather.getPressure());
        weatherStation.setHumidity(weather.getHumidity());
        weatherStation.setPm(weather.getAir_pm25());
        weatherStation.setRadiation("0");
        weatherStation.setRainfallAmount(weather.getWea());
        weatherStation.setTemperature(weather.getTem());
        weatherStation.setWindDirection(weather.getWin());
        weatherStation.setWindPower(weather.getWin_meter());
        weatherStation.setWindSpeed(weather.getWin_speed());
        grain.setWeatherStation(weatherStation);

        BaseResp resp = new BaseResp();
        resp.setData(JSONObject.toJSONString(grain));

        return resp;
    }
}


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

@@ -1,11 +1,8 @@
package com.fzzy.gateway.hx2023.service;

import com.fzzy.gateway.api.GatewaySyncIdCardService;
import com.fzzy.gateway.api.GatewaySyncLedService;
import com.fzzy.gateway.api.GatewaySyncLprService;
import com.fzzy.gateway.api.GatewaySyncWeightService;
import com.fzzy.gateway.data.BaseReqData;
import com.fzzy.gateway.hx2023.data.IdCardData;
import com.fzzy.gateway.hx2023.data.*;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
@@ -25,7 +22,7 @@
    }

    @Override
    public IdCardData syncIdCard(SyncReqData reqData) {
    public IdCardData syncIdCard(BaseReqData reqData) {
        return null;
    }
}

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

@@ -1,7 +1,7 @@
package com.fzzy.gateway.hx2023.service;

import com.fzzy.gateway.api.GatewaySyncLedService;
import com.fzzy.gateway.hx2023.data.SyncReqData;
import com.fzzy.gateway.data.BaseReqData;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
@@ -21,7 +21,7 @@
    }

    @Override
    public String push2Led(SyncReqData reqData) {
    public String push2Led(BaseReqData reqData) {
        return null;
    }
}

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

@@ -7,10 +7,8 @@
import com.fzzy.gateway.entity.GatewayDevice;
import com.fzzy.gateway.hx2023.ScConstant;
import com.fzzy.gateway.hx2023.data.CloudSendData;
import com.fzzy.gateway.hx2023.data.SyncReqData;
import com.fzzy.gateway.data.BaseReqData;
import com.fzzy.mqtt.MqttGatewayService;
import com.fzzy.mqtt.MqttProviderConfig;
import com.fzzy.mqtt.MqttPublishService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

@@ -39,9 +37,7 @@
        try {
            CloudSendData cloudSendData = JSONObject.parseObject(message, CloudSendData.class);


            String functionId = cloudSendData.getFunctionId();

            //粮情采集
            if (ScConstant.FUNCTION_getTAndRHInfo.equals(functionId)) {
                getTAndRHInfo(cloudSendData);
@@ -59,7 +55,7 @@

        GatewayDevice device = GatewayUtils.getCacheByDeviceId(deviceId);

        SyncReqData syncReqData = new SyncReqData();
        BaseReqData syncReqData = new BaseReqData();
        syncReqData.setDeviceId(deviceId);
        syncReqData.setMessageId(cloudSendData.getMessageId());
        syncReqData.setMessageType(cloudSendData.getMessageType());
@@ -67,19 +63,17 @@
        syncReqData.setAutoReplay(true);
        syncReqData.setDevice(device);

        BaseResp resp = gatewayRemoteManager.getSyncGrainService(device.getSyncProtocol()).syncGrain2(syncReqData);

        //TODO --->>暂时返回测试数据

        //BaseResp resp = gatewayRemoteManager.getSyncGrainService(device.getSyncProtocol()).syncGrain(syncReqData);

        BaseResp resp = gatewayRemoteManager.getGatewayTestService(device.getPushProtocol()).testGrain(syncReqData);

        //自动推送
        if (200 == resp.getCode() && syncReqData.isAutoReplay()) {
            String topic = ScConstant.TOPIC_REPORT;

            topic = topic.replace("${productId}", device.getProductId()).replace("${deviceId}", device.getDeviceId());

            mqttGatewayService.publishMqttWithTopic(resp.getData(), topic);

            log.info("----------------------------推送MQTT信息---------------------------");
            log.info("-----TOPIC-----{}",topic);
            log.info("-----Message-----{}",resp.getData());
            syncReqData.setData(resp.getData());
            gatewayRemoteManager.getDeviceReportService(device.getPushProtocol()).reportGrainData(syncReqData);
        }
    }
}

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

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,363 @@
package com.fzzy.gateway.hx2023.service;

import com.alibaba.fastjson2.JSONObject;
import com.fzzy.api.data.PushProtocol;
import com.fzzy.api.utils.DateUtil;
import com.fzzy.api.utils.NumberUtil;
import com.fzzy.async.fzzy40.Fzzy40CommonService;
import com.fzzy.async.fzzy40.entity.Fz40Grain;
import com.fzzy.gateway.api.GatewayDeviceReportService;
import com.fzzy.gateway.api.GatewayDeviceTestService;
import com.fzzy.gateway.data.BaseReqData;
import com.fzzy.gateway.data.BaseResp;
import com.fzzy.gateway.data.WeatherWebDto;
import com.fzzy.gateway.entity.GatewayDevice;
import com.fzzy.gateway.hx2023.ScConstant;
import com.fzzy.gateway.hx2023.data.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.time.DateFormatUtils;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

@Slf4j
@Component
public class ScGatewayTestServiceImpl implements GatewayDeviceTestService {

    @Resource
    private Fzzy40CommonService fzzy40CommonService;

    @Override
    public String getProtocol() {
        return PushProtocol.GATEWAY_SC_2023.getCode();
    }

    @Override
    public BaseResp testGrain(BaseReqData reqData) {

        GatewayDevice device = reqData.getDevice();

        if (StringUtils.isEmpty(device.getCableCir())) {
            return getGrainTest1(reqData, device);
        }

        BaseResp resp = new BaseResp();
        resp.setCode(500);
        resp.setMsg("没有匹配到规则");
        return resp;
    }

    @Override
    public BaseResp testGrainKafka(BaseReqData reqData) {

        Date start = DateUtil.getCurZero(reqData.getDayTime());
        Date end = DateUtil.getNextZero(reqData.getDayTime());

        return pushByV40(reqData, start, end);
    }

    private BaseResp pushByV40(BaseReqData reqData, Date start, Date end) {


        GatewayDevice device = reqData.getDevice();
        String depotIdSys = device.getDepotIdSys();

        if (StringUtils.isEmpty(depotIdSys)) {
            log.error("--------设备--{}-未配置系统相关仓库编码，无法执行当前操作", device.getDeviceName());
            return new BaseResp(BaseResp.CODE_50, "未配置系统相关仓库编码，无法执行当前操作");
        }

        List<Fz40Grain> listGrain = fzzy40CommonService.listGrain(depotIdSys, start, end);
        if (null == listGrain || listGrain.isEmpty()) {
            log.error("---------设备---{}--未同步到粮情信息，请确认当前条件下是否有数据", device.getDeviceName());
            return new BaseResp(BaseResp.CODE_50, "未同步到粮情信息，请确认当前条件下是否有数据");
        }

        //获取最后一条粮情作为当前数据
        Fz40Grain lastData = listGrain.get(listGrain.size() - 1);
        KafaGrainData pushData = this.lastData2PushData(lastData, device);

        return new BaseResp(JSONObject.toJSONString(pushData));
    }

    /**
     * å°†ç²®æƒ…数据转换为推送需要的数据格式
     *
     * @param lastData
     * @return
     */
    private KafaGrainData lastData2PushData(Fz40Grain lastData, GatewayDevice device) {
        KafaGrainData result = new KafaGrainData();

        result.setMessageId(ScConstant.getMessageId());
        result.setMessgeId(result.getMessageId());
        result.setDeviceID(device.getDeviceId());
        result.setAvgTemperature(lastData.getTempAve() + "");
        result.setMinTemperature(lastData.getTempMin() + "");
        result.setMaxTemperature(lastData.getTempMax() + "");
        result.setCollectTime(DateFormatUtils.format(lastData.getReceiveDate(), "yyyy-MM-dd HH:mm:ss"));

        //层-行-列
        String[] attrCable = lastData.getCable().split("-");


        if (StringUtils.isNotEmpty(lastData.getCableCir())) {
            return lastData2PushData2(lastData, device);
        }

        //层行列
        int cableZ = Integer.valueOf(attrCable[0]);
        int cableY = Integer.valueOf(attrCable[1]);
        int cableX = Integer.valueOf(attrCable[2]);

        //温度集合
        String[] attr = lastData.getPoints().split(",");

        //根号
        int cableNum = 1, position = 0;

        String curTemp;
        List<KafkaGrainDataDetail1> temperature = new ArrayList<>();

        int x = 0, y = 0, z = 0;
        for (int i = 0; i < attr.length; i++) {

            position = i;

            z = i % cableZ + 1;
            x = i / (cableZ * cableY);
            y = x * (cableZ * cableY);
            y = (i - y) / cableZ;

            // å€’转Xè½´
            x = cableX - 1 - x;

            //根号
            cableNum = (i / cableZ) + 1;

            curTemp = attr[i];


            //判断最大
            if (curTemp.equals(result.getMaxTemperature())) {
                result.setMaxX(x + "");
                result.setMaxY(y + "");
                result.setMaxZ(position + "");
            }


            //判断最小
            if (curTemp.equals(result.getMinTemperature())) {
                result.setMinX(x + "");
                result.setMinY(y + "");
                result.setMinZ(position + "");
            }

            temperature.add(new KafkaGrainDataDetail1(cableNum + "", z + "", curTemp, position + "", x + "", y + ""));
        }

        //粮温信息

        JSONObject trhInfo = new JSONObject();
        // TRHInfo trhInfo = new TRHInfo();
        trhInfo.put("temperature", temperature);


        //仓温度信息
        KafkaGrainTH grainTH = new KafkaGrainTH();
        grainTH.setHumidity(lastData.getHumidityIn() + "");
        grainTH.setTemperature(lastData.getTempIn() + "");
        grainTH.setAirHumidity(lastData.getHumidityOut() + "");
        grainTH.setAirTemperature(lastData.getTempOut() + "");

        List<KafkaGrainTH> temperatureAndhumidity = new ArrayList<>();
        temperatureAndhumidity.add(grainTH);

        trhInfo.put("temperatureAndhumidity", temperatureAndhumidity);
        //trhInfo.put("temperatureAndhumidity",grainTH);

        JSONObject params = new JSONObject();
        params.put("TRHInfo", trhInfo);

        result.setParams(params);

        return result;
    }

    /**
     * é’ˆå¯¹ç­’仓  TODO -----
     *
     * @param lastData
     * @param device
     * @return
     */
    private KafaGrainData lastData2PushData2(Fz40Grain lastData, GatewayDevice device) {

        return null;
    }


    @Override
    public BaseResp testWeight(BaseReqData reqData) {

        GatewayDevice device = reqData.getDevice();
        WebSocketPacket packet = new WebSocketPacket();

        WebSocketPacketHeader header = new WebSocketPacketHeader();
        header.setDeviceName(device.getDeviceName());
        header.setProductId(device.getProductId());
        packet.setHeaders(header);
        packet.setMessageType(ScConstant.MESSAGE_TYPE_REPORT_PROPERTY);
        packet.setDeviceId(device.getDeviceId());

        //设置信息主体
        WeightInfo weightInfo = new WeightInfo();
        weightInfo.setGrossWeight(reqData.getWeight());
        weightInfo.setNetWeight(reqData.getWeight());
        weightInfo.setNetWeight(reqData.getWeight());
        weightInfo.setWeightUnit("KG");
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("weightInfo", JSONObject.toJSONString(weightInfo));

        packet.setProperties(jsonObject);

        packet.setTimestamp(System.currentTimeMillis());

        return new BaseResp(JSONObject.toJSONString(packet));
    }

    @Override
    public BaseResp testLpr(BaseReqData reqData) {


        WebSocketPacket packet = new WebSocketPacket();
        WebSocketPacketHeader header = new WebSocketPacketHeader();
        header.setDeviceName(reqData.getDeviceName());
        header.setProductId(reqData.getProductId());

        packet.setHeaders(header);
        packet.setMessageType(ScConstant.MESSAGE_TYPE_REPORT_PROPERTY);
        packet.setDeviceId(reqData.getDeviceId());
        packet.setMessageId(System.currentTimeMillis() + "");
        //设置信息主体
        LprData lpr = new LprData();
        lpr.setDeviceId(reqData.getDeviceId());
        lpr.setCarNumber(reqData.getCarNumber());
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("carNumber", reqData.getCarNumber());
        jsonObject.put("position", "big");
        packet.setProperties(jsonObject);

        packet.setTimestamp(System.currentTimeMillis());

        return new BaseResp(JSONObject.toJSONString(packet));
    }

    private BaseResp getGrainTest1(BaseReqData reqData, GatewayDevice device) {

        String[] cableRule = device.getCableRule().split("-");

        int cableZ = Integer.valueOf(cableRule[0]);
        int cableY = Integer.valueOf(cableRule[1]);
        int cableX = Integer.valueOf(cableRule[2]);
        int sumNum = cableZ * cableY * cableX;


        WeatherWebDto weather = WeatherWebDto.contextMap.get("default");
        double tMIn = 20, tMax = 25;
        if (null != weather) {
            double tOut = Double.valueOf(weather.getTem());
            tMIn = tOut - 4;
            if (tMIn < 4) tMIn = 5;
            tMax = tOut + 0;
            if (tMax < 15) tMax = 15;
        }


        //数据封装
        GrainData grain = new GrainData();
        grain.setMessageId(ScConstant.getMessageId());
        grain.setDeviceId(device.getDeviceId());
        grain.setTimestamp(System.currentTimeMillis() + "");

        ClientHeaders headers = new ClientHeaders();
        headers.setDeviceName(device.getDeviceName());
        headers.setProductId(device.getProductId());
        headers.setOrgId(device.getOrgId());
        headers.setMsgId(ScConstant.getMessageId());
        grain.setHeaders(headers);

        GrainOutPut outPut = new GrainOutPut();

        outPut.setAvgTemperature(NumberUtil.keepPrecision((tMax + tMIn) / 2, 1) + "");
        outPut.setMinTemperature(tMax + "");
        outPut.setMaxTemperature(tMIn + "");


        List<GrainTemp> temperature = new ArrayList<>();
        //根号
        int cableNum = 1, position = 0;

        double curTemp = tMIn;
        double randomNumber = tMIn;
        int x = 0, y = 0, z = 0;
        for (int i = 0; i < sumNum; i++) {
            if ((i) % Integer.valueOf(cableRule[0]) == 0) {
                randomNumber = Math.random() * (tMax - tMIn + 1) + tMIn;

            } else if ((i) % Integer.valueOf(cableRule[0]) == 1) {
                randomNumber = Math.random() * (tMax - tMIn + 1) + tMIn;

            } else if ((i) % Integer.valueOf(cableRule[0]) == 2) {
                randomNumber = Math.random() * (tMax - tMIn + 1) + tMIn;

            } else if ((i) % Integer.valueOf(cableRule[0]) == 3) {
                randomNumber = Math.random() * (tMax - tMIn + 1) + tMIn;

            } else if ((i) % Integer.valueOf(cableRule[0]) == 4) {
                randomNumber = Math.random() * (tMax - tMIn + 1) + tMIn;

            } else {
                randomNumber = Math.random() * (tMax - tMIn + 1) + tMIn;

            }
            // randomNumber = Math.random() * (tMax - tMIn + 1) + tMIn;
            curTemp = NumberUtil.keepPrecision(randomNumber, 1);
            position = i;
            z = i % cableZ + 1;
            x = i / (cableZ * cableY);
            y = x * (cableZ * cableY);
            y = (i - y) / cableZ;

            //根号
            cableNum = (i / cableZ) + 1;

            temperature.add(new GrainTemp(cableNum + "", z + "", curTemp + "", position + ""));
        }

        outPut.setTemperature(temperature);

        grain.setOutPut(outPut);

        //气象信息
        GrainWeather weatherStation = new GrainWeather();
        weatherStation.setMessageId(ScConstant.getMessageId());
        weatherStation.setId(device.getDeviceId());
        weatherStation.setAirPressure(weather.getPressure());
        weatherStation.setHumidity(weather.getHumidity());
        weatherStation.setPm(weather.getAir_pm25());
        weatherStation.setRadiation("0");
        weatherStation.setRainfallAmount(weather.getWea());
        weatherStation.setTemperature(weather.getTem());
        weatherStation.setWindDirection(weather.getWin());
        weatherStation.setWindPower(weather.getWin_meter());
        weatherStation.setWindSpeed(weather.getWin_speed());
        grain.setWeatherStation(weatherStation);

        return new BaseResp(JSONObject.toJSONString(grain));
    }
}

 src/main/java/com/fzzy/gateway/service/GatewayDeviceTestPR.java

@@ -1,20 +1,18 @@
package com.fzzy.gateway.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.bstek.dorado.annotation.Expose;
import com.fzzy.api.data.GatewayDeviceType;
import com.fzzy.api.utils.DateUtil;
import com.fzzy.async.fzzy40.Fzzy40CommonService;
import com.fzzy.async.fzzy40.entity.Fz40Grain;
import com.fzzy.gateway.api.DeviceReportService;
import com.fzzy.gateway.api.GatewayDeviceReportService;
import com.fzzy.gateway.api.GatewayRemoteManager;
import com.fzzy.gateway.data.BaseResp;
import com.fzzy.gateway.data.BaseReqData;
import com.fzzy.gateway.entity.GatewayDevice;
import com.fzzy.gateway.hx2023.ScConstant;
import com.fzzy.gateway.hx2023.data.*;
import com.fzzy.gateway.service.repository.GatewayDeviceRep;
import com.fzzy.mqtt.MqttGatewayService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.time.DateFormatUtils;
@@ -59,7 +57,7 @@
     */
    @Expose
    public String ajaxTestGrain2(GatewayDevice data) {
        SyncReqData reqData = new SyncReqData();
        BaseReqData reqData = new BaseReqData();
        reqData.setDevice(data);
        reqData.setAutoReplay(true);
        reqData.setMessageType(ScConstant.MESSAGE_TYPE_INVOKE_FUNCTION);
@@ -130,7 +128,7 @@

        KafaGrainData pushData;

        DeviceReportService deviceReportService = null;
        GatewayDeviceReportService deviceReportService = null;
        for (GatewayDevice device : list) {
            depotIdSys = device.getDepotIdSys();


 src/main/java/com/fzzy/gateway/view/GatewayDevice.view.xml

@@ -48,9 +48,9 @@
        <Property></Property>
        <Property name="label">视频RTSP</Property>
      </PropertyDef>
      <PropertyDef name="provinceProtocol">
      <PropertyDef name="pushProtocol">
        <Property></Property>
        <Property name="label">省平台协议</Property>
        <Property name="label">上行协议</Property>
        <Property name="mapping">
          <Property name="mapValues">${dorado.getDataProvider(&quot;apiTriggerService#triggerPushProtocol&quot;).getResult()}</Property>
          <Property name="keyProperty">code</Property>
@@ -58,19 +58,9 @@
        </Property>
        <Property name="required">true</Property>
      </PropertyDef>
      <PropertyDef name="pushProtocol">
        <Property></Property>
        <Property name="label">与系统通讯协议</Property>
        <Property name="mapping">
          <Property name="mapValues">${dorado.getDataProvider(&quot;apiTriggerService#triggerGatewayProtocol&quot;).getResult()}</Property>
          <Property name="keyProperty">code</Property>
          <Property name="valueProperty">name</Property>
        </Property>
        <Property name="required">true</Property>
      </PropertyDef>
      <PropertyDef name="syncProtocol">
        <Property></Property>
        <Property name="label">与设备通讯协议</Property>
        <Property name="label">下行协议</Property>
        <Property name="mapping">
          <Property name="keyProperty">code</Property>
          <Property name="valueProperty">name</Property>
@@ -364,11 +354,6 @@
            <AutoFormElement>
              <Property name="name">password</Property>
              <Property name="property">password</Property>
              <Editor/>
            </AutoFormElement>
            <AutoFormElement>
              <Property name="name">provinceProtocol</Property>
              <Property name="property">provinceProtocol</Property>
              <Editor/>
            </AutoFormElement>
            <AutoFormElement>

 src/main/java/com/fzzy/mqtt/MqttConsumerConfig.java

@@ -67,8 +67,7 @@
        String[] topics = mqttProperties.getTopics().split(",");
        //topicList.toArray(topics);

        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(mqttProperties.getClientInId(), receiverMqttClientFactoryForSub(), topics);
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(mqttProperties.getClientInId(), receiverMqttClientFactoryForSub(), topics);
        adapter.setCompletionTimeout(mqttProperties.getCompletionTimeout());
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(mqttProperties.getQos());