水源消息同步至全域

amos-boot-module/amos-boot-module-biz/amos-boot-module-common-biz/src/main/java/com/yeejoin/amos/boot/module/common/biz/controller/WaterResourceController.java

 package com.yeejoin.amos.boot.module.common.biz.controller;
 
 import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
@@ -15,19 +16,23 @@ import com.yeejoin.amos.boot.biz.common.utils.QRCodeUtil;
 import com.yeejoin.amos.boot.module.common.api.core.framework.PersonIdentify;
 import com.yeejoin.amos.boot.module.common.api.dto.*;
 import com.yeejoin.amos.boot.module.common.api.entity.*;
+import com.yeejoin.amos.boot.module.common.api.enums.DataSyncTopicEnum;
 import com.yeejoin.amos.boot.module.common.api.enums.OrgPersonEnum;
 import com.yeejoin.amos.boot.module.common.api.enums.WaterResourceTypeEnum;
 import com.yeejoin.amos.boot.module.common.api.feign.EquipFeignClient;
 import com.yeejoin.amos.boot.module.common.biz.service.impl.*;
 import com.yeejoin.amos.boot.module.common.biz.utils.CommonResponseUtil;
+import com.yeejoin.amos.component.rule.config.RuleConfig;
 import io.swagger.annotations.Api;
 import io.swagger.annotations.ApiOperation;
 import org.apache.commons.lang3.ObjectUtils;
 import org.apache.commons.lang3.StringUtils;
+import org.eclipse.paho.client.mqttv3.MqttException;
 import org.springframework.beans.BeanUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.web.bind.annotation.*;
+import org.typroject.tyboot.component.emq.EmqKeeper;
 import org.typroject.tyboot.core.foundation.enumeration.UserType;
 import org.typroject.tyboot.core.foundation.utils.ValidationUtil;
 import org.typroject.tyboot.core.restful.doc.TycloudOperation;
@@ -72,6 +77,9 @@ public class WaterResourceController extends BaseController {
     @Value("${auth-key-fire-water:fire_water_info}")
     private String authKey;
 
+    @Autowired
+    private EmqKeeper emqKeeper;
+
     /**
      * 新增
      *
@@ -180,9 +188,21 @@ public class WaterResourceController extends BaseController {
         if (model.getSequenceNbr() != null) {
             model.setResourceId(model.getSequenceNbr());
         }
+        // 消息转发至全域
+        syncWaterDataToCenter(model, "create");
         return ResponseHelper.buildResponse(model);
     }
 
+    private void syncWaterDataToCenter(WaterResourceDto waterResourceDto, String operateType) {
+        JSONObject message = new JSONObject();
+        message.put("topic", "jcs/sync/water" + "/" + operateType);
+        message.put("data", waterResourceDto);
+        try {
+            emqKeeper.getMqttClient().publish("emq.bussSign.created", message.toJSONString().getBytes(), RuleConfig.DEFAULT_QOS, false);
+        } catch (MqttException e) {
+            e.printStackTrace();
+        }
+    }
 
     /**
      * 获取关联装备定义物联参数字段
@@ -326,6 +346,7 @@ public class WaterResourceController extends BaseController {
             }
             waterResourceIndexServiceImpl.saveOrUpdateBatch(list);
         }
+        syncWaterDataToCenter(waterResourceDto, "update");
         return ResponseHelper.buildResponse(waterResourceDto);
     }
 
@@ -344,6 +365,7 @@ public class WaterResourceController extends BaseController {
         BeanUtils.copyProperties(waterResource, waterResourceDto);
         waterResourceDto.setIsDelete(false);
         waterResourceServiceImpl.updateWithModel(waterResourceDto);
+        syncWaterDataToCenter(waterResourceDto, "delete");
         return ResponseHelper.buildResponse(true);
     }