add mqtt module

xueyi-api/xueyi-api-system/src/main/java/com/xueyi/system/api/digitalmans/domain/po/DmVisitRecordsPo.java

@@ -27,6 +27,10 @@ public class DmVisitRecordsPo extends TBaseEntity {
     @Serial
     private static final long serialVersionUID = 1L;
 
+    public static final Integer STATUS_VISITING = 0;
+    public static final Integer STATUS_VISITED = 8;
+    public static final Integer STATUS_EXPIRED = 9;
+
     /** 访客ID */
     @Excel(name = "访客ID")
     protected Long visitorId;

xueyi-common/pom.xml

@@ -20,6 +20,7 @@
         <module>xueyi-common-datascope</module>
 		<module>xueyi-common-datasource</module>
 		<module>xueyi-common-sms</module>
+		<module>xueyi-common-mqtt</module>
     </modules>
 
     <artifactId>xueyi-common</artifactId>

xueyi-common/xueyi-common-mqtt/pom.xml

@@ -0,0 +1,34 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xmlns="http://maven.apache.org/POM/4.0.0"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <groupId>com.xueyi</groupId>
+        <artifactId>xueyi-common</artifactId>
+        <version>2.5.0</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+    
+    <artifactId>xueyi-common-mqtt</artifactId>
+	
+    <description>
+        xueyi-common-mqtt通信服务
+    </description>
+
+    <dependencies>
+		
+        <!-- SpringBoot Boot Mqtt -->
+        <dependency>
+            <groupId>org.eclipse.paho</groupId>
+            <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
+            <version>1.2.5</version>
+        </dependency>
+        
+        <!-- XueYi Common Core-->
+        <dependency>
+            <groupId>com.xueyi</groupId>
+            <artifactId>xueyi-common-core</artifactId>
+        </dependency>
+        
+    </dependencies>
+</project>

xueyi-common/xueyi-common-mqtt/src/main/java/com/xueyi/common/mqtt/configure/MqttConfig.java

@@ -0,0 +1,50 @@
+package com.xueyi.common.mqtt.configure;
+
+import lombok.Data;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.cache.annotation.EnableCaching;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * mqtt配置
+ *
+ * @author xueyi
+ */
+@Configuration
+@EnableCaching
+@Data
+public class MqttConfig {
+    @Value("${emqx.brokerUrl}")
+    private String brokerUrl;
+
+    @Value("${emqx.clientId}")
+    private String clientId;
+
+    @Value("${emqx.username}")
+    private String username;
+
+    @Value("${emqx.password}")
+    private String password;
+
+
+    @Value("${emqx.topics}")
+    private String topics;
+
+    @Value("${emqx.fetchLogTopic}")
+    private String fetchLogTopic;
+
+    @Bean
+    public MqttClient mqttClient() throws Exception {
+        MqttConnectOptions options = new MqttConnectOptions();
+        options.setCleanSession(true);
+        options.setUserName(username);
+        options.setPassword(password.toCharArray());
+        MqttClient client = new MqttClient(brokerUrl, clientId);
+        client.connect(options);
+        return client;
+    }
+
+}

xueyi-common/xueyi-common-mqtt/src/main/java/com/xueyi/common/mqtt/service/MqttMessageHandler.java

@@ -0,0 +1,10 @@
+package com.xueyi.common.mqtt.service;
+
+/**
+ * @author yk
+ * @description
+ * @date 2023-05-10 19:05
+ */
+public interface MqttMessageHandler {
+    void handleMessage(String topic, String message);
+}

xueyi-common/xueyi-common-mqtt/src/main/java/com/xueyi/common/mqtt/service/MqttMessageHandlerService.java

@@ -0,0 +1,22 @@
+package com.xueyi.common.mqtt.service;
+
+/**
+ * @author yk
+ * @description
+ * @date 2023-05-10 19:19
+ */
+public class MqttMessageHandlerService{
+
+    private final MqttService mqttService;
+    private final MqttMessageHandler mqttMessageHandler;
+
+    public MqttMessageHandlerService(MqttService mqttService, MqttMessageHandler mqttMessageHandler) {
+        this.mqttService = mqttService;
+        this.mqttMessageHandler = mqttMessageHandler;
+        init();
+    }
+
+    private void init() {
+        mqttService.addObserver(mqttMessageHandler);
+    }
+}

xueyi-common/xueyi-common-mqtt/src/main/java/com/xueyi/common/mqtt/service/MqttService.java

@@ -0,0 +1,84 @@
+package com.xueyi.common.mqtt.service;
+
+import com.baomidou.mybatisplus.core.toolkit.StringUtils;
+import com.xueyi.common.mqtt.configure.MqttConfig;
+import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
+import org.eclipse.paho.client.mqttv3.MqttCallback;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttException;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * spring mqtt 工具类
+ *
+ * @author xueyi
+ **/
+@Component
+@SuppressWarnings(value = {"unchecked", "rawtypes"})
+public class MqttService implements MqttCallback{
+
+    private List<MqttMessageHandler> observers = new ArrayList<>();
+
+    public void addObserver(MqttMessageHandler observer) {
+        observers.add(observer);
+    }
+
+    public void removeObserver(MqttMessageHandler observer) {
+        observers.remove(observer);
+    }
+    @Autowired
+    private MqttClient mqttClient;
+
+    @Autowired
+    private MqttConfig mqttConfig;
+
+    /**
+     * 发布消息到指定 topic
+     *
+     * @param topic 消息主题
+     * @param message 消息内容
+     */
+    public void sendMessage(String topic, Object message) throws MqttException {
+        MqttMessage msg = new MqttMessage();
+        msg.setPayload(message.toString().getBytes());
+        mqttClient.publish(topic, msg);
+        System.out.println("发送消息到 topic: " + topic);
+        System.out.println("消息内容: " + message);
+    }
+
+
+    @PostConstruct
+    public void subscribe() throws MqttException {
+        mqttClient.setCallback(this);
+        for (String topic : Arrays.stream(mqttConfig.getTopics().split(",")).toList()) {
+            if (StringUtils.isNotEmpty(topic)) {
+                mqttClient.subscribe(topic, 0);
+            }
+        }
+    }
+
+    @Override
+    public void connectionLost(Throwable throwable) {
+
+    }
+
+    @Override
+    public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
+        for (MqttMessageHandler observer : observers) {
+            observer.handleMessage(s, new String(mqttMessage.getPayload(), StandardCharsets.UTF_8));
+        }
+    }
+
+    @Override
+    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
+
+    }
+}

xueyi-common/xueyi-common-mqtt/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -0,0 +1,2 @@
+com.xueyi.common.mqtt.configure.MqttConfig
+com.xueyi.common.mqtt.service.MqttService

xueyi-modules/xueyi-system/pom.xml

@@ -75,6 +75,12 @@
             <artifactId>xueyi-common-sms</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>com.xueyi</groupId>
+            <artifactId>xueyi-common-mqtt</artifactId>
+            <version>2.5.0</version>
+        </dependency>
+
 
         <dependency>
             <groupId>com.konghq</groupId>

xueyi-modules/xueyi-system/src/main/java/com/xueyi/system/exception/controller/DmExceptionLogController.java

@@ -1,18 +1,33 @@
 package com.xueyi.system.exception.controller;
 
+import com.alibaba.fastjson2.JSONObject;
 import com.xueyi.common.core.web.result.AjaxResult;
 import com.xueyi.common.core.web.validate.V_A;
 import com.xueyi.common.core.web.validate.V_E;
 import com.xueyi.common.log.annotation.Log;
 import com.xueyi.common.log.enums.BusinessType;
+import com.xueyi.common.mqtt.configure.MqttConfig;
+import com.xueyi.common.mqtt.service.MqttMessageHandlerService;
+import com.xueyi.common.mqtt.service.MqttService;
 import com.xueyi.common.security.annotation.RequiresPermissions;
 import com.xueyi.common.web.entity.controller.BaseController;
 import com.xueyi.system.exception.domain.dto.DmExceptionLogDto;
 import com.xueyi.system.exception.domain.query.DmExceptionLogQuery;
 import com.xueyi.system.exception.service.IDmExceptionLogService;
+import com.xueyi.system.exception.service.LogMqttMessageHandler;
+import org.eclipse.paho.client.mqttv3.MqttException;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.validation.annotation.Validated;
-import org.springframework.web.bind.annotation.*;
+import org.springframework.web.bind.annotation.DeleteMapping;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.PutMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
 
+import javax.annotation.PostConstruct;
 import java.io.Serializable;
 import java.util.List;
 
@@ -25,12 +40,27 @@ import java.util.List;
 @RequestMapping("/log")
 public class DmExceptionLogController extends BaseController<DmExceptionLogQuery, DmExceptionLogDto, IDmExceptionLogService> {
 
+    @Autowired
+    private MqttService mqttService;
+
+    @Autowired
+    private  LogMqttMessageHandler logMqttMessageHandler;
+
+    @Autowired
+    private MqttConfig mqttConfig;
+
     /** 定义节点名称 */
     @Override
     protected String getNodeName() {
         return "数字人状态事件上报日志" ;
     }
 
+    @PostConstruct
+    private void init() {
+        System.err.println(System.nanoTime()+"-------heelo");
+        MqttMessageHandlerService mqttMessageHandlerService = new MqttMessageHandlerService(mqttService, logMqttMessageHandler);
+    }
+
     /**
      * 查询数字人状态事件上报日志列表
      */
@@ -84,6 +114,27 @@ public class DmExceptionLogController extends BaseController<DmExceptionLogQuery
         return super.batchRemove(idList);
     }
 
+
+    /*
+     * @Author yangkai
+     * @Description //主动拉取日志操作
+     * @Date 2023/5/10
+     * @Param [devId, uploadType]
+     * @return com.xueyi.common.core.web.result.AjaxResult
+     **/
+    @PostMapping("/fetch-log")
+    public AjaxResult sendMqttMsgToDev(@RequestBody JSONObject json){
+        JSONObject jsonObject = new JSONObject();
+        jsonObject.put("device_id", "12345");
+        try {
+            mqttService.sendMessage(mqttConfig.getFetchLogTopic(), jsonObject.toJSONString());
+        } catch (MqttException e) {
+            e.printStackTrace();
+            return error(e.getMessage());
+        }
+        return success("操作成功");
+    }
+
     /**
      * 获取数字人状态事件上报日志选择框列表
      */

xueyi-modules/xueyi-system/src/main/java/com/xueyi/system/exception/service/LogMqttMessageHandler.java

@@ -0,0 +1,19 @@
+package com.xueyi.system.exception.service;
+
+import com.xueyi.common.mqtt.service.MqttMessageHandler;
+import org.springframework.stereotype.Service;
+
+/**
+ * @author yk
+ * @description
+ * @date 2023-05-10 19:08
+ */
+@Service
+public class LogMqttMessageHandler implements MqttMessageHandler {
+
+    @Override
+    public void handleMessage(String topic, String message) {
+        // 处理消息的业务逻辑
+        System.out.println("收到主题eee：" + topic + "，消息内容eee：" + message);
+    }
+}

xueyi-modules/xueyi-system/src/main/java/com/xueyi/system/staff/controller/api/DmVisitorInnerApiController.java

@@ -39,6 +39,7 @@ import org.springframework.web.bind.annotation.RequestParam;
 import org.springframework.web.bind.annotation.RestController;
 
 import java.util.ArrayList;
+import java.util.Date;
 import java.util.List;
 import java.util.stream.Collectors;
 
@@ -77,7 +78,9 @@ public class DmVisitorInnerApiController extends BaseApiController {
             return output(ResponseCode.NOT_FOUND_VISITOR).toJSON();
         } else {
 
-            json.put("id", v.getId());
+            v.setRecordStatus(DmVisitRecordsDto.STATUS_VISITED);
+            v.setUpdateTime(DateUtils.dateToLocalDateTime(new Date()));
+            dmVisitRecordsMapper.updateById(v);
             json.put("empId", v.getUserId());
             json.put("visitorName", visitorName);
             DmStaffPo e = dmStaffMapper.selectById(v.getUserId());