设置监听

mhfire-controller/src/main/resources/application-dev.properties

@@ -79,7 +79,8 @@ mqtt.completionTimeout=5000
 mqtt.keep-alive-interval=60
 mqtt.username=usky
 mqtt.password=usky
-mqtt.sub-topics=mh/water/info,mh/water/alert,mh/water/statistics
+mqtt.sub-topics=mh/water/info,mh/water/alert,mh/water/statistics,\
+  device/info,device/alert,device/detail,device/aj
 mqtt.url=tcp://47.98.201.73:1883
 
 

mhfire-controller/src/main/resources/application-prod.properties

@@ -75,7 +75,8 @@ mqtt.completionTimeout=5000
 mqtt.keep-alive-interval=60
 mqtt.username=usky
 mqtt.password=usky
-mqtt.sub-topics=mh/water/info,mh/water/alert,mh/water/statistics
+mqtt.sub-topics=mh/water/info,mh/water/alert,mh/water/statistics,\
+  device/info,device/alert,device/detail,device/aj
 mqtt.url=tcp://47.98.201.73:1883
 
 

mhfire-service/src/main/java/com/bizmatics/mhfire/service/api/mhWater/OneCardApi.java

@@ -96,7 +96,6 @@ public class OneCardApi {
             if ("0".equals(arrNode.get("status").asText())){
                 JsonNode data = arrNode.get("data");
                 consumer.accept(data.asText());
-                log.info("闵行水系统接口-----调用成功");
             }else {
                 log.error("闵行水系统接口-----调用异常:"+arrNode.get("msg").asText());
             }
@@ -118,7 +117,6 @@ public class OneCardApi {
      * 调用告警接口
      */
     public void callAlarmApi(Object param){
-
         log.info("闵行水系统告警---开始");
         sendApiBase(String.format("%s%s",path,"/iot/alarm/yt"),param,data -> log.info("获取到的数据"+ data));
     }

mhfire-service/src/main/java/com/bizmatics/mhfire/service/listener/MqttListener.java

@@ -1,10 +1,11 @@
 package com.bizmatics.mhfire.service.listener;
 
 import com.bizmatics.common.spring.util.JsonUtils;
-import com.bizmatics.mhfire.model.WaterAj;
-import com.bizmatics.mhfire.service.WaterAjService;
+import com.bizmatics.mhfire.model.*;
+import com.bizmatics.mhfire.service.*;
 import com.bizmatics.mhfire.service.api.mhWater.OneCardApi;
 import com.bizmatics.mhfire.service.config.mqtt.MqttInConfig;
+import com.fasterxml.jackson.core.type.TypeReference;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
@@ -13,6 +14,7 @@ import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.stereotype.Component;
 
+import java.util.List;
 import java.util.Map;
 
 /**
@@ -29,6 +31,14 @@ public class MqttListener {
     private WaterAjService waterAjService;
 
     public static final String MESSAGE_NAME = "messageInput";
+    @Autowired
+    private DeviceService deviceService;
+    @Autowired
+    private DeviceInfoService deviceInfoService;
+    @Autowired
+    private DeviceAlertService deviceAlertService;
+    @Autowired
+    private DeviceAjService deviceAjService;
 
     /**
      * 处理消息-消费者
@@ -45,15 +55,27 @@ public class MqttListener {
             if (null != mqttReceivedTopic){
                 String topic = mqttReceivedTopic.toString();
                 Map map = JsonUtils.fromJson(payload, Map.class);
-                if (topic.equals("mh/water/info")){
+                if ("mh/water/info".equals(topic)){
                     oneCardApi.callInfoApi(map);
-                }else if (topic.equals("mh/water/alert")){
+                }else if ("mh/water/alert".equals(topic)){
                     oneCardApi.callAlarmApi(map);
-                }else if (topic.equals("mh/water/statistics")){
-                    log.info("城运对接数据------开始");
+                }else if ("mh/water/statistics".equals(topic)){
                     WaterAj waterAj = JsonUtils.fromJson(payload, WaterAj.class);
                     waterAjService.saveOrUpdate(waterAj);
-                    log.info("城运对接数据------结束");
+                }else if ("device/detail".equals(topic)){
+                    //设备录入
+                    List<Device> devices = JsonUtils.fromJson(payload, new TypeReference<List<Device>>() {
+                    });
+                    deviceService.saveOrUpdateBatch(devices);
+                }else if ("device/info".equals(topic)){
+                    DeviceInfo deviceInfo = JsonUtils.fromJson(payload, DeviceInfo.class);
+                    deviceInfoService.save(deviceInfo);
+                }else if ("device/alert".equals(topic)){
+                    DeviceAlert deviceAlert = JsonUtils.fromJson(payload, DeviceAlert.class);
+                    deviceAlertService.save(deviceAlert);
+                }else if ("device/aj".equals(topic)){
+                    DeviceAj deviceAj = JsonUtils.fromJson(payload, DeviceAj.class);
+                    deviceAjService.saveOrUpdate(deviceAj);
                 }
             }
         };