调整KFKA数据推送测试2

jiazx0107@163.com

2024-01-04 50da1f165cc96d49d72694606920b77c74e1747c

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

@@ -1,63 +1,92 @@
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.data.ConfigData;
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.LprData;
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 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;

@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;
    @Resource
    private ConfigData configData;

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

    @Override
    public String report2GatewayBySn(double weigh, GatewayDevice device) {
    public BaseResp reportGrainData(BaseReqData reqData) {

        if (null == device) {
            log.error("-----------没有获取到设备配置信息-----");
            return "ERROR:没有获取到设备配置信息";
        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();
        }

        //使用WEBSOCKET返回
        if (GatewayProtocol.GATE_WEBSOCKET.equals(device.getPushProtocol())) {
        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());

        if (null == reqData.getData()) {
            GatewayDevice device = reqData.getDevice();

            WebSocketPacket packet = new WebSocketPacket();

            //productId的值在webSocket执行发送方法中补充
            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(weigh);
            weightInfo.setNetWeight(weigh);
            weightInfo.setNetWeight(weigh);

            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));

@@ -65,20 +94,70 @@

            packet.setTimestamp(System.currentTimeMillis());


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

        if (GatewayProtocol.GATE_MQTT.equals(device.getPushProtocol())) {
            //TODO ---->>>后期添加支持
        }
        publishService.publishMqttWithTopic(reqData.getData(), topic);

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

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

        return kafkaDeviceReport.sendGrainData2Cloud(data);
        GatewayDevice device = reqData.getDevice();

        if (StringUtils.isEmpty(reqData.getData())) {
            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", device.getPosition());
            packet.setProperties(jsonObject);
            packet.setTimestamp(System.currentTimeMillis());

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

        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_ZLJ_GRAIN_TEMPERATURE;

        //如果是测试模式不执行推送
        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();
    }
}