BJ_ZTEITS-Developers / parking-cloud-yudao

Commit f58cf282dd0417b95c17ecd979017cf237d223e9

Authored by haohao
1 parent d7c57f50

feat：【IoT 物联网】新增 MQTT 协议支持，包含上行和下行消息处理器，完善设备认证和属性、事件、服务处理逻辑，更新配置以启用 EMQX 组件

Inline Side-by-side
Showing 12 changed files with 1157 additions and 3 deletions
yudao-module-iot/yudao-module-iot-gateway/pom.xml
  View file @f58cf28
@@ -41,6 +41,12 @@ @@ -41,6 +41,12 @@
41 <groupId>io.vertx</groupId> 41 <groupId>io.vertx</groupId>
42 <artifactId>vertx-web</artifactId> 42 <artifactId>vertx-web</artifactId>
43 </dependency> 43 </dependency>
  44 +
  45 + <!-- MQTT 相关 -->
  46 + <dependency>
  47 + <groupId>io.vertx</groupId>
  48 + <artifactId>vertx-mqtt</artifactId>
  49 + </dependency>
44 </dependencies> 50 </dependencies>
45 51
46 <build> 52 <build>
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayConfiguration.java
  View file @f58cf28
1 package cn.iocoder.yudao.module.iot.gateway.config; 1 package cn.iocoder.yudao.module.iot.gateway.config;
2 2
3 import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus; 3 import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
4 -import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer;  
5 import cn.iocoder.yudao.module.iot.gateway.protocol.http.IotHttpDownstreamSubscriber; 4 import cn.iocoder.yudao.module.iot.gateway.protocol.http.IotHttpDownstreamSubscriber;
6 import cn.iocoder.yudao.module.iot.gateway.protocol.http.IotHttpUpstreamProtocol; 5 import cn.iocoder.yudao.module.iot.gateway.protocol.http.IotHttpUpstreamProtocol;
  6 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttDownstreamSubscriber;
  7 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttUpstreamProtocol;
7 import lombok.extern.slf4j.Slf4j; 8 import lombok.extern.slf4j.Slf4j;
8 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; 9 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
9 import org.springframework.boot.context.properties.EnableConfigurationProperties; 10 import org.springframework.boot.context.properties.EnableConfigurationProperties;
@@ -31,7 +32,27 @@ public class IotGatewayConfiguration { @@ -31,7 +32,27 @@ public class IotGatewayConfiguration {
31 @Bean 32 @Bean
32 public IotHttpDownstreamSubscriber iotHttpDownstreamSubscriber(IotHttpUpstreamProtocol httpUpstreamProtocol, 33 public IotHttpDownstreamSubscriber iotHttpDownstreamSubscriber(IotHttpUpstreamProtocol httpUpstreamProtocol,
33 IotMessageBus messageBus) { 34 IotMessageBus messageBus) {
34 - return new IotHttpDownstreamSubscriber(httpUpstreamProtocol,messageBus); 35 + return new IotHttpDownstreamSubscriber(httpUpstreamProtocol, messageBus);
  36 + }
  37 + }
  38 +
  39 + /**
  40 + * IoT 网关 MQTT 协议配置类
  41 + */
  42 + @Configuration
  43 + @ConditionalOnProperty(prefix = "yudao.iot.gateway.protocol.emqx", name = "enabled", havingValue = "true")
  44 + @Slf4j
  45 + public static class MqttProtocolConfiguration {
  46 +
  47 + @Bean
  48 + public IotMqttUpstreamProtocol iotMqttUpstreamProtocol(IotGatewayProperties gatewayProperties) {
  49 + return new IotMqttUpstreamProtocol(gatewayProperties.getProtocol().getEmqx());
  50 + }
  51 +
  52 + @Bean
  53 + public IotMqttDownstreamSubscriber iotMqttDownstreamSubscriber(IotMqttUpstreamProtocol mqttUpstreamProtocol,
  54 + IotMessageBus messageBus) {
  55 + return new IotMqttDownstreamSubscriber(mqttUpstreamProtocol, messageBus);
35 } 56 }
36 } 57 }
37 58
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttDownstreamSubscriber.java 0 → 100644
  View file @f58cf28
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt;
  2 +
  3 +import cn.hutool.json.JSONObject;
  4 +import cn.hutool.json.JSONUtil;
  5 +import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
  6 +import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
  7 +import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber;
  8 +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
  9 +import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
  10 +import cn.iocoder.yudao.module.iot.gateway.enums.IotDeviceTopicEnum;
  11 +import jakarta.annotation.PostConstruct;
  12 +import lombok.RequiredArgsConstructor;
  13 +import lombok.extern.slf4j.Slf4j;
  14 +
  15 +/**
  16 + * IoT 网关 MQTT 订阅者：接收下行给设备的消息
  17 + *
  18 + * @author 芋道源码
  19 + */
  20 +@RequiredArgsConstructor
  21 +@Slf4j
  22 +public class IotMqttDownstreamSubscriber implements IotMessageSubscriber<IotDeviceMessage> {
  23 +
  24 + private final IotMqttUpstreamProtocol protocol;
  25 + private final IotMessageBus messageBus;
  26 +
  27 + @PostConstruct
  28 + public void init() {
  29 + messageBus.register(this);
  30 + }
  31 +
  32 + @Override
  33 + public String getTopic() {
  34 + return IotDeviceMessageUtils.buildMessageBusGatewayDeviceMessageTopic(protocol.getServerId());
  35 + }
  36 +
  37 + @Override
  38 + public String getGroup() {
  39 + // 保证点对点消费，需要保证独立的 Group，所以使用 Topic 作为 Group
  40 + return getTopic();
  41 + }
  42 +
  43 + @Override
  44 + public void onMessage(IotDeviceMessage message) {
  45 + log.info("[onMessage][接收到下行消息：{}]", message);
  46 +
  47 + try {
  48 + // 根据消息类型处理不同的下行消息
  49 + String messageType = message.getType();
  50 + switch (messageType) {
  51 + case "property":
  52 + handlePropertyMessage(message);
  53 + break;
  54 + case "service":
  55 + handleServiceMessage(message);
  56 + break;
  57 + case "config":
  58 + handleConfigMessage(message);
  59 + break;
  60 + case "ota":
  61 + handleOtaMessage(message);
  62 + break;
  63 + default:
  64 + log.warn("[onMessage][未知的消息类型：{}]", messageType);
  65 + break;
  66 + }
  67 + } catch (Exception e) {
  68 + log.error("[onMessage][处理下行消息失败：{}]", message, e);
  69 + }
  70 + }
  71 +
  72 + /**
  73 + * 处理属性相关消息
  74 + *
  75 + * @param message 设备消息
  76 + */
  77 + private void handlePropertyMessage(IotDeviceMessage message) {
  78 + String identifier = message.getIdentifier();
  79 + String productKey = message.getProductKey();
  80 + String deviceName = message.getDeviceName();
  81 +
  82 + if ("set".equals(identifier)) {
  83 + // 属性设置
  84 + String topic = IotDeviceTopicEnum.buildPropertySetTopic(productKey, deviceName);
  85 + JSONObject payload = buildDownstreamPayload(message, "thing.service.property.set");
  86 + protocol.publishMessage(topic, payload.toString());
  87 + log.info("[handlePropertyMessage][发送属性设置消息][topic: {}]", topic);
  88 + } else if ("get".equals(identifier)) {
  89 + // 属性获取
  90 + String topic = IotDeviceTopicEnum.buildPropertyGetTopic(productKey, deviceName);
  91 + JSONObject payload = buildDownstreamPayload(message, "thing.service.property.get");
  92 + protocol.publishMessage(topic, payload.toString());
  93 + log.info("[handlePropertyMessage][发送属性获取消息][topic: {}]", topic);
  94 + } else {
  95 + log.warn("[handlePropertyMessage][未知的属性操作：{}]", identifier);
  96 + }
  97 + }
  98 +
  99 + /**
  100 + * 处理服务调用消息
  101 + *
  102 + * @param message 设备消息
  103 + */
  104 + private void handleServiceMessage(IotDeviceMessage message) {
  105 + String identifier = message.getIdentifier();
  106 + String productKey = message.getProductKey();
  107 + String deviceName = message.getDeviceName();
  108 +
  109 + String topic = IotDeviceTopicEnum.buildServiceTopic(productKey, deviceName, identifier);
  110 + JSONObject payload = buildDownstreamPayload(message, "thing.service." + identifier);
  111 + protocol.publishMessage(topic, payload.toString());
  112 + log.info("[handleServiceMessage][发送服务调用消息][topic: {}]", topic);
  113 + }
  114 +
  115 + /**
  116 + * 处理配置设置消息
  117 + *
  118 + * @param message 设备消息
  119 + */
  120 + private void handleConfigMessage(IotDeviceMessage message) {
  121 + String productKey = message.getProductKey();
  122 + String deviceName = message.getDeviceName();
  123 +
  124 + String topic = IotDeviceTopicEnum.buildConfigSetTopic(productKey, deviceName);
  125 + JSONObject payload = buildDownstreamPayload(message, "thing.service.config.set");
  126 + protocol.publishMessage(topic, payload.toString());
  127 + log.info("[handleConfigMessage][发送配置设置消息][topic: {}]", topic);
  128 + }
  129 +
  130 + /**
  131 + * 处理 OTA 升级消息
  132 + *
  133 + * @param message 设备消息
  134 + */
  135 + private void handleOtaMessage(IotDeviceMessage message) {
  136 + String productKey = message.getProductKey();
  137 + String deviceName = message.getDeviceName();
  138 +
  139 + String topic = IotDeviceTopicEnum.buildOtaUpgradeTopic(productKey, deviceName);
  140 + JSONObject payload = buildDownstreamPayload(message, "thing.service.ota.upgrade");
  141 + protocol.publishMessage(topic, payload.toString());
  142 + log.info("[handleOtaMessage][发送 OTA 升级消息][topic: {}]", topic);
  143 + }
  144 +
  145 + /**
  146 + * 构建下行消息载荷
  147 + *
  148 + * @param message 设备消息
  149 + * @param method 方法名
  150 + * @return JSON 载荷
  151 + */
  152 + private JSONObject buildDownstreamPayload(IotDeviceMessage message, String method) {
  153 + JSONObject payload = new JSONObject();
  154 + payload.set("id", message.getMessageId());
  155 + payload.set("version", "1.0");
  156 + payload.set("method", method);
  157 + payload.set("params", message.getData());
  158 + return payload;
  159 + }
  160 +
  161 +}
0 \ No newline at end of file 162 \ No newline at end of file
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttUpstreamProtocol.java 0 → 100644
  View file @f58cf28
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt;
  2 +
  3 +import cn.hutool.core.collection.CollUtil;
  4 +import cn.hutool.core.util.ArrayUtil;
  5 +import cn.hutool.core.util.IdUtil;
  6 +import cn.hutool.core.util.ObjUtil;
  7 +import cn.hutool.core.util.StrUtil;
  8 +import cn.hutool.extra.spring.SpringUtil;
  9 +import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer;
  10 +import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
  11 +import cn.iocoder.yudao.module.iot.gateway.config.IotGatewayProperties;
  12 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router.IotMqttAuthRouter;
  13 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router.IotMqttUpstreamRouter;
  14 +import io.netty.handler.codec.mqtt.MqttQoS;
  15 +import io.vertx.core.Vertx;
  16 +import io.vertx.mqtt.MqttClient;
  17 +import io.vertx.mqtt.MqttClientOptions;
  18 +import jakarta.annotation.PostConstruct;
  19 +import jakarta.annotation.PreDestroy;
  20 +import lombok.RequiredArgsConstructor;
  21 +import lombok.extern.slf4j.Slf4j;
  22 +
  23 +import java.util.List;
  24 +import java.util.concurrent.CompletableFuture;
  25 +import java.util.concurrent.TimeUnit;
  26 +
  27 +/**
  28 + * IoT 网关 MQTT 协议：接收设备上行消息
  29 + *
  30 + * @author 芋道源码
  31 + */
  32 +@RequiredArgsConstructor
  33 +@Slf4j
  34 +public class IotMqttUpstreamProtocol {
  35 +
  36 + /**
  37 + * 默认 QoS 级别
  38 + */
  39 + private static final MqttQoS DEFAULT_QOS = MqttQoS.AT_LEAST_ONCE;
  40 +
  41 + private final IotGatewayProperties.EmqxProperties emqxProperties;
  42 +
  43 + private Vertx vertx;
  44 + private MqttClient mqttClient;
  45 + private IotMqttUpstreamRouter messageRouter;
  46 + private IotMqttAuthRouter authRouter;
  47 + private IotDeviceMessageProducer deviceMessageProducer;
  48 +
  49 + /**
  50 + * 服务运行状态标志
  51 + */
  52 + private volatile boolean isRunning = false;
  53 +
  54 + @PostConstruct
  55 + public void start() {
  56 + if (isRunning) {
  57 + log.warn("[start][MQTT 协议服务已经在运行中，请勿重复启动]");
  58 + return;
  59 + }
  60 + log.info("[start][开始启动 MQTT 协议服务]");
  61 +
  62 + // 初始化组件
  63 + this.vertx = Vertx.vertx();
  64 + this.deviceMessageProducer = SpringUtil.getBean(IotDeviceMessageProducer.class);
  65 + this.messageRouter = new IotMqttUpstreamRouter(this);
  66 + this.authRouter = new IotMqttAuthRouter(this);
  67 +
  68 + // 创建 MQTT 客户端
  69 + MqttClientOptions options = new MqttClientOptions()
  70 + .setClientId("yudao-iot-gateway-" + IdUtil.fastSimpleUUID())
  71 + .setUsername(emqxProperties.getMqttUsername())
  72 + .setPassword(emqxProperties.getMqttPassword())
  73 + .setSsl(ObjUtil.defaultIfNull(emqxProperties.getMqttSsl(), false));
  74 +
  75 + this.mqttClient = MqttClient.create(vertx, options);
  76 +
  77 + // 连接 MQTT Broker
  78 + connectMqtt();
  79 + }
  80 +
  81 + @PreDestroy
  82 + public void stop() {
  83 + if (!isRunning) {
  84 + log.warn("[stop][MQTT 协议服务已经停止，无需再次停止]");
  85 + return;
  86 + }
  87 + log.info("[stop][开始停止 MQTT 协议服务]");
  88 +
  89 + // 1. 取消 MQTT 主题订阅
  90 + if (mqttClient != null && mqttClient.isConnected()) {
  91 + List<String> topicList = emqxProperties.getMqttTopics();
  92 + if (CollUtil.isNotEmpty(topicList)) {
  93 + for (String topic : topicList) {
  94 + try {
  95 + mqttClient.unsubscribe(topic);
  96 + log.debug("[stop][取消订阅主题: {}]", topic);
  97 + } catch (Exception e) {
  98 + log.warn("[stop][取消订阅主题异常: {}]", topic, e);
  99 + }
  100 + }
  101 + }
  102 + }
  103 +
  104 + // 2. 关闭 MQTT 客户端
  105 + try {
  106 + if (mqttClient != null && mqttClient.isConnected()) {
  107 + mqttClient.disconnect();
  108 + }
  109 + } catch (Exception e) {
  110 + log.warn("[stop][关闭 MQTT 客户端异常]", e);
  111 + }
  112 +
  113 + // 3. 关闭 Vertx
  114 + try {
  115 + if (vertx != null) {
  116 + vertx.close();
  117 + }
  118 + } catch (Exception e) {
  119 + log.warn("[stop][关闭 Vertx 异常]", e);
  120 + }
  121 +
  122 + // 4. 更新状态
  123 + isRunning = false;
  124 + log.info("[stop][MQTT 协议服务已停止]");
  125 + }
  126 +
  127 + /**
  128 + * 连接 MQTT Broker 并订阅主题
  129 + */
  130 + private void connectMqtt() {
  131 + // 检查必要的 MQTT 配置
  132 + String host = emqxProperties.getMqttHost();
  133 + Integer port = emqxProperties.getMqttPort();
  134 + if (StrUtil.isBlank(host)) {
  135 + String msg = "[connectMqtt][MQTT Host 为空，无法连接]";
  136 + log.error(msg);
  137 + return;
  138 + }
  139 + if (port == null) {
  140 + log.warn("[connectMqtt][MQTT Port 为 null，使用默认端口 1883]");
  141 + port = 1883; // 默认 MQTT 端口
  142 + }
  143 +
  144 + final Integer finalPort = port;
  145 + CompletableFuture<Void> connectFuture = mqttClient.connect(finalPort, host)
  146 + .toCompletionStage()
  147 + .toCompletableFuture()
  148 + .thenAccept(connAck -> {
  149 + log.info("[connectMqtt][MQTT 客户端连接成功]");
  150 + // 设置断开重连监听器
  151 + mqttClient.closeHandler(closeEvent -> {
  152 + log.warn("[closeHandler][MQTT 连接已断开，准备重连]");
  153 + reconnectWithDelay();
  154 + });
  155 + // 设置消息处理器
  156 + setupMessageHandler();
  157 + // 订阅主题
  158 + subscribeToTopics();
  159 + })
  160 + .exceptionally(error -> {
  161 + log.error("[connectMqtt][连接 MQTT Broker 失败]", error);
  162 + reconnectWithDelay();
  163 + return null;
  164 + });
  165 +
  166 + // 等待连接完成
  167 + try {
  168 + connectFuture.get(10, TimeUnit.SECONDS);
  169 + isRunning = true;
  170 + log.info("[connectMqtt][MQTT 协议服务启动完成]");
  171 + } catch (Exception e) {
  172 + log.error("[connectMqtt][MQTT 协议服务启动失败]", e);
  173 + }
  174 + }
  175 +
  176 + /**
  177 + * 设置 MQTT 消息处理器
  178 + */
  179 + private void setupMessageHandler() {
  180 + mqttClient.publishHandler(messageRouter::route);
  181 + log.debug("[setupMessageHandler][MQTT 消息处理器设置完成]");
  182 + }
  183 +
  184 + /**
  185 + * 订阅设备上行消息主题
  186 + */
  187 + private void subscribeToTopics() {
  188 + List<String> topicList = emqxProperties.getMqttTopics();
  189 + if (CollUtil.isEmpty(topicList)) {
  190 + log.warn("[subscribeToTopics][未配置 MQTT 主题，使用默认主题]");
  191 + topicList = List.of("/sys/#"); // 默认订阅所有系统主题
  192 + }
  193 +
  194 + for (String topic : topicList) {
  195 + if (StrUtil.isBlank(topic)) {
  196 + log.warn("[subscribeToTopics][跳过空主题]");
  197 + continue;
  198 + }
  199 +
  200 + mqttClient.subscribe(topic, DEFAULT_QOS.value())
  201 + .onSuccess(ack -> log.info("[subscribeToTopics][订阅主题成功: {}]", topic))
  202 + .onFailure(err -> log.error("[subscribeToTopics][订阅主题失败: {}]", topic, err));
  203 + }
  204 + }
  205 +
  206 + /**
  207 + * 重连 MQTT 客户端
  208 + */
  209 + private void reconnectWithDelay() {
  210 + if (!isRunning) {
  211 + log.info("[reconnectWithDelay][服务已停止，不再尝试重连]");
  212 + return;
  213 + }
  214 +
  215 + // 默认重连延迟 5 秒
  216 + int reconnectDelayMs = 5000;
  217 + vertx.setTimer(reconnectDelayMs, id -> {
  218 + log.info("[reconnectWithDelay][开始重新连接 MQTT]");
  219 + connectMqtt();
  220 + });
  221 + }
  222 +
  223 + /**
  224 + * 发布消息到 MQTT
  225 + *
  226 + * @param topic 主题
  227 + * @param payload 消息内容
  228 + */
  229 + public void publishMessage(String topic, String payload) {
  230 + if (mqttClient == null || !mqttClient.isConnected()) {
  231 + log.warn("[publishMessage][MQTT 客户端未连接，无法发送消息][topic: {}]", topic);
  232 + return;
  233 + }
  234 +
  235 + mqttClient.publish(topic, io.vertx.core.buffer.Buffer.buffer(payload), DEFAULT_QOS, false, false)
  236 + .onSuccess(v -> log.debug("[publishMessage][发送消息成功][topic: {}]", topic))
  237 + .onFailure(err -> log.error("[publishMessage][发送消息失败][topic: {}]", topic, err));
  238 + }
  239 +
  240 + /**
  241 + * 获取服务器 ID
  242 + *
  243 + * @return 服务器 ID
  244 + */
  245 + public String getServerId() {
  246 + return IotDeviceMessageUtils.generateServerId(emqxProperties.getMqttPort());
  247 + }
  248 +
  249 + /**
  250 + * 获取 MQTT 客户端
  251 + *
  252 + * @return MQTT 客户端
  253 + */
  254 + public MqttClient getMqttClient() {
  255 + return mqttClient;
  256 + }
  257 +
  258 + /**
  259 + * 获取认证路由器
  260 + *
  261 + * @return 认证路由器
  262 + */
  263 + public IotMqttAuthRouter getAuthRouter() {
  264 + return authRouter;
  265 + }
  266 +
  267 +}
0 \ No newline at end of file 268 \ No newline at end of file
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttAbstractHandler.java 0 → 100644
  View file @f58cf28
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router;
  2 +
  3 +import lombok.extern.slf4j.Slf4j;
  4 +
  5 +/**
  6 + * IoT 网关 MQTT 协议的路由处理器抽象基类
  7 + * <p>
  8 + * 提供通用的处理方法，所有 MQTT 消息处理器都应继承此类
  9 + *
  10 + * @author 芋道源码
  11 + */
  12 +@Slf4j
  13 +public abstract class IotMqttAbstractHandler {
  14 +
  15 + /**
  16 + * 处理 MQTT 消息
  17 + *
  18 + * @param topic 主题
  19 + * @param payload 消息内容
  20 + */
  21 + public abstract void handle(String topic, String payload);
  22 +
  23 + /**
  24 + * 解析主题，获取主题各部分
  25 + *
  26 + * @param topic 主题
  27 + * @return 主题各部分数组，如果解析失败返回 null
  28 + */
  29 + protected String[] parseTopic(String topic) {
  30 + String[] topicParts = topic.split("/");
  31 + if (topicParts.length < 7) {
  32 + log.warn("[parseTopic][主题格式不正确][topic: {}]", topic);
  33 + return null;
  34 + }
  35 + return topicParts;
  36 + }
  37 +
  38 +}
0 \ No newline at end of file 39 \ No newline at end of file
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttAuthRouter.java 0 → 100644
  View file @f58cf28
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router;
  2 +
  3 +import cn.hutool.core.util.StrUtil;
  4 +import cn.hutool.extra.spring.SpringUtil;
  5 +import cn.iocoder.yudao.framework.common.pojo.CommonResult;
  6 +import cn.iocoder.yudao.module.iot.core.biz.IotDeviceCommonApi;
  7 +import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceAuthReqDTO;
  8 +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
  9 +import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer;
  10 +import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils;
  11 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttUpstreamProtocol;
  12 +import cn.iocoder.yudao.module.iot.gateway.service.auth.IotDeviceTokenService;
  13 +import lombok.RequiredArgsConstructor;
  14 +import lombok.extern.slf4j.Slf4j;
  15 +
  16 +/**
  17 + * IoT 网关 MQTT 认证路由器
  18 + * <p>
  19 + * 处理设备的 MQTT 连接认证和连接状态管理
  20 + *
  21 + * @author 芋道源码
  22 + */
  23 +@RequiredArgsConstructor
  24 +@Slf4j
  25 +public class IotMqttAuthRouter {
  26 +
  27 + private final IotMqttUpstreamProtocol protocol;
  28 + private final IotDeviceMessageProducer deviceMessageProducer;
  29 + private final IotDeviceTokenService deviceTokenService;
  30 + private final IotDeviceCommonApi deviceCommonApi;
  31 +
  32 + public IotMqttAuthRouter(IotMqttUpstreamProtocol protocol) {
  33 + this.protocol = protocol;
  34 + this.deviceMessageProducer = SpringUtil.getBean(IotDeviceMessageProducer.class);
  35 + this.deviceTokenService = SpringUtil.getBean(IotDeviceTokenService.class);
  36 + this.deviceCommonApi = SpringUtil.getBean(IotDeviceCommonApi.class);
  37 + }
  38 +
  39 + /**
  40 + * 处理设备认证
  41 + *
  42 + * @param clientId 客户端 ID
  43 + * @param username 用户名
  44 + * @param password 密码
  45 + * @return 认证结果
  46 + */
  47 + public boolean authenticate(String clientId, String username, String password) {
  48 + try {
  49 + log.info("[authenticate][开始认证设备][clientId: {}][username: {}]", clientId, username);
  50 +
  51 + // 1. 参数校验
  52 + if (StrUtil.isEmpty(clientId) || StrUtil.isEmpty(username) || StrUtil.isEmpty(password)) {
  53 + log.warn("[authenticate][认证参数不完整][clientId: {}][username: {}]", clientId, username);
  54 + return false;
  55 + }
  56 +
  57 + // 2. 执行认证
  58 + CommonResult<Boolean> result = deviceCommonApi.authDevice(new IotDeviceAuthReqDTO()
  59 + .setClientId(clientId).setUsername(username).setPassword(password));
  60 + result.checkError();
  61 + if (!Boolean.TRUE.equals(result.getData())) {
  62 + log.warn("[authenticate][设备认证失败][clientId: {}][username: {}]", clientId, username);
  63 + return false;
  64 + }
  65 +
  66 + log.info("[authenticate][设备认证成功][clientId: {}][username: {}]", clientId, username);
  67 + return true;
  68 + } catch (Exception e) {
  69 + log.error("[authenticate][设备认证异常][clientId: {}][username: {}]", clientId, username, e);
  70 + return false;
  71 + }
  72 + }
  73 +
  74 + /**
  75 + * 处理设备连接事件
  76 + *
  77 + * @param clientId 客户端 ID
  78 + * @param username 用户名
  79 + */
  80 + public void handleClientConnected(String clientId, String username) {
  81 + try {
  82 + log.info("[handleClientConnected][设备连接][clientId: {}][username: {}]", clientId, username);
  83 +
  84 + // 解析设备信息并发送上线消息
  85 + handleDeviceStateChange(username, true);
  86 + } catch (Exception e) {
  87 + log.error("[handleClientConnected][处理设备连接事件异常][clientId: {}][username: {}]", clientId, username, e);
  88 + }
  89 + }
  90 +
  91 + /**
  92 + * 处理设备断开连接事件
  93 + *
  94 + * @param clientId 客户端 ID
  95 + * @param username 用户名
  96 + */
  97 + public void handleClientDisconnected(String clientId, String username) {
  98 + try {
  99 + log.info("[handleClientDisconnected][设备断开连接][clientId: {}][username: {}]", clientId, username);
  100 +
  101 + // 解析设备信息并发送下线消息
  102 + handleDeviceStateChange(username, false);
  103 + } catch (Exception e) {
  104 + log.error("[handleClientDisconnected][处理设备断开连接事件异常][clientId: {}][username: {}]", clientId, username, e);
  105 + }
  106 + }
  107 +
  108 + /**
  109 + * 处理设备状态变化
  110 + *
  111 + * @param username 用户名
  112 + * @param online 是否在线
  113 + */
  114 + private void handleDeviceStateChange(String username, boolean online) {
  115 + // 解析设备信息
  116 + if (StrUtil.isEmpty(username) || "undefined".equals(username)) {
  117 + log.warn("[handleDeviceStateChange][用户名为空或未定义][username: {}]", username);
  118 + return;
  119 + }
  120 +
  121 + IotDeviceAuthUtils.DeviceInfo deviceInfo = deviceTokenService.parseUsername(username);
  122 + if (deviceInfo == null) {
  123 + log.warn("[handleDeviceStateChange][无法解析设备信息][username: {}]", username);
  124 + return;
  125 + }
  126 +
  127 + try {
  128 + // 发送设备状态消息
  129 + IotDeviceMessage message = IotDeviceMessage.of(
  130 + deviceInfo.getProductKey(), deviceInfo.getDeviceName(), protocol.getServerId());
  131 +
  132 + if (online) {
  133 + message = message.ofStateOnline();
  134 + log.info("[handleDeviceStateChange][发送设备上线消息成功][username: {}]", username);
  135 + } else {
  136 + message = message.ofStateOffline();
  137 + log.info("[handleDeviceStateChange][发送设备下线消息成功][username: {}]", username);
  138 + }
  139 +
  140 + deviceMessageProducer.sendDeviceMessage(message);
  141 + } catch (Exception e) {
  142 + log.error("[handleDeviceStateChange][发送设备状态消息失败][username: {}][online: {}]", username, online, e);
  143 + }
  144 + }
  145 +
  146 +}
0 \ No newline at end of file 147 \ No newline at end of file
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttEventHandler.java 0 → 100644
  View file @f58cf28
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router;
  2 +
  3 +import cn.hutool.json.JSONObject;
  4 +import cn.hutool.json.JSONUtil;
  5 +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
  6 +import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer;
  7 +import cn.iocoder.yudao.module.iot.gateway.enums.IotDeviceTopicEnum;
  8 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttUpstreamProtocol;
  9 +import lombok.RequiredArgsConstructor;
  10 +import lombok.extern.slf4j.Slf4j;
  11 +
  12 +import java.util.Map;
  13 +
  14 +/**
  15 + * IoT 网关 MQTT 事件处理器
  16 + * <p>
  17 + * 处理设备事件相关的 MQTT 消息
  18 + *
  19 + * @author 芋道源码
  20 + */
  21 +@RequiredArgsConstructor
  22 +@Slf4j
  23 +public class IotMqttEventHandler extends IotMqttAbstractHandler {
  24 +
  25 + private final IotMqttUpstreamProtocol protocol;
  26 + private final IotDeviceMessageProducer deviceMessageProducer;
  27 +
  28 + @Override
  29 + public void handle(String topic, String payload) {
  30 + try {
  31 + log.info("[handle][接收到设备事件上报][topic: {}]", topic);
  32 +
  33 + // 解析消息内容
  34 + JSONObject jsonObject = JSONUtil.parseObj(payload);
  35 + String[] topicParts = parseTopic(topic);
  36 + if (topicParts == null) {
  37 + return;
  38 + }
  39 +
  40 + // 构建设备消息
  41 + String productKey = topicParts[2];
  42 + String deviceName = topicParts[3];
  43 + String eventIdentifier = getEventIdentifier(topicParts, topic);
  44 + if (eventIdentifier == null) {
  45 + return;
  46 + }
  47 +
  48 + Map<String, Object> eventData = parseEventDataFromPayload(jsonObject);
  49 + IotDeviceMessage message = IotDeviceMessage.of(productKey, deviceName, protocol.getServerId());
  50 + // 设置事件消息类型和标识符
  51 + message.setType("event");
  52 + message.setIdentifier(eventIdentifier);
  53 + message.setData(eventData);
  54 +
  55 + // 发送消息
  56 + deviceMessageProducer.sendDeviceMessage(message);
  57 + log.info("[handle][处理设备事件上报成功][topic: {}]", topic);
  58 +
  59 + // 发送响应消息
  60 + String method = "thing.event." + eventIdentifier + ".post";
  61 + sendResponse(topic, jsonObject, method);
  62 + } catch (Exception e) {
  63 + log.error("[handle][处理设备事件上报失败][topic: {}][payload: {}]", topic, payload, e);
  64 + }
  65 + }
  66 +
  67 + /**
  68 + * 从主题部分中获取事件标识符
  69 + *
  70 + * @param topicParts 主题各部分
  71 + * @param topic 原始主题，用于日志
  72 + * @return 事件标识符，如果获取失败返回 null
  73 + */
  74 + private String getEventIdentifier(String[] topicParts, String topic) {
  75 + try {
  76 + return topicParts[6];
  77 + } catch (ArrayIndexOutOfBoundsException e) {
  78 + log.warn("[getEventIdentifier][无法从主题中获取事件标识符][topic: {}]", topic);
  79 + return null;
  80 + }
  81 + }
  82 +
  83 + /**
  84 + * 从消息载荷解析事件数据
  85 + *
  86 + * @param jsonObject 消息 JSON 对象
  87 + * @return 事件数据映射
  88 + */
  89 + private Map<String, Object> parseEventDataFromPayload(JSONObject jsonObject) {
  90 + JSONObject params = jsonObject.getJSONObject("params");
  91 + if (params == null) {
  92 + log.warn("[parseEventDataFromPayload][消息格式不正确，缺少 params 字段][jsonObject: {}]", jsonObject);
  93 + return Map.of();
  94 + }
  95 + return params;
  96 + }
  97 +
  98 + /**
  99 + * 发送响应消息
  100 + *
  101 + * @param topic 原始主题
  102 + * @param jsonObject 原始消息 JSON 对象
  103 + * @param method 响应方法
  104 + */
  105 + private void sendResponse(String topic, JSONObject jsonObject, String method) {
  106 + String replyTopic = IotDeviceTopicEnum.getReplyTopic(topic);
  107 +
  108 + // 构建响应消息
  109 + JSONObject response = new JSONObject();
  110 + response.set("id", jsonObject.getStr("id"));
  111 + response.set("code", 200);
  112 + response.set("method", method);
  113 + response.set("data", new JSONObject());
  114 +
  115 + // 发送响应
  116 + protocol.publishMessage(replyTopic, response.toString());
  117 + log.debug("[sendResponse][发送响应消息][topic: {}]", replyTopic);
  118 + }
  119 +
  120 +}
0 \ No newline at end of file 121 \ No newline at end of file
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttPropertyHandler.java 0 → 100644
  View file @f58cf28
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router;
  2 +
  3 +import cn.hutool.json.JSONObject;
  4 +import cn.hutool.json.JSONUtil;
  5 +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
  6 +import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer;
  7 +import cn.iocoder.yudao.module.iot.gateway.enums.IotDeviceTopicEnum;
  8 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttUpstreamProtocol;
  9 +import lombok.RequiredArgsConstructor;
  10 +import lombok.extern.slf4j.Slf4j;
  11 +
  12 +import java.util.Map;
  13 +
  14 +/**
  15 + * IoT 网关 MQTT 属性处理器
  16 + * <p>
  17 + * 处理设备属性相关的 MQTT 消息
  18 + *
  19 + * @author 芋道源码
  20 + */
  21 +@RequiredArgsConstructor
  22 +@Slf4j
  23 +public class IotMqttPropertyHandler extends IotMqttAbstractHandler {
  24 +
  25 + private final IotMqttUpstreamProtocol protocol;
  26 + private final IotDeviceMessageProducer deviceMessageProducer;
  27 +
  28 + @Override
  29 + public void handle(String topic, String payload) {
  30 + if (topic.endsWith(IotDeviceTopicEnum.PROPERTY_POST_TOPIC.getTopic())) {
  31 + // 属性上报
  32 + handlePropertyPost(topic, payload);
  33 + } else if (topic.contains(IotDeviceTopicEnum.PROPERTY_SET_TOPIC.getTopic())) {
  34 + // 属性设置响应
  35 + handlePropertySetReply(topic, payload);
  36 + } else if (topic.contains(IotDeviceTopicEnum.PROPERTY_GET_TOPIC.getTopic())) {
  37 + // 属性获取响应
  38 + handlePropertyGetReply(topic, payload);
  39 + } else {
  40 + log.warn("[handle][未知的属性主题][topic: {}]", topic);
  41 + }
  42 + }
  43 +
  44 + /**
  45 + * 处理设备属性上报消息
  46 + *
  47 + * @param topic 主题
  48 + * @param payload 消息内容
  49 + */
  50 + private void handlePropertyPost(String topic, String payload) {
  51 + try {
  52 + log.info("[handlePropertyPost][接收到设备属性上报][topic: {}]", topic);
  53 +
  54 + // 解析消息内容
  55 + JSONObject jsonObject = JSONUtil.parseObj(payload);
  56 + String[] topicParts = parseTopic(topic);
  57 + if (topicParts == null) {
  58 + return;
  59 + }
  60 +
  61 + // 构建设备消息
  62 + String productKey = topicParts[2];
  63 + String deviceName = topicParts[3];
  64 + Map<String, Object> properties = parsePropertiesFromPayload(jsonObject);
  65 +
  66 + IotDeviceMessage message = IotDeviceMessage.of(productKey, deviceName, protocol.getServerId())
  67 + .ofPropertyReport(properties);
  68 +
  69 + // 发送消息
  70 + deviceMessageProducer.sendDeviceMessage(message);
  71 + log.info("[handlePropertyPost][处理设备属性上报成功][topic: {}]", topic);
  72 +
  73 + // 发送响应消息
  74 + sendResponse(topic, jsonObject, "thing.event.property.post");
  75 + } catch (Exception e) {
  76 + log.error("[handlePropertyPost][处理设备属性上报失败][topic: {}][payload: {}]", topic, payload, e);
  77 + }
  78 + }
  79 +
  80 + /**
  81 + * 处理属性设置响应消息
  82 + *
  83 + * @param topic 主题
  84 + * @param payload 消息内容
  85 + */
  86 + private void handlePropertySetReply(String topic, String payload) {
  87 + try {
  88 + log.info("[handlePropertySetReply][接收到属性设置响应][topic: {}]", topic);
  89 + // TODO: 处理属性设置响应逻辑
  90 + } catch (Exception e) {
  91 + log.error("[handlePropertySetReply][处理属性设置响应失败][topic: {}][payload: {}]", topic, payload, e);
  92 + }
  93 + }
  94 +
  95 + /**
  96 + * 处理属性获取响应消息
  97 + *
  98 + * @param topic 主题
  99 + * @param payload 消息内容
  100 + */
  101 + private void handlePropertyGetReply(String topic, String payload) {
  102 + try {
  103 + log.info("[handlePropertyGetReply][接收到属性获取响应][topic: {}]", topic);
  104 + // TODO: 处理属性获取响应逻辑
  105 + } catch (Exception e) {
  106 + log.error("[handlePropertyGetReply][处理属性获取响应失败][topic: {}][payload: {}]", topic, payload, e);
  107 + }
  108 + }
  109 +
  110 + /**
  111 + * 从消息载荷解析属性
  112 + *
  113 + * @param jsonObject 消息 JSON 对象
  114 + * @return 属性映射
  115 + */
  116 + private Map<String, Object> parsePropertiesFromPayload(JSONObject jsonObject) {
  117 + JSONObject params = jsonObject.getJSONObject("params");
  118 + if (params == null) {
  119 + log.warn("[parsePropertiesFromPayload][消息格式不正确，缺少 params 字段][jsonObject: {}]", jsonObject);
  120 + return Map.of();
  121 + }
  122 + return params;
  123 + }
  124 +
  125 + /**
  126 + * 发送响应消息
  127 + *
  128 + * @param topic 原始主题
  129 + * @param jsonObject 原始消息 JSON 对象
  130 + * @param method 响应方法
  131 + */
  132 + private void sendResponse(String topic, JSONObject jsonObject, String method) {
  133 + String replyTopic = IotDeviceTopicEnum.getReplyTopic(topic);
  134 +
  135 + // 构建响应消息
  136 + JSONObject response = new JSONObject();
  137 + response.set("id", jsonObject.getStr("id"));
  138 + response.set("code", 200);
  139 + response.set("method", method);
  140 + response.set("data", new JSONObject());
  141 +
  142 + // 发送响应
  143 + protocol.publishMessage(replyTopic, response.toString());
  144 + log.debug("[sendResponse][发送响应消息][topic: {}]", replyTopic);
  145 + }
  146 +
  147 +}
0 \ No newline at end of file 148 \ No newline at end of file
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttServiceHandler.java 0 → 100644
  View file @f58cf28
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router;
  2 +
  3 +import cn.hutool.json.JSONObject;
  4 +import cn.hutool.json.JSONUtil;
  5 +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
  6 +import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer;
  7 +import cn.iocoder.yudao.module.iot.gateway.enums.IotDeviceTopicEnum;
  8 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttUpstreamProtocol;
  9 +import lombok.RequiredArgsConstructor;
  10 +import lombok.extern.slf4j.Slf4j;
  11 +
  12 +import java.util.Map;
  13 +
  14 +/**
  15 + * IoT 网关 MQTT 服务处理器
  16 + * <p>
  17 + * 处理设备服务调用相关的 MQTT 消息
  18 + *
  19 + * @author 芋道源码
  20 + */
  21 +@RequiredArgsConstructor
  22 +@Slf4j
  23 +public class IotMqttServiceHandler extends IotMqttAbstractHandler {
  24 +
  25 + private final IotMqttUpstreamProtocol protocol;
  26 + private final IotDeviceMessageProducer deviceMessageProducer;
  27 +
  28 + @Override
  29 + public void handle(String topic, String payload) {
  30 + try {
  31 + log.info("[handle][接收到设备服务调用响应][topic: {}]", topic);
  32 +
  33 + // 解析消息内容
  34 + JSONObject jsonObject = JSONUtil.parseObj(payload);
  35 + String[] topicParts = parseTopic(topic);
  36 + if (topicParts == null) {
  37 + return;
  38 + }
  39 +
  40 + // 构建设备消息
  41 + String productKey = topicParts[2];
  42 + String deviceName = topicParts[3];
  43 + String serviceIdentifier = getServiceIdentifier(topicParts, topic);
  44 + if (serviceIdentifier == null) {
  45 + return;
  46 + }
  47 +
  48 + Map<String, Object> serviceData = parseServiceDataFromPayload(jsonObject);
  49 + IotDeviceMessage message = IotDeviceMessage.of(productKey, deviceName, protocol.getServerId());
  50 + // 设置服务消息类型和标识符
  51 + message.setType("service");
  52 + message.setIdentifier(serviceIdentifier);
  53 + message.setData(serviceData);
  54 +
  55 + // 发送消息
  56 + deviceMessageProducer.sendDeviceMessage(message);
  57 + log.info("[handle][处理设备服务调用响应成功][topic: {}]", topic);
  58 +
  59 + // 发送响应消息
  60 + String method = "thing.service." + serviceIdentifier;
  61 + sendResponse(topic, jsonObject, method);
  62 + } catch (Exception e) {
  63 + log.error("[handle][处理设备服务调用响应失败][topic: {}][payload: {}]", topic, payload, e);
  64 + }
  65 + }
  66 +
  67 + /**
  68 + * 从主题部分中获取服务标识符
  69 + *
  70 + * @param topicParts 主题各部分
  71 + * @param topic 原始主题，用于日志
  72 + * @return 服务标识符，如果获取失败返回 null
  73 + */
  74 + private String getServiceIdentifier(String[] topicParts, String topic) {
  75 + try {
  76 + // 服务主题格式：/sys/{productKey}/{deviceName}/thing/service/{serviceIdentifier}
  77 + return topicParts[6];
  78 + } catch (ArrayIndexOutOfBoundsException e) {
  79 + log.warn("[getServiceIdentifier][无法从主题中获取服务标识符][topic: {}]", topic);
  80 + return null;
  81 + }
  82 + }
  83 +
  84 + /**
  85 + * 从消息载荷解析服务数据
  86 + *
  87 + * @param jsonObject 消息 JSON 对象
  88 + * @return 服务数据映射
  89 + */
  90 + private Map<String, Object> parseServiceDataFromPayload(JSONObject jsonObject) {
  91 + JSONObject params = jsonObject.getJSONObject("params");
  92 + if (params == null) {
  93 + log.warn("[parseServiceDataFromPayload][消息格式不正确，缺少 params 字段][jsonObject: {}]", jsonObject);
  94 + return Map.of();
  95 + }
  96 + return params;
  97 + }
  98 +
  99 + /**
  100 + * 发送响应消息
  101 + *
  102 + * @param topic 原始主题
  103 + * @param jsonObject 原始消息 JSON 对象
  104 + * @param method 响应方法
  105 + */
  106 + private void sendResponse(String topic, JSONObject jsonObject, String method) {
  107 + String replyTopic = IotDeviceTopicEnum.getReplyTopic(topic);
  108 +
  109 + // 构建响应消息
  110 + JSONObject response = new JSONObject();
  111 + response.set("id", jsonObject.getStr("id"));
  112 + response.set("code", 200);
  113 + response.set("method", method);
  114 + response.set("data", new JSONObject());
  115 +
  116 + // 发送响应
  117 + protocol.publishMessage(replyTopic, response.toString());
  118 + log.debug("[sendResponse][发送响应消息][topic: {}]", replyTopic);
  119 + }
  120 +
  121 +}
0 \ No newline at end of file 122 \ No newline at end of file
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttUpstreamRouter.java 0 → 100644
  View file @f58cf28
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router;
  2 +
  3 +import cn.hutool.core.util.StrUtil;
  4 +import cn.hutool.extra.spring.SpringUtil;
  5 +import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer;
  6 +import cn.iocoder.yudao.module.iot.gateway.enums.IotDeviceTopicEnum;
  7 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttUpstreamProtocol;
  8 +import io.vertx.mqtt.messages.MqttPublishMessage;
  9 +import lombok.RequiredArgsConstructor;
  10 +import lombok.extern.slf4j.Slf4j;
  11 +
  12 +/**
  13 + * IoT 网关 MQTT 上行路由器
  14 + * <p>
  15 + * 根据消息主题路由到不同的处理器
  16 + *
  17 + * @author 芋道源码
  18 + */
  19 +@RequiredArgsConstructor
  20 +@Slf4j
  21 +public class IotMqttUpstreamRouter {
  22 +
  23 + private final IotMqttUpstreamProtocol protocol;
  24 + private final IotDeviceMessageProducer deviceMessageProducer;
  25 +
  26 + // 处理器实例
  27 + private IotMqttPropertyHandler propertyHandler;
  28 + private IotMqttEventHandler eventHandler;
  29 + private IotMqttServiceHandler serviceHandler;
  30 +
  31 + public IotMqttUpstreamRouter(IotMqttUpstreamProtocol protocol) {
  32 + this.protocol = protocol;
  33 + this.deviceMessageProducer = SpringUtil.getBean(IotDeviceMessageProducer.class);
  34 + // 初始化处理器
  35 + this.propertyHandler = new IotMqttPropertyHandler(protocol, deviceMessageProducer);
  36 + this.eventHandler = new IotMqttEventHandler(protocol, deviceMessageProducer);
  37 + this.serviceHandler = new IotMqttServiceHandler(protocol, deviceMessageProducer);
  38 + }
  39 +
  40 + /**
  41 + * 路由 MQTT 消息
  42 + *
  43 + * @param message MQTT 发布消息
  44 + */
  45 + public void route(MqttPublishMessage message) {
  46 + String topic = message.topicName();
  47 + String payload = message.payload().toString();
  48 + log.info("[route][接收到 MQTT 消息][topic: {}][payload: {}]", topic, payload);
  49 +
  50 + try {
  51 + if (StrUtil.isEmpty(payload)) {
  52 + log.warn("[route][消息内容为空][topic: {}]", topic);
  53 + return;
  54 + }
  55 +
  56 + // 根据主题路由到相应的处理器
  57 + if (isPropertyTopic(topic)) {
  58 + propertyHandler.handle(topic, payload);
  59 + } else if (isEventTopic(topic)) {
  60 + eventHandler.handle(topic, payload);
  61 + } else if (isServiceTopic(topic)) {
  62 + serviceHandler.handle(topic, payload);
  63 + } else {
  64 + log.warn("[route][未知的消息类型][topic: {}]", topic);
  65 + }
  66 + } catch (Exception e) {
  67 + log.error("[route][处理 MQTT 消息失败][topic: {}][payload: {}]", topic, payload, e);
  68 + }
  69 + }
  70 +
  71 + /**
  72 + * 判断是否为属性相关主题
  73 + *
  74 + * @param topic 主题
  75 + * @return 是否为属性主题
  76 + */
  77 + private boolean isPropertyTopic(String topic) {
  78 + return topic.endsWith(IotDeviceTopicEnum.PROPERTY_POST_TOPIC.getTopic()) ||
  79 + topic.contains(IotDeviceTopicEnum.PROPERTY_SET_TOPIC.getTopic()) ||
  80 + topic.contains(IotDeviceTopicEnum.PROPERTY_GET_TOPIC.getTopic());
  81 + }
  82 +
  83 + /**
  84 + * 判断是否为事件相关主题
  85 + *
  86 + * @param topic 主题
  87 + * @return 是否为事件主题
  88 + */
  89 + private boolean isEventTopic(String topic) {
  90 + return topic.contains(IotDeviceTopicEnum.EVENT_POST_TOPIC_PREFIX.getTopic()) &&
  91 + topic.endsWith(IotDeviceTopicEnum.EVENT_POST_TOPIC_SUFFIX.getTopic());
  92 + }
  93 +
  94 + /**
  95 + * 判断是否为服务相关主题
  96 + *
  97 + * @param topic 主题
  98 + * @return 是否为服务主题
  99 + */
  100 + private boolean isServiceTopic(String topic) {
  101 + return topic.contains(IotDeviceTopicEnum.SERVICE_TOPIC_PREFIX.getTopic()) &&
  102 + !isPropertyTopic(topic); // 排除属性相关的服务调用
  103 + }
  104 +
  105 +}
0 \ No newline at end of file 106 \ No newline at end of file
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/package-info.java 0 → 100644
  View file @f58cf28
  1 +/**
  2 + * MQTT 协议路由器包
  3 + * <p>
  4 + * 包含 MQTT 协议的所有路由处理器和抽象基类：
  5 + * <ul>
  6 + * <li>{@link cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router.IotMqttAbstractHandler}
  7 + * - 抽象路由处理器基类</li>
  8 + * <li>{@link cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router.IotMqttUpstreamRouter}
  9 + * - 上行消息路由器</li>
  10 + * <li>{@link cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router.IotMqttAuthRouter}
  11 + * - 认证路由器</li>
  12 + * <li>{@link cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router.IotMqttPropertyHandler}
  13 + * - 属性处理器</li>
  14 + * <li>{@link cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router.IotMqttEventHandler}
  15 + * - 事件处理器</li>
  16 + * <li>{@link cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router.IotMqttServiceHandler}
  17 + * - 服务处理器</li>
  18 + * </ul>
  19 + *
  20 + * @author 芋道源码
  21 + */
  22 +package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router;
0 \ No newline at end of file 23 \ No newline at end of file
yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application.yaml
  View file @f58cf28
@@ -37,7 +37,7 @@ yudao: @@ -37,7 +37,7 @@ yudao:
37 # 针对引入的 EMQX 组件的配置 37 # 针对引入的 EMQX 组件的配置
38 # ==================================== 38 # ====================================
39 emqx: 39 emqx:
40 - enabled: false 40 + enabled: true
41 mqtt-ssl: false 41 mqtt-ssl: false
42 mqtt-topics: 42 mqtt-topics:
43 - "/sys/#" # 系统主题（设备上报） 43 - "/sys/#" # 系统主题（设备上报）