代码提交

iot-access/access-boot/src/main/java/com/njcn/access/handler/MqttMessageHandler.java

@@ -5,26 +5,39 @@ import com.github.tocrhz.mqtt.annotation.MqttSubscribe;
 import com.github.tocrhz.mqtt.annotation.NamedValue;
 import com.github.tocrhz.mqtt.annotation.Payload;
 import com.github.tocrhz.mqtt.publisher.MqttPublisher;
-import com.njcn.access.enums.AccessStatusEnum;
-import com.njcn.access.pojo.dto.*;
+import com.njcn.access.enums.AccessEnum;
+import com.njcn.access.enums.AccessResponseEnum;
+import com.njcn.access.enums.TypeEnum;
+import com.njcn.access.pojo.dto.DevModInfoDto;
+import com.njcn.access.pojo.dto.ModelDto;
+import com.njcn.access.pojo.dto.TopicDto;
+import com.njcn.access.pojo.dto.heart.HeartBeatDto;
+import com.njcn.access.pojo.param.ReqAndResParam;
+import com.njcn.access.pojo.po.CsTopicPO;
+import com.njcn.access.service.ICsTopicService;
 import com.njcn.algorithm.api.DevModelFeignClient;
 import com.njcn.algorithm.api.EquipmentFeignClient;
 import com.njcn.algorithm.pojo.param.CsDevModelQueryListParm;
 import com.njcn.algorithm.pojo.vo.CsDevModelPageVO;
 import com.njcn.algorithm.pojo.vo.CsEquipmentDeliveryVO;
 import com.njcn.common.pojo.response.HttpResult;
+import com.njcn.common.utils.PubUtils;
 import com.njcn.redis.utils.RedisUtil;
 import com.njcn.system.api.DicDataFeignClient;
+import com.njcn.system.enums.DicDataEnum;
 import com.njcn.system.pojo.po.DictData;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
 
+import javax.validation.ConstraintViolation;
+import javax.validation.Validator;
 import java.nio.charset.StandardCharsets;
 import java.time.Instant;
-import java.util.List;
-import java.util.Objects;
+import java.util.*;
 
 /**
  * @author hongawen
@@ -46,35 +59,82 @@ public class MqttMessageHandler {
 
     private final RedisUtil redisUtil;
 
+    private final ICsTopicService csTopicService;
+
+    @Autowired
+    Validator validator;
+
+    @MqttSubscribe(value = "/Dev/Topic/{edgeId}",qos = 1)
+    @Transactional(rollbackFor = Exception.class)
+    public void devTopic(String topic, MqttMessage message, @NamedValue("edgeId") String nDid, @Payload String payload){
+        Gson gson = new Gson();
+        ReqAndResParam.Res res = gson.fromJson(new String(message.getPayload(), StandardCharsets.UTF_8), ReqAndResParam.Res.class);
+        //检验传递的参数是否准确
+        Set<ConstraintViolation<ReqAndResParam.Res>> validate = validator.validate(res);
+        validate.forEach(constraintViolation -> {
+            System.out.println(constraintViolation.getMessage());
+        });
+        if (Objects.equals(res.getCode(),AccessEnum.SUCCESS.getCode())){
+            if (Objects.equals(res.getType(), TypeEnum.TYPE_1.getCode())){
+                List<CsTopicPO> list = new ArrayList<>();
+                //fixme 这边获取数据需要调整
+                Map<String,List<String>> map = (Map<String,List<String>>)res.getMsg();
+                List<String> topicList = map.get("Topic");
+                topicList.forEach(item->{
+                    CsTopicPO csTopicPo = new CsTopicPO();
+                    csTopicPo.setNdid(nDid);
+                    csTopicPo.setTopic(item);
+                    csTopicPo.setType(0);
+                    list.add(csTopicPo);
+                });
+                csTopicService.addList(list);
+            } else {
+                log.info(AccessResponseEnum.MESSAGE_TYPE_ERROR.getMessage());
+            }
+        } else {
+            log.info(AccessResponseEnum.RESPONSE_ERROR.getMessage());
+        }
+    }
+
     /**
-     * 接收装置接入响应
+     * 装置注册应答
      * 1.收到注册信息，修改装置出厂表，装置的状态，调整为注册；然后开始接入流程
      * 2.询问当前装置类型的模板。有则完成接入；没有则告警出来，需要人工手动上传模板信息
      * @param topic
      * @param message
      * @param payload
      */
-    @MqttSubscribe(value = "/device/register/{nDid}",qos = 1)
-    public void devOperation(String topic, MqttMessage message, @NamedValue("nDid") String nDid, @Payload String payload){
+    @MqttSubscribe(value = "/Dev/Reg/{edgeId}",qos = 1)
+    @Transactional(rollbackFor = Exception.class)
+    public void devOperation(String topic, MqttMessage message, @NamedValue("edgeId") String nDid, @Payload String payload){
         Gson gson = new Gson();
-        RegisterDTO.RegisterResponse registerDTO = gson.fromJson(new String(message.getPayload(), StandardCharsets.UTF_8), RegisterDTO.RegisterResponse.class);
-        if (registerDTO.getCode() == 200){
-            equipmentFeignClient.updateStatusBynDid(nDid, AccessStatusEnum.REGISTERED.getCode());
-            PublicDto publicDto = new PublicDto();
-            publicDto.setMid(Long.toString(Instant.now().toEpochMilli()));
-            publicDto.setNDid(nDid);
-            publicDto.setTimestamp(Instant.now().toEpochMilli());
-            publicDto.setType("CMD_DEV_DATA");
-            AccessDto accessDto = new AccessDto();
-            accessDto.setNDid(nDid);
-            accessDto.setDevType(registerDTO.getParam().getDev_type());
-            publicDto.setParam(accessDto);
-            publisher.send("/platform/devcmd/"+nDid,new Gson().toJson(publicDto),1,false);
+        ReqAndResParam.Res res = gson.fromJson(new String(message.getPayload(), StandardCharsets.UTF_8), ReqAndResParam.Res.class);
+        //检验传递的参数是否准确
+        Set<ConstraintViolation<ReqAndResParam.Res>> validate = validator.validate(res);
+        validate.forEach(constraintViolation -> {
+            System.out.println(constraintViolation.getMessage());
+        });
+        if (Objects.equals(res.getCode(),AccessEnum.SUCCESS.getCode())){
+            if (Objects.equals(res.getType(),TypeEnum.TYPE_17.getCode())){
+                equipmentFeignClient.updateStatusBynDid(nDid, AccessEnum.REGISTERED.getCode());
+                //询问模板数据
+                ReqAndResParam.Req reqAndResParam = new ReqAndResParam.Req();
+                reqAndResParam.setMid(1);
+                reqAndResParam.setDid("0");
+                reqAndResParam.setPri(AccessEnum.FIRST_CHANNEL.getCode());
+                reqAndResParam.setType(TypeEnum.TYPE_3.getCode());
+                reqAndResParam.setExpire(-1);
+                publisher.send("/Pfm/DevCmd/V1/"+nDid,new Gson().toJson(reqAndResParam),1,false);
+            } else {
+                log.info(AccessResponseEnum.MESSAGE_TYPE_ERROR.getMessage());
+            }
+        } else {
+            log.info(AccessResponseEnum.RESPONSE_ERROR.getMessage());
         }
     }
 
     /**
-     * 装置类型模板相应
+     * 装置类型模板应答
      * 1.判断网关的类型
      * 2.直联设备的DevCfg和DevMod是以直联设备为准，上送平台端，平台端保存。通过校验DevMod模板信息来从平台端模板池中选取对应的模板，如果找不到匹配模板需告警提示人工干预处理。
      * 3.平台端需读取装置的DevMod来判断网关支持的设备模板（包含设备型号和模板版本），根据app提交的接入子设备DID匹配数据模板（型号及版本），生成DevCfg下发给网关，网关根据下发信息生成就地设备点表。
@@ -83,42 +143,92 @@ public class MqttMessageHandler {
      * @param nDid
      * @param payload
      */
-    @MqttSubscribe(value = "/device/devack/{nDid}",qos = 1)
-    public void devModelOperation(String topic, MqttMessage message, @NamedValue("nDid") String nDid, @Payload String payload){
+    @MqttSubscribe(value = "/Pfm/DevRsp/{version}/{edgeId}",qos = 1)
+    @Transactional(rollbackFor = Exception.class)
+    public void devModelOperation(String topic, MqttMessage message,  @NamedValue("version") String version, @NamedValue("edgeId") String nDid, @Payload String payload){
         Gson gson = new Gson();
         ModelDto modelDto = gson.fromJson(new String(message.getPayload(), StandardCharsets.UTF_8), ModelDto.class);
         HttpResult<CsEquipmentDeliveryVO> pojo = equipmentFeignClient.queryEquipmentByndid(nDid);
         if (!Objects.isNull(pojo)){
             String devType = pojo.getData().getDevType();
-            if (Objects.equals(devType,"直连设备")){
-                List<DevModelDto> list = modelDto.getDevMod();
+            String dictCode = dicDataFeignClient.getDicDataById(devType).getData().getCode();
+            //直连设备处理
+            if (Objects.equals(dictCode, DicDataEnum.CONNECT_DEV.getCode())){
+                List<DevModInfoDto> list = modelDto.getDevMod();
                 list.forEach(item->{
                     DictData dicData = dicDataFeignClient.getDicDataByCode(item.getDevType()).getData();
                     CsDevModelQueryListParm csDevModelQueryListParm = new CsDevModelQueryListParm();
                     if (Objects.isNull(dicData)) {
-                        log.info("新增模板失败,获取装置类型字典数据为空,请先录入装置类型!");
+                        log.info(AccessResponseEnum.DEV_TYPE_NOT_FIND.getMessage());
                         return;
                     } else {
-                        csDevModelQueryListParm.setDevType( dicData.getId());
+                        csDevModelQueryListParm.setDevType(dicData.getId());
                     }
                     csDevModelQueryListParm.setVersionNo(item.getVersionNo());
                     csDevModelQueryListParm.setVersionDate(item.getVersionDate());
                     CsDevModelPageVO csDevModelPageVO = devModelFeignClient.queryDevModelOne(csDevModelQueryListParm).getData();
                     if (Objects.isNull(csDevModelPageVO)){
-                        log.info("模板不存在,请先录入模板数据!");
+                        log.info(AccessResponseEnum.MODEL_NO_FIND.getMessage());
                     } else {
-                        equipmentFeignClient.updateStatusBynDid(nDid, AccessStatusEnum.ACCESS.getCode());
-                        //todo 录入装置和模板的关系表
-
-
+                        ReqAndResParam.Req reqAndResParam = new ReqAndResParam.Req();
+                        reqAndResParam.setMid(1);
+                        reqAndResParam.setDid("0");
+                        reqAndResParam.setPri(AccessEnum.FIRST_CHANNEL.getCode());
+                        reqAndResParam.setType(TypeEnum.TYPE_5.getCode());
+                        reqAndResParam.setExpire(-1);
+                        publisher.send("/Pfm/DevCmd/"+version+"/"+nDid,new Gson().toJson(reqAndResParam),1,false);
+                        //将装置状态改为接入
+                        equipmentFeignClient.updateStatusBynDid(nDid, AccessEnum.ACCESS.getCode());
+                        //设置心跳时间，超时改为掉线
+                        redisUtil.saveByKeyWithExpire("MQTT:" + nDid,Instant.now().toEpochMilli(),180L);
                     }
                 });
-            } else if (Objects.equals(devType,"网关")){
+            }
+            //网关处理 生成报文下发给装置,装置响应版本
+            else if (Objects.equals(dictCode, DicDataEnum.GATEWAY_DEV.getCode())){
                 //todo 处理待定
                 System.out.println("网关设备判断");
             }
         } else {
-            log.info("通过nDid未找到相关装置信息!");
+            log.info(AccessResponseEnum.DEV_NO_FIND.getMessage());
+        }
+    }
+
+
+    /**
+     * 装置心跳
+     * @param topic
+     * @param message
+     * @param version
+     * @param nDid
+     * @param payload
+     */
+    @MqttSubscribe(value = "/Dev/PfmCmd/{version}/{edgeId}",qos = 1)
+    @Transactional(rollbackFor = Exception.class)
+    public void devHeartBeat(String topic, MqttMessage message,  @NamedValue("version") String version, @NamedValue("edgeId") String nDid, @Payload String payload) {
+        //响应请求
+        ReqAndResParam.Req reqAndResParam = new ReqAndResParam.Req();
+        HeartBeatDto heartBeatDto = new HeartBeatDto();
+        heartBeatDto.setTime(System.currentTimeMillis()/1000);
+        reqAndResParam.setMid(1);
+        reqAndResParam.setDid("0");
+        reqAndResParam.setPri(AccessEnum.FIRST_CHANNEL.getCode());
+        reqAndResParam.setType(TypeEnum.TYPE_14.getCode());
+        reqAndResParam.setExpire(-1);
+        reqAndResParam.setMsg(heartBeatDto);
+        publisher.send("/Dev/PfmRsp/"+version+"/"+nDid,PubUtils.obj2json(reqAndResParam),1,false);
+        //处理业务逻辑
+        ReqAndResParam.Res res = PubUtils.json2obj(new String(message.getPayload(), StandardCharsets.UTF_8), ReqAndResParam.Res.class);
+        Object object = res.getMsg();
+        if (!Objects.isNull(object)){
+            List<String> abnormalList = new ArrayList<>();
+            if (object instanceof ArrayList<?>){
+                abnormalList.addAll((List<String>) object);
+            }
+            //todo 需要处理异常异常设备
+            abnormalList.forEach(item->{
+                System.out.println("异常设备ID："+item);
+            });
         }
     }