Merge branch 'develop' of http://192.168.11.14:51037/gitadmin/tuoheng_air_monitor into develop

tuoheng-service/tuoheng-admin/pom.xml

@@ -161,6 +161,20 @@
             <version>2.1.7</version>
         </dependency>
 
+        <!--mqtt 相关依赖-->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-integration</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-stream</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-mqtt</artifactId>
+        </dependency>
+
     </dependencies>
 
     <!-- 环境变量配置 -->

tuoheng-service/tuoheng-admin/src/main/java/com/tuoheng/admin/entity/domain/AirDataType.java

@@ -28,6 +28,11 @@ public class AirDataType extends BaseEntity {
      */
     private String name;
 
+    /**
+     * 检测方式
+     */
+    private String detection;
+
     /**
      * 参照指标图片地址
      */

tuoheng-service/tuoheng-admin/src/main/java/com/tuoheng/admin/mqtt/consumer/MqttConsumerCallBack.java

@@ -0,0 +1,45 @@
+package com.tuoheng.admin.mqtt.consumer;
+
+
+import com.alibaba.fastjson.JSONObject;
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
+import org.eclipse.paho.client.mqttv3.MqttCallback;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+
+@Slf4j
+public class MqttConsumerCallBack implements MqttCallback {
+
+    /**
+     * 客户端断开连接的回调
+     */
+    @Override
+    public void connectionLost(Throwable throwable) {
+        System.out.println("与服务器断开连接，可重连");
+        /*MqttProviderConfig client = SpringUtil.getBean(MqttProviderConfig.class);
+        System.out.println(client.isconnect());
+        if (!client.isconnect()) {
+            client.connect();
+            System.out.println("重连成功！");
+        }*/
+    }
+
+    /**
+     * 消息到达的回调
+     */
+    @Override
+    public void messageArrived(String topic, MqttMessage message) throws Exception {
+        log.info("接收消息主题 : {}\n{}", topic, JSONObject.parse(message.getPayload()));
+    }
+
+    /**
+     * 消息发布成功的回调
+     */
+    @Override
+    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
+
+    }
+
+
+}
+

tuoheng-service/tuoheng-admin/src/main/java/com/tuoheng/admin/mqtt/provider/MqttProviderCallBack.java

@@ -0,0 +1,44 @@
+package com.tuoheng.admin.mqtt.provider;
+
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.IMqttAsyncClient;
+import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
+import org.eclipse.paho.client.mqttv3.MqttCallback;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Configuration;
+
+@Slf4j
+@Configuration
+public class MqttProviderCallBack implements MqttCallback {
+
+    @Value("${spring.mqtt.client.id}")
+    private String clientId;
+
+    /**
+     * 与服务器断开的回调
+     */
+    @Override
+    public void connectionLost(Throwable cause) {
+       log.info(clientId + "与服务器断开连接");
+    }
+
+    /**
+     * 消息到达的回调
+     */
+    @Override
+    public void messageArrived(String topic, MqttMessage message) throws Exception {
+
+    }
+
+    /**
+     * 消息发布成功的回调
+     */
+    @Override
+    public void deliveryComplete(IMqttDeliveryToken token) {
+        IMqttAsyncClient client = token.getClient();
+        log.info(client.getClientId() + "发布消息成功！");
+    }
+
+}
+

tuoheng-service/tuoheng-admin/src/main/java/com/tuoheng/admin/mqtt/provider/MqttProviderConfig.java

@@ -0,0 +1,149 @@
+package com.tuoheng.admin.mqtt.provider;
+
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.tuoheng.admin.mqtt.consumer.MqttConsumerCallBack;
+import com.tuoheng.common.core.utils.IpUtils;
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.*;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Configuration;
+
+import javax.annotation.PostConstruct;
+
+@Slf4j
+@Configuration
+public class MqttProviderConfig {
+    @Value("${spring.mqtt.username}")
+    private String username;
+
+    @Value("${spring.mqtt.password}")
+    private String password;
+
+    @Value("${spring.mqtt.url}")
+    private String hostUrl;
+
+    @Value("${spring.mqtt.client.id}")
+    private String clientId;
+
+    @Value("${spring.mqtt.default.topic}")
+    private String defaultTopic;
+
+    /**
+     * 客户端对象
+     */
+    private MqttClient client;
+
+    /**
+     * 在bean初始化后连接到服务器
+     */
+    @PostConstruct
+    public void init() {
+        connect();
+    }
+
+    /**
+     * 客户端连接服务端
+     */
+    public void connect() {
+        try {
+            String ip = IpUtils.getHostIp();
+            //创建MQTT客户端对象
+            client = new MqttClient(hostUrl, ip, new MemoryPersistence());
+            //连接设置
+            MqttConnectOptions options = new MqttConnectOptions();
+            //是否清空session，设置false表示服务器会保留客户端的连接记录（订阅主题，qos）,客户端重连之后能获取到服务器在客户端断开连接期间推送的消息
+            //设置为true表示每次连接服务器都是以新的身份
+            options.setCleanSession(true);
+            //设置连接用户名
+            options.setUserName(username);
+            //设置连接密码
+            options.setPassword(password.toCharArray());
+            //设置超时时间，单位为秒
+            options.setConnectionTimeout(100);
+            //设置心跳时间 单位为秒，表示服务器每隔 1.5*20秒的时间向客户端发送心跳判断客户端是否在线
+            options.setKeepAliveInterval(20);
+            //设置遗嘱消息的话题，若客户端和服务器之间的连接意外断开，服务器将发布客户端的遗嘱信息
+            options.setWill("willTopic", (ip + "与服务器断开连接").getBytes(), 0, false);
+
+            //防止 ERROR o.e.p.c.mqttv3.internal.ClientState - Timed out as no activity  错误
+            options.setConnectionTimeout(0);
+
+            //mqttClient.reconnect(); 这个方法或者回调已经设置了重连
+            //options.setAutomaticReconnect(true);
+            //设置回调
+            client.setCallback(new MqttConsumerCallBack());
+            client.connect(options);
+
+//            client.subscribe("/v1/" +airport.getEdgeId()+"/data/DoorMotor",0);
+
+            QueryWrapper wrapper = new QueryWrapper();
+            wrapper.eq("mark", 1);
+        } catch (MqttException e) {
+            e.printStackTrace();
+        }
+    }
+
+    public boolean isconnect() {
+        try {
+            String ip = IpUtils.getHostIp();
+            //创建MQTT客户端对象
+            client = new MqttClient(hostUrl, ip, new MemoryPersistence());
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+        return client.isConnected();
+    }
+
+    public MqttDeliveryToken publish(int qos, boolean retained, String topic, String message) {
+        if (!client.isConnected()) {
+            connect();
+        }
+        MqttDeliveryToken token = null;
+        MqttMessage mqttMessage = new MqttMessage();
+        mqttMessage.setQos(qos);
+        mqttMessage.setRetained(retained);
+        mqttMessage.setPayload(message.getBytes());
+        //主题的目的地，用于发布/订阅信息
+        MqttTopic mqttTopic = client.getTopic(topic);
+        //提供一种机制来跟踪消息的传递进度
+        //用于在以非阻塞方式（在后台运行）执行发布是跟踪消息的传递进度
+        try {
+            //将指定消息发布到主题，但不等待消息传递完成，返回的token可用于跟踪消息的传递状态
+            //一旦此方法干净地返回，消息就已被客户端接受发布，当连接可用，将在后台完成消息传递。
+            token = mqttTopic.publish(mqttMessage);
+            token.waitForCompletion();
+        } catch (MqttException e) {
+            e.printStackTrace();
+        }
+        return token;
+    }
+
+    /**
+     * 断开连接
+     */
+    public void disConnect() {
+        try {
+            client.disconnect();
+        } catch (MqttException e) {
+            e.printStackTrace();
+        }
+    }
+
+    /**
+     * 订阅主题
+     */
+    public void subscribe(String topic, int qos) {
+        try {
+            client.subscribe(topic, qos);
+        } catch (MqttException e) {
+            e.printStackTrace();
+        }
+    }
+
+    public MqttClient getClient() {
+        return client;
+    }
+
+}
+

tuoheng-service/tuoheng-admin/src/main/java/com/tuoheng/admin/service/airData/report/ExportReportService.java

@@ -96,10 +96,7 @@ public class ExportReportService {
             if (ObjectUtil.isNotEmpty(endAirData)) {
                 detectionTime += endAirData.getCUtcTime();
             }
-
-            String analyticalMethods = "激光散射";
-
-            result = generateReportWordService.buildWord(filePath, request, airDataType, detectionTime, inspection.getEquipmentMountCode(), analyticalMethods);
+            result = generateReportWordService.buildWord(filePath, request, airDataType, detectionTime, inspection.getEquipmentMountCode());
             if (0 != result.getCode()) {
                 log.info("导出报告业务：生成word失败：{}", result.getMsg());
                 return result;

tuoheng-service/tuoheng-admin/src/main/java/com/tuoheng/admin/service/airData/report/GenerateReportWordService.java

@@ -25,10 +25,10 @@ public class GenerateReportWordService {
      * @param airDataType
      * @param detectionTime
      * @param equipmentMountCode
-     * @param analyticalMethods
+     *
      * @return
      */
-    public JsonResult buildWord(String filepath, ExportReportRequest request, AirDataType airDataType, String detectionTime, String equipmentMountCode, String analyticalMethods) {
+    public JsonResult buildWord(String filepath, ExportReportRequest request, AirDataType airDataType, String detectionTime, String equipmentMountCode) {
         WordUtils wordUtils = new WordUtils();
         wordUtils.getDocument().setPageSize(PageSize.A4);
         wordUtils.getDocument().setMargins(71f, 71f, 72f, 72f);
@@ -43,7 +43,7 @@ public class GenerateReportWordService {
 
             wordUtils.insertContext("检测时间:" + detectionTime, 10, Font.NORMAL, Element.ALIGN_LEFT, 20, 0, 0);
             wordUtils.insertContext("监测设备:" + equipmentMountCode, 10, Font.NORMAL, Element.ALIGN_LEFT, 20, 0, 0);
-            wordUtils.insertContext("分析方法:" + analyticalMethods, 10, Font.NORMAL, Element.ALIGN_LEFT, 20, 0, 0);
+            wordUtils.insertContext("检测方式:" + airDataType.getDetection(), 10, Font.NORMAL, Element.ALIGN_LEFT, 20, 0, 0);
             wordUtils.insertContext("采样点数量:" + request.getSamplingPointsNum(), 10, Font.NORMAL, Element.ALIGN_LEFT, 20, 0, 0);
             wordUtils.insertContext("检测区域网格平均尺寸:" + request.getAverageGridSize(), 10, Font.NORMAL, Element.ALIGN_LEFT, 20, 0, 0);
             wordUtils.insertContext("总检测区域网格面积:" + request.getGridArea(), 10, Font.NORMAL, Element.ALIGN_LEFT, 20, 0, 0);

tuoheng-service/tuoheng-admin/src/main/resources/application-dev.yml

@@ -105,6 +105,21 @@ spring:
         max-idle: 10      # 连接池中的最大空闲连接
         min-idle: 1       # 连接池中的最小空闲连接
 
+  mqtt:
+    onlineUrl: http://192.168.11.22:18083
+    #MQTT服务地址，端口号默认11883，如果有多个，用逗号隔开
+    url: tcp://192.168.11.22:1883
+    #用户名
+    username: admin
+    #密码
+    password: public
+    #客户端id(不能重复)
+    client:
+      id: provider-id
+    #MQTT默认的消息推送主题，实际可在调用接口是指定
+    default:
+      topic: topic
+
   servlet:
     multipart:
       # 过滤springmvc的文件上传

tuoheng-service/tuoheng-admin/src/main/resources/application-local.yml

@@ -116,6 +116,21 @@ spring:
         max-idle: 10      # 连接池中的最大空闲连接
         min-idle: 1       # 连接池中的最小空闲连接
 
+  mqtt:
+    onlineUrl: http://127.0.0.1:18083
+    #MQTT服务地址，端口号默认11883，如果有多个，用逗号隔开
+    url: tcp://127.0.0.1:1883
+    #用户名
+    username: admin
+    #密码
+    password: admin##123
+    #客户端id(不能重复)
+    client:
+      id: provider-id
+    #MQTT默认的消息推送主题，实际可在调用接口是指定
+    default:
+      topic: topic
+
   servlet:
     multipart:
       # 过滤springmvc的文件上传

tuoheng-service/tuoheng-admin/src/main/resources/application-prod.yml

@@ -105,6 +105,21 @@ spring:
         max-idle: 10      # 连接池中的最大空闲连接
         min-idle: 1       # 连接池中的最小空闲连接
 
+  mqtt:
+    onlineUrl: https://mqtt.t-aaron.com
+    #MQTT服务地址，端口号默认11883，如果有多个，用逗号隔开
+    url: tcp://mqtt.t-aaron.com:10883
+    #用户名
+    username: admin
+    #密码
+    password: admin##123
+    #客户端id(不能重复)
+    client:
+      id: provider-id
+    #MQTT默认的消息推送主题，实际可在调用接口是指定
+    default:
+      topic: topic
+
   servlet:
     multipart:
       # 过滤springmvc的文件上传

tuoheng-service/tuoheng-admin/src/main/resources/application-test.yml

@@ -106,6 +106,21 @@ spring:
         max-idle: 10      # 连接池中的最大空闲连接
         min-idle: 1       # 连接池中的最小空闲连接
 
+  mqtt:
+    onlineUrl: http://106.15.120.154:18083
+    #MQTT服务地址，端口号默认11883，如果有多个，用逗号隔开
+    url: tcp://106.15.120.154:1883
+    #用户名
+    username: admin
+    #密码
+    password: admin##123
+    #客户端id(不能重复)
+    client:
+      id: provider-id
+    #MQTT默认的消息推送主题，实际可在调用接口是指定
+    default:
+      topic: topic
+
   servlet:
     multipart:
       # 过滤springmvc的文件上传