提交网关心跳和设备状态

jiazx0107@163.com

2023-12-12 46203ee88249d9a4046f3d453deb41edc562cf6c

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

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

import com.alibaba.fastjson2.JSONObject;
import com.fzzy.api.Constant;
import com.fzzy.api.data.GatewayDeviceType;
import com.fzzy.api.data.PushProtocol;
import com.fzzy.api.utils.MyMD5Util;
import com.fzzy.api.utils.RSAUtils;
import com.fzzy.api.utils.RedisConst;
import com.fzzy.api.utils.RedisUtil;
import com.fzzy.api.utils.ContextUtil;
import com.fzzy.api.view.repository.ApiLogRep;
import com.fzzy.gateway.service.GatewayConfService;
import com.fzzy.gateway.util.GatewayHttpUtil;
import com.fzzy.gateway.GatewayUtils;
import com.fzzy.gateway.api.GatewayRemoteService;
import com.fzzy.gateway.data.BaseResp;
import com.fzzy.gateway.entity.GatewayConf;
import com.fzzy.gateway.entity.GatewayDevice;
import com.fzzy.gateway.hx2023.ScConstant;
import com.fzzy.gateway.hx2023.data.CloudResp;
import com.fzzy.gateway.util.GatewayRSAUtils;
import com.fzzy.gateway.hx2023.data.DeviceStatusData;
import com.fzzy.gateway.hx2023.data.HeartBeatData;
import com.fzzy.gateway.hx2023.kafka.KafkaDeviceReportService;
import com.fzzy.gateway.hx2023.util.OpenApiSignatureUtils;
import com.fzzy.gateway.service.GatewayConfService;
import com.fzzy.gateway.util.GatewayHttpUtil;
import lombok.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.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

@@ -30,12 +38,15 @@
public class HxGatewayRemoteServiceImpl implements GatewayRemoteService {


    String charset = "utf-8";

    @Resource
    private ApiLogRep apiLogRep;


    @Resource
    private GatewayConfService gatewayConfService;
    @Resource
    private KafkaDeviceReportService kafkaDeviceReportService;


    @Override
@@ -62,7 +73,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");
@@ -73,14 +84,15 @@


            //获取 AccessToken 接口
            String sign = getSign(params, gatewayConf.getPrivateKey());
            String sign = getSign(params, gatewayConf);
            params.put("sign", sign);
            url = gatewayConf.getApiUrl() + "reserver/api/token/apply";

            jsonStr = GatewayHttpUtil.doGet(url, params);
            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");
@@ -98,34 +110,42 @@

    @Override
    public void heartbeat(GatewayConf gatewayConf) {

        log.info("------------定时心跳执行---------{}", gatewayConf.getKqmc());

        try {

            if (StringUtils.isEmpty(gatewayConf.getApiUrl())) {
                return;
            }

            gatewayConf = getCacheConf(gatewayConf.getKqdm());
            //网关心跳接口
            Map<String, String> params = new HashMap<>();
            params.put("token", gatewayConf.getAccessToken());
            params.put("gatewayId", gatewayConf.getGatewayId());
            params.put("gatewayIp", gatewayConf.getGatewayIp());
            params.put("gatewayMac", gatewayConf.getGatewayMac());
            params.put("heartbeat", "1");
            params.put("timestamp", System.currentTimeMillis() + "");

            String sign = getSign(params, gatewayConf.getPrivateKey());
            params.put("sign", sign);

            String url = gatewayConf.getApiUrl() + "reserver/api/iot/equipment/heartbeat";
            HeartBeatData heartBeatData = new HeartBeatData();

            String jsonStr = GatewayHttpUtil.doGet(url, params);
            log.info("---网关心跳接口-返回---{}", jsonStr);
            heartBeatData.setMessageId(ContextUtil.getUUID());
            heartBeatData.setApISource(ScConstant.API_SOURCE_TARGET_EQUIPMENT_HEARTBEAT);
            heartBeatData.setGatewayId(gatewayConf.getGatewayId());
            heartBeatData.setGatewayIp(gatewayConf.getGatewayIp());
            heartBeatData.setGatewayMac(null == gatewayConf.getGatewayMac() ? "无" : gatewayConf.getGatewayMac());
            heartBeatData.setHeartbeat(1);
            heartBeatData.setReportTime(DateFormatUtils.format(new Date(), "yyyy-MM-dd hh:mm:ss"));
            heartBeatData.setMemo("FZZY");

            String topic = ScConstant.TOPIC_EQUIPMENT_HEARTBEAT;
            topic = topic.replace("{kqdm}", gatewayConf.getKqdm());

            String messageInfo = JSONObject.toJSONString(heartBeatData);

            log.info("---网关心跳推送--{}-{}", topic, messageInfo);

            kafkaDeviceReportService.publishWithTopic(messageInfo, topic);


        } catch (Exception e) {
            log.error("------网关心跳接口--执行失败-----{}", e);
        }

        //执行当前网关的设备状态推送
        pushDeviceStatus(gatewayConf);
    }

    @Override
@@ -142,13 +162,13 @@
            params.put("token", gatewayConf.getAccessToken());
            params.put("gatewayId", gatewayConf.getGatewayId());
            params.put("gatewayIp", gatewayConf.getGatewayIp());
            params.put("gatewayMac", gatewayConf.getGatewayMac());
            params.put("gatewayCPU", gatewayConf.getGatewayCPU());
            params.put("gatewayMem", gatewayConf.getGatewayMem());
            params.put("gatewayHardDisk", gatewayConf.getGatewayHardDisk());
            params.put("gatewayMac", null == gatewayConf.getGatewayMac() ? "无" : gatewayConf.getGatewayMac());
            params.put("gatewayCPU", null == gatewayConf.getGatewayCPU() ? "无" : gatewayConf.getGatewayCPU());
            params.put("gatewayMem", null == gatewayConf.getGatewayMem() ? "无" : gatewayConf.getGatewayMem());
            params.put("gatewayHardDisk", null == gatewayConf.getGatewayHardDisk() ? "无" : gatewayConf.getGatewayHardDisk());
            params.put("timestamp", System.currentTimeMillis() + "");

            String sign = getSign(params, gatewayConf.getPrivateKey());
            String sign = getSign(params, gatewayConf);
            params.put("sign", sign);

            String url = gatewayConf.getApiUrl() + "reserver/api/iot/equipment/heartbeat";
@@ -162,27 +182,63 @@
    }


    public String getSign(Map<String, String> parames, String priKey) {
        //参数拼接
        String msg = "";
        for (Map.Entry<String, String> param : parames.entrySet()) {
            msg += param.getKey() + "=" + param.getValue() + "&";
    @Override
    public void pushDeviceStatus(GatewayConf gatewayConf) {

        //获取设备列表
        Collection<GatewayDevice> list = GatewayUtils.allCacheDevice();
        if (null == list || list.isEmpty()) {
            log.info("--------系统未获取到当前系统设备列表，不执行状态推送-----");
            return;
        }
        msg = msg.substring(0, msg.length() - 1);

        log.debug("------待加密信息-----{}", msg);
        //封装设备报文信息
        DeviceStatusData statusData;
        String messageInfo;
        String topic = ScConstant.TOPIC_EQUIPMENT_STATUS;
        topic = topic.replace("{kqdm}", gatewayConf.getKqdm());

        //MD5加密
        String md5sign = MyMD5Util.getMD5(msg);
        log.debug("------md5加密-----{}", md5sign);
        for (GatewayDevice device : list) {
            //如果设备没有配置productId直接跳过
            if (StringUtils.isEmpty(device.getProductId())) continue;
            statusData = new DeviceStatusData();
            statusData.setMessageId(ContextUtil.getUUID());
            statusData.setApISource(ScConstant.API_SOURCE_TARGET_EQUIPMENT_STATUS);
            statusData.setGatewayId(gatewayConf.getGatewayId());
            statusData.setGatewayIp(gatewayConf.getGatewayIp());
            statusData.setProductId(device.getProductId());
            statusData.setEquipId(device.getDeviceId());
            statusData.setReportTime(DateFormatUtils.format(new Date(), "yyyy-MM-dd hh:mm:ss"));
            statusData.setMemo("FZZY");

        //RSA加密
        String result = GatewayRSAUtils.encryptByPrivate(md5sign, priKey);
        log.debug("------RSA加密-----{}", result);
            //针对粮情设备
            if (GatewayDeviceType.TYPE_07.getCode().equals(device.getType())) {
                if (Constant.YN_N.equals(device.getStatus())) statusData.setStatus(0);
            }

        return result;
            messageInfo = JSONObject.toJSONString(statusData);
            //推送设备状态
            kafkaDeviceReportService.publishWithTopic(messageInfo, topic);

        }
    }


    public String getSign(Map<String, String> parames, GatewayConf gatewayConf) throws Exception {

        //参数调整
        String signContent = OpenApiSignatureUtils.getSignContent(parames);
        log.debug("------待加密信息-----{}", signContent);

        //MD5加密
        String md5sign = OpenApiSignatureUtils.getMd5Content(signContent, charset);

        //RSA加密
        String singValue = OpenApiSignatureUtils.doSignByHex(gatewayConf.getGatewayId(), md5sign, charset, gatewayConf.getPublicKey(), gatewayConf.getPrivateKey());
        log.debug("------RSA加密签名-----{}", singValue);

        return singValue;
    }

    /**
     * @param kqdm
@@ -193,9 +249,7 @@
    }

    private void updateAuthToken(GatewayConf conf) {

        gatewayConfService.updateCache(conf);

    }