From 69e25eeaaccfa6aa81b252ea4de7c7159fe5aec0 Mon Sep 17 00:00:00 2001 From: haohao <1036606149@qq.com> Date: Sat, 14 Jun 2025 20:44:55 +0800 Subject: [PATCH] reactor：【IoT 物联网】重构 EMQX 协议处理逻辑，新增认证和下行消息处理器 --- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayConfiguration.java | 20 +++++++++++++------- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/IotEmqxAuthEventProtocol.java | 113 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/IotEmqxDownstreamSubscriber.java | 68 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/IotEmqxUpstreamProtocol.java | 328 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/router/IotEmqxAuthEventHandler.java | 252 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/router/IotEmqxDownstreamHandler.java | 96 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/router/IotEmqxUpstreamHandler.java | 60 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttDownstreamSubscriber.java | 66 ------------------------------------------------------------------ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttUpstreamProtocol.java | 430 ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttDownstreamHandler.java | 96 ------------------------------------------------------------------------------------------------ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttHttpAuthHandler.java | 250 ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttUpstreamHandler.java | 60 ------------------------------------------------------------ 12 files changed, 930 insertions(+), 909 deletions(-) create mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/IotEmqxAuthEventProtocol.java create mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/IotEmqxDownstreamSubscriber.java create mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/IotEmqxUpstreamProtocol.java create mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/router/IotEmqxAuthEventHandler.java create mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/router/IotEmqxDownstreamHandler.java create mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/router/IotEmqxUpstreamHandler.java delete mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttDownstreamSubscriber.java delete mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttUpstreamProtocol.java delete mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttDownstreamHandler.java delete mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttHttpAuthHandler.java delete mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttUpstreamHandler.java diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayConfiguration.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayConfiguration.java index d730e92..e9b0001 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayConfiguration.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayConfiguration.java @@ -1,10 +1,11 @@ package cn.iocoder.yudao.module.iot.gateway.config; import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus; +import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.IotEmqxAuthEventProtocol; +import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.IotEmqxDownstreamSubscriber; +import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.IotEmqxUpstreamProtocol; import cn.iocoder.yudao.module.iot.gateway.protocol.http.IotHttpDownstreamSubscriber; import cn.iocoder.yudao.module.iot.gateway.protocol.http.IotHttpUpstreamProtocol; -import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttDownstreamSubscriber; -import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttUpstreamProtocol; import lombok.extern.slf4j.Slf4j; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; @@ -37,7 +38,7 @@ public class IotGatewayConfiguration { } /** - * IoT 网关 MQTT 协议配置类 + * IoT 网关 EMQX 协议配置类 */ @Configuration @ConditionalOnProperty(prefix = "yudao.iot.gateway.protocol.emqx", name = "enabled", havingValue = "true") @@ -45,14 +46,19 @@ public class IotGatewayConfiguration { public static class MqttProtocolConfiguration { @Bean - public IotMqttUpstreamProtocol iotMqttUpstreamProtocol(IotGatewayProperties gatewayProperties) { - return new IotMqttUpstreamProtocol(gatewayProperties.getProtocol().getEmqx()); + public IotEmqxAuthEventProtocol iotEmqxAuthEventProtocol(IotGatewayProperties gatewayProperties) { + return new IotEmqxAuthEventProtocol(gatewayProperties.getProtocol().getEmqx()); } @Bean - public IotMqttDownstreamSubscriber iotMqttDownstreamSubscriber(IotMqttUpstreamProtocol mqttUpstreamProtocol, + public IotEmqxUpstreamProtocol iotEmqxUpstreamProtocol(IotGatewayProperties gatewayProperties) { + return new IotEmqxUpstreamProtocol(gatewayProperties.getProtocol().getEmqx()); + } + + @Bean + public IotEmqxDownstreamSubscriber iotEmqxDownstreamSubscriber(IotEmqxUpstreamProtocol mqttUpstreamProtocol, IotMessageBus messageBus) { - return new IotMqttDownstreamSubscriber(mqttUpstreamProtocol, messageBus); + return new IotEmqxDownstreamSubscriber(mqttUpstreamProtocol, messageBus); } } diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/IotEmqxAuthEventProtocol.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/IotEmqxAuthEventProtocol.java new file mode 100644 index 0000000..2ba902c --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/IotEmqxAuthEventProtocol.java @@ -0,0 +1,113 @@ +package cn.iocoder.yudao.module.iot.gateway.protocol.emqx; + +import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils; +import cn.iocoder.yudao.module.iot.gateway.config.IotGatewayProperties; +import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.router.IotEmqxAuthEventHandler; +import cn.iocoder.yudao.module.iot.gateway.util.IotMqttTopicUtils; +import io.vertx.core.Vertx; +import io.vertx.core.http.HttpServer; +import io.vertx.ext.web.Router; +import io.vertx.ext.web.handler.BodyHandler; +import jakarta.annotation.PostConstruct; +import jakarta.annotation.PreDestroy; +import lombok.extern.slf4j.Slf4j; + +/** + * IoT 网关 EMQX 认证事件协议服务 + *

+ * 为 EMQX 提供 HTTP 接口服务，包括： + * 1. 设备认证接口 - 对应 EMQX HTTP 认证插件 + * 2. 设备事件处理接口 - 对应 EMQX Webhook 事件通知 + * + * @author 芋道源码 + */ +@Slf4j +public class IotEmqxAuthEventProtocol { + + private final IotGatewayProperties.EmqxProperties emqxProperties; + + private final String serverId; + + private Vertx vertx; + + private HttpServer httpServer; + + public IotEmqxAuthEventProtocol(IotGatewayProperties.EmqxProperties emqxProperties) { + this.emqxProperties = emqxProperties; + this.serverId = IotDeviceMessageUtils.generateServerId(emqxProperties.getMqttPort()); + } + + @PostConstruct + public void start() { + try { + // 创建 Vertx 实例 + this.vertx = Vertx.vertx(); + startHttpServer(); + log.info("[start][IoT 网关 EMQX 认证事件协议服务启动成功, 端口: {}]", emqxProperties.getHttpPort()); + } catch (Exception e) { + log.error("[start][IoT 网关 EMQX 认证事件协议服务启动失败]", e); + throw e; + } + } + + @PreDestroy + public void stop() { + stopHttpServer(); + + // 关闭 Vertx 实例 + if (vertx != null) { + try { + vertx.close(); + log.debug("[stop][Vertx 实例已关闭]"); + } catch (Exception e) { + log.warn("[stop][关闭 Vertx 实例失败]", e); + } + } + + log.info("[stop][IoT 网关 EMQX 认证事件协议服务已停止]"); + } + + /** + * 启动 HTTP 服务器 + */ + private void startHttpServer() { + int port = emqxProperties.getHttpPort(); + + // 1. 创建路由 + Router router = Router.router(vertx); + router.route().handler(BodyHandler.create()); + + // 2. 创建处理器，传入 serverId + IotEmqxAuthEventHandler handler = new IotEmqxAuthEventHandler(serverId); + router.post(IotMqttTopicUtils.MQTT_AUTH_PATH).handler(handler::handleAuth); + router.post(IotMqttTopicUtils.MQTT_EVENT_PATH).handler(handler::handleEvent); + + // 3. 启动 HTTP 服务器 + try { + httpServer = vertx.createHttpServer() + .requestHandler(router) + .listen(port) + .result(); + } catch (Exception e) { + log.error("[startHttpServer][HTTP 服务器启动失败, 端口: {}]", port, e); + throw e; + } + } + + /** + * 停止 HTTP 服务器 + */ + private void stopHttpServer() { + if (httpServer == null) { + return; + } + + try { + httpServer.close().result(); + log.info("[stopHttpServer][HTTP 服务器已停止]"); + } catch (Exception e) { + log.error("[stopHttpServer][HTTP 服务器停止失败]", e); + } + } + +} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/IotEmqxDownstreamSubscriber.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/IotEmqxDownstreamSubscriber.java new file mode 100644 index 0000000..61bf123 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/IotEmqxDownstreamSubscriber.java @@ -0,0 +1,68 @@ +package cn.iocoder.yudao.module.iot.gateway.protocol.emqx; + +import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus; +import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber; +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; +import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils; +import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.router.IotEmqxDownstreamHandler; +import jakarta.annotation.PostConstruct; +import lombok.extern.slf4j.Slf4j; + +/** + * IoT 网关 EMQX 订阅者：接收下行给设备的消息 + * + * @author 芋道源码 + */ +@Slf4j +public class IotEmqxDownstreamSubscriber implements IotMessageSubscriber { + + private final IotEmqxDownstreamHandler downstreamHandler; + + private final IotMessageBus messageBus; + + private final IotEmqxUpstreamProtocol protocol; + + public IotEmqxDownstreamSubscriber(IotEmqxUpstreamProtocol protocol, IotMessageBus messageBus) { + this.protocol = protocol; + this.messageBus = messageBus; + this.downstreamHandler = new IotEmqxDownstreamHandler(protocol); + } + + @PostConstruct + public void init() { + messageBus.register(this); + } + + @Override + public String getTopic() { + return IotDeviceMessageUtils.buildMessageBusGatewayDeviceMessageTopic(protocol.getServerId()); + } + + @Override + public String getGroup() { + // 保证点对点消费，需要保证独立的 Group，所以使用 Topic 作为 Group + return getTopic(); + } + + @Override + public void onMessage(IotDeviceMessage message) { + log.debug("[onMessage][接收到下行消息, messageId: {}, method: {}, deviceId: {}]", + message.getId(), message.getMethod(), message.getDeviceId()); + try { + // 1. 校验 + String method = message.getMethod(); + if (method == null) { + log.warn("[onMessage][消息方法为空, messageId: {}, deviceId: {}]", + message.getId(), message.getDeviceId()); + return; + } + + // 2. 处理下行消息 + downstreamHandler.handle(message); + } catch (Exception e) { + log.error("[onMessage][处理下行消息失败, messageId: {}, method: {}, deviceId: {}]", + message.getId(), message.getMethod(), message.getDeviceId(), e); + } + } + +} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/IotEmqxUpstreamProtocol.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/IotEmqxUpstreamProtocol.java new file mode 100644 index 0000000..17431da --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/IotEmqxUpstreamProtocol.java @@ -0,0 +1,328 @@ +package cn.iocoder.yudao.module.iot.gateway.protocol.emqx; + +import cn.hutool.core.collection.CollUtil; +import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils; +import cn.iocoder.yudao.module.iot.gateway.config.IotGatewayProperties; +import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.router.IotEmqxUpstreamHandler; +import io.netty.handler.codec.mqtt.MqttQoS; +import io.vertx.core.Vertx; +import io.vertx.core.buffer.Buffer; +import io.vertx.mqtt.MqttClient; +import io.vertx.mqtt.MqttClientOptions; +import jakarta.annotation.PostConstruct; +import jakarta.annotation.PreDestroy; +import lombok.Getter; +import lombok.extern.slf4j.Slf4j; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * IoT 网关 EMQX 协议：接收设备上行消息 + * + * @author 芋道源码 + */ +@Slf4j +public class IotEmqxUpstreamProtocol { + + private final IotGatewayProperties.EmqxProperties emqxProperties; + + private volatile boolean isRunning = false; + + private Vertx vertx; + + @Getter + private final String serverId; + + private MqttClient mqttClient; + + private IotEmqxUpstreamHandler upstreamHandler; + + public IotEmqxUpstreamProtocol(IotGatewayProperties.EmqxProperties emqxProperties) { + this.emqxProperties = emqxProperties; + this.serverId = IotDeviceMessageUtils.generateServerId(emqxProperties.getMqttPort()); + } + + @PostConstruct + public void start() { + if (isRunning) { + return; + } + + try { + // 1. 初始化 Vertx 实例 + this.vertx = Vertx.vertx(); + + // 2. 启动 MQTT 客户端 + startMqttClient(); + + // 3. 标记服务为运行状态 + isRunning = true; + log.info("[start][IoT 网关 EMQX 协议启动成功]"); + } catch (Exception e) { + log.error("[start][IoT 网关 EMQX 协议服务启动失败，应用将关闭]", e); + stop(); + + // 异步关闭应用，避免阻塞当前线程 + new Thread(() -> { + try { + Thread.sleep(1000); // 等待1秒让日志输出完成 + log.error("[start][由于 MQTT 连接失败，正在关闭应用]"); + } catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + } finally { + System.exit(1); // 直接关闭 JVM + } + }).start(); + + throw e; + } + } + + @PreDestroy + public void stop() { + if (!isRunning) { + return; + } + + // 1. 停止 MQTT 客户端 + stopMqttClient(); + + // 2. 关闭 Vertx 实例 + if (vertx != null) { + try { + vertx.close(); + } catch (Exception e) { + log.warn("[stop][关闭 Vertx 实例失败]", e); + } + } + + // 3. 标记服务为停止状态 + isRunning = false; + log.info("[stop][IoT 网关 MQTT 协议服务已停止]"); + } + + /** + * 启动 MQTT 客户端 + */ + private void startMqttClient() { + try { + // 2.1. 初始化消息处理器 + this.upstreamHandler = new IotEmqxUpstreamHandler(this); + + // 2.2. 创建 MQTT 客户端 + createMqttClient(); + + // 2.3. 连接 MQTT Broker（同步等待首次连接结果） + boolean connected = connectMqttSync(); + if (!connected) { + throw new RuntimeException("首次连接 MQTT Broker 失败"); + } + } catch (Exception e) { + log.error("[startMqttClient][MQTT 客户端启动失败]", e); + throw new RuntimeException("MQTT 客户端启动失败", e); + } + } + + /** + * 连接 MQTT Broker + * + * @param isReconnect 是否为重连 + * @param isSync 是否同步等待连接结果 + * @return 当 isSync 为 true 时返回连接是否成功，否则返回 null + */ + private Boolean connectMqtt(boolean isReconnect, boolean isSync) { + String host = emqxProperties.getMqttHost(); + Integer port = emqxProperties.getMqttPort(); + + // 2.3.1. 如果是重连，则需要重新创建 MQTT 客户端 + if (isReconnect) { + createMqttClient(); + } + + // 2.3.2. 连接 MQTT Broker + CountDownLatch latch = isSync ? new CountDownLatch(1) : null; + AtomicBoolean success = isSync + ? new AtomicBoolean(false) + : null; + + mqttClient.connect(port, host, connectResult -> { + if (connectResult.succeeded()) { + if (isReconnect) { + log.info("[connectMqtt][MQTT 客户端重连成功]"); + } else { + log.info("[connectMqtt][MQTT 客户端连接成功, host: {}, port: {}]", host, port); + } + // 设置处理器和订阅主题 + setupMqttHandlers(); + subscribeToTopics(); + if (success != null) { + success.set(true); + } + } else { + log.error("[connectMqtt][连接 MQTT Broker 失败, host: {}, port: {}]", host, port, connectResult.cause()); + if (isReconnect) { + reconnectWithDelay(); + } else { + log.error("[connectMqtt][首次连接失败，连接终止]"); + } + } + + if (latch != null) { + latch.countDown(); + } + }); + + // 2.3.3. 如果需要同步等待连接结果，则等待 + if (isSync) { + try { + latch.await(10, java.util.concurrent.TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + log.error("[connectMqtt][等待连接结果被中断]", e); + } + return success.get(); + } + + return null; + } + + /** + * 同步连接 MQTT Broker + * + * @return 是否连接成功 + */ + private boolean connectMqttSync() { + Boolean result = connectMqtt(false, true); + return result != null ? result : false; + } + + /** + * 停止 MQTT 客户端 + */ + private void stopMqttClient() { + // 1.1. 取消订阅所有主题 + if (mqttClient != null && mqttClient.isConnected()) { + List topicList = emqxProperties.getMqttTopics(); + if (CollUtil.isNotEmpty(topicList)) { + for (String topic : topicList) { + try { + mqttClient.unsubscribe(topic); + } catch (Exception e) { + log.warn("[stopMqttClient][取消订阅主题({})异常]", topic, e); + } + } + } + } + + // 1.2. 断开 MQTT 客户端连接 + if (mqttClient != null && mqttClient.isConnected()) { + try { + mqttClient.disconnect(); + } catch (Exception e) { + log.warn("[stopMqttClient][关闭 MQTT 客户端异常]", e); + } + } + } + + /** + * 创建 MQTT 客户端 + */ + private void createMqttClient() { + MqttClientOptions options = new MqttClientOptions() + .setClientId(emqxProperties.getMqttClientId()) + .setUsername(emqxProperties.getMqttUsername()) + .setPassword(emqxProperties.getMqttPassword()) + .setSsl(emqxProperties.getMqttSsl()); + this.mqttClient = MqttClient.create(vertx, options); + } + + /** + * 设置 MQTT 处理器 + */ + private void setupMqttHandlers() { + // 1. 设置断开重连监听器 + mqttClient.closeHandler(closeEvent -> { + if (isRunning) { + log.warn("[closeHandler][MQTT 连接已断开, 准备重连]"); + reconnectWithDelay(); + } + }); + + // 2. 设置异常处理器 + mqttClient.exceptionHandler(exception -> log.error("[exceptionHandler][MQTT 客户端异常]", exception)); + + // 3. 设置消息处理器 + mqttClient.publishHandler(upstreamHandler::handle); + } + + /** + * 订阅设备上行消息主题 + */ + private void subscribeToTopics() { + // 1. 校验 MQTT 客户端是否连接 + List topicList = emqxProperties.getMqttTopics(); + if (mqttClient == null || !mqttClient.isConnected()) { + log.warn("[subscribeToTopics][MQTT 客户端未连接, 跳过订阅]"); + return; + } + + int qos = emqxProperties.getMqttQos(); + + // 2. 构建主题-QoS 映射，批量订阅 + Map topicQosMap = new HashMap<>(); + for (String topic : topicList) { + topicQosMap.put(topic, qos); + } + + // 3. 批量订阅所有主题 + mqttClient.subscribe(topicQosMap, subscribeResult -> { + if (subscribeResult.succeeded()) { + log.info("[subscribeToTopics][订阅主题成功, 共 {} 个主题]", topicList.size()); + } else { + log.error("[subscribeToTopics][订阅主题失败, 共 {} 个主题, 原因: {}]", + topicList.size(), subscribeResult.cause().getMessage(), subscribeResult.cause()); + } + }); + } + + /** + * 延迟重连 + */ + private void reconnectWithDelay() { + long delay = emqxProperties.getReconnectDelayMs(); + vertx.setTimer(delay, timerId -> { + if (!isRunning) { + return; + } + if (mqttClient != null && mqttClient.isConnected()) { + return; + } + log.info("[reconnectWithDelay][开始重连 MQTT Broker]"); + try { + connectMqtt(true, false); + } catch (Exception e) { + log.error("[reconnectWithDelay][重连失败, 将继续尝试]", e); + } + }); + } + + /** + * 发布消息到 MQTT Broker + * + * @param topic 主题 + * @param payload 消息内容 + */ + public void publishMessage(String topic, String payload) { + if (mqttClient == null || !mqttClient.isConnected()) { + log.warn("[publishMessage][MQTT 客户端未连接, 无法发布消息]"); + return; + } + MqttQoS qos = MqttQoS.valueOf(emqxProperties.getMqttQos()); + mqttClient.publish(topic, Buffer.buffer(payload), qos, false, false); + } + +} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/router/IotEmqxAuthEventHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/router/IotEmqxAuthEventHandler.java new file mode 100644 index 0000000..df22f98 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/router/IotEmqxAuthEventHandler.java @@ -0,0 +1,252 @@ +package cn.iocoder.yudao.module.iot.gateway.protocol.emqx.router; + +import cn.hutool.core.util.BooleanUtil; +import cn.hutool.core.util.StrUtil; +import cn.hutool.extra.spring.SpringUtil; +import cn.iocoder.yudao.framework.common.pojo.CommonResult; +import cn.iocoder.yudao.module.iot.core.biz.IotDeviceCommonApi; +import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceAuthReqDTO; +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; +import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils; +import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService; +import io.vertx.core.json.JsonObject; +import io.vertx.ext.web.RoutingContext; +import lombok.extern.slf4j.Slf4j; + +/** + * IoT 网关 EMQX 认证事件处理器 + *

+ * 为 EMQX 提供 HTTP 接口服务，包括： + * 1. 设备认证接口 - 对应 EMQX HTTP 认证插件 + * 2. 设备事件处理接口 - 对应 EMQX Webhook 事件通知 + * 提供统一的错误处理和参数校验 + * + * @author 芋道源码 + */ +@Slf4j +public class IotEmqxAuthEventHandler { + + /** + * HTTP 成功状态码（EMQX 要求固定使用 200） + */ + private static final int SUCCESS_STATUS_CODE = 200; + + /** + * 认证允许结果 + */ + private static final String RESULT_ALLOW = "allow"; + /** + * 认证拒绝结果 + */ + private static final String RESULT_DENY = "deny"; + /** + * 认证忽略结果 + */ + private static final String RESULT_IGNORE = "ignore"; + + /** + * EMQX 事件类型常量 + */ + private static final String EVENT_CLIENT_CONNECTED = "client.connected"; + private static final String EVENT_CLIENT_DISCONNECTED = "client.disconnected"; + + private final String serverId; + + private final IotDeviceMessageService deviceMessageService; + + private final IotDeviceCommonApi deviceApi; + + public IotEmqxAuthEventHandler(String serverId) { + this.serverId = serverId; + this.deviceMessageService = SpringUtil.getBean(IotDeviceMessageService.class); + this.deviceApi = SpringUtil.getBean(IotDeviceCommonApi.class); + } + + /** + * EMQX 认证接口 + */ + public void handleAuth(RoutingContext context) { + try { + // 1. 参数校验 + JsonObject body = parseRequestBody(context); + if (body == null) { + return; + } + String clientId = body.getString("clientid"); + String username = body.getString("username"); + String password = body.getString("password"); + log.debug("[handleAuth][设备认证请求: clientId={}, username={}]", clientId, username); + if (StrUtil.hasEmpty(clientId, username, password)) { + log.info("[handleAuth][认证参数不完整: clientId={}, username={}]", clientId, username); + sendAuthResponse(context, RESULT_DENY); + return; + } + + // 2. 执行认证 + boolean authResult = performDeviceAuth(clientId, username, password); + log.info("[handleAuth][设备认证结果: {} -> {}]", username, authResult); + if (authResult) { + sendAuthResponse(context, RESULT_ALLOW); + } else { + sendAuthResponse(context, RESULT_DENY); + } + } catch (Exception e) { + log.error("[handleAuth][设备认证异常]", e); + sendAuthResponse(context, RESULT_IGNORE); + } + } + + /** + * EMQX 统一事件处理接口 + * 根据 EMQX 官方 Webhook 设计，统一处理所有客户端事件 + * 支持的事件类型：client.connected、client.disconnected 等 + */ + public void handleEvent(RoutingContext context) { + JsonObject body = null; + try { + // 1. 解析请求体 + body = parseRequestBody(context); + if (body == null) { + return; + } + String event = body.getString("event"); + String username = body.getString("username"); + log.debug("[handleEvent][收到事件: {} - {}]", event, username); + + // 2. 根据事件类型进行分发处理 + switch (event) { + case EVENT_CLIENT_CONNECTED: + handleClientConnected(body); + break; + case EVENT_CLIENT_DISCONNECTED: + handleClientDisconnected(body); + break; + default: + break; + } + + // EMQX Webhook 只需要 200 状态码，无需响应体 + context.response().setStatusCode(SUCCESS_STATUS_CODE).end(); + } catch (Exception e) { + log.error("[handleEvent][事件处理失败][body={}]", body != null ? body.encode() : "null", e); + // 即使处理失败，也返回 200 避免EMQX重试 + context.response().setStatusCode(SUCCESS_STATUS_CODE).end(); + } + } + + /** + * 处理客户端连接事件 + */ + private void handleClientConnected(JsonObject body) { + String username = body.getString("username"); + log.info("[handleClientConnected][设备上线: {}]", username); + handleDeviceStateChange(username, true); + } + + /** + * 处理客户端断开连接事件 + */ + private void handleClientDisconnected(JsonObject body) { + String username = body.getString("username"); + String reason = body.getString("reason"); + log.info("[handleClientDisconnected][设备下线: {} ({})]", username, reason); + handleDeviceStateChange(username, false); + } + + /** + * 解析请求体 + * + * @param context 路由上下文 + * @return 请求体JSON对象，解析失败时返回null + */ + private JsonObject parseRequestBody(RoutingContext context) { + String rawBody = null; + try { + rawBody = context.body().asString(); + JsonObject body = context.body().asJsonObject(); + if (body == null) { + log.info("[parseRequestBody][请求体为空][rawBody={}]", rawBody); + sendAuthResponse(context, RESULT_IGNORE); + return null; + } + return body; + } catch (Exception e) { + log.error("[parseRequestBody][解析请求体失败][rawBody={}]", rawBody, e); + sendAuthResponse(context, RESULT_IGNORE); + return null; + } + } + + /** + * 执行设备认证 + * + * @param clientId 客户端ID + * @param username 用户名 + * @param password 密码 + * @return 认证是否成功 + */ + private boolean performDeviceAuth(String clientId, String username, String password) { + try { + CommonResult result = deviceApi.authDevice(new IotDeviceAuthReqDTO() + .setClientId(clientId).setUsername(username).setPassword(password)); + result.checkError(); + return BooleanUtil.isTrue(result.getData()); + } catch (Exception e) { + log.error("[performDeviceAuth][认证接口调用失败: {}]", username, e); + throw e; + } + } + + /** + * 处理设备状态变化 + * + * @param username 用户名 + * @param online 是否在线 true 在线 false 离线 + */ + private void handleDeviceStateChange(String username, boolean online) { + // 1. 解析设备信息 + IotDeviceAuthUtils.DeviceInfo deviceInfo = IotDeviceAuthUtils.parseUsername(username); + if (deviceInfo == null) { + log.debug("[handleDeviceStateChange][跳过非设备连接: {}]", username); + return; + } + + try { + // 2. 构建设备状态消息 + IotDeviceMessage message = online ? IotDeviceMessage.buildStateOnline() + : IotDeviceMessage.buildStateOffline(); + + // 3. 发送设备状态消息 + deviceMessageService.sendDeviceMessage(message, + deviceInfo.getProductKey(), deviceInfo.getDeviceName(), serverId); + } catch (Exception e) { + log.error("[handleDeviceStateChange][发送设备状态消息失败: {}]", username, e); + } + } + + /** + * 发送 EMQX 认证响应 + * 根据 EMQX 官方文档要求，必须返回 JSON 格式响应 + * + * @param context 路由上下文 + * @param result 认证结果：allow、deny、ignore + */ + private void sendAuthResponse(RoutingContext context, String result) { + // 构建符合 EMQX 官方规范的响应 + JsonObject response = new JsonObject() + .put("result", result) + .put("is_superuser", false); + + // 可以根据业务需求添加客户端属性 + // response.put("client_attrs", new JsonObject().put("role", "device")); + + // 可以添加认证过期时间（可选） + // response.put("expire_at", System.currentTimeMillis() / 1000 + 3600); + + context.response() + .setStatusCode(SUCCESS_STATUS_CODE) + .putHeader("Content-Type", "application/json; charset=utf-8") + .end(response.encode()); + } + +} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/router/IotEmqxDownstreamHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/router/IotEmqxDownstreamHandler.java new file mode 100644 index 0000000..43ec9a2 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/router/IotEmqxDownstreamHandler.java @@ -0,0 +1,96 @@ +package cn.iocoder.yudao.module.iot.gateway.protocol.emqx.router; + +import cn.hutool.core.util.StrUtil; +import cn.hutool.extra.spring.SpringUtil; +import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceRespDTO; +import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum; +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; +import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils; +import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.IotEmqxUpstreamProtocol; +import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceService; +import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService; +import cn.iocoder.yudao.module.iot.gateway.util.IotMqttTopicUtils; +import lombok.extern.slf4j.Slf4j; + +/** + * IoT 网关 EMQX 下行消息处理器 + *

+ * 从消息总线接收到下行消息，然后发布到 MQTT Broker，从而被设备所接收 + * + * @author 芋道源码 + */ +@Slf4j +public class IotEmqxDownstreamHandler { + + private final IotEmqxUpstreamProtocol protocol; + + private final IotDeviceService deviceService; + + private final IotDeviceMessageService deviceMessageService; + + public IotEmqxDownstreamHandler(IotEmqxUpstreamProtocol protocol) { + this.protocol = protocol; + this.deviceService = SpringUtil.getBean(IotDeviceService.class); + this.deviceMessageService = SpringUtil.getBean(IotDeviceMessageService.class); + } + + /** + * 处理下行消息 + * + * @param message 设备消息 + */ + public void handle(IotDeviceMessage message) { + // 1. 获取设备信息（使用缓存） + IotDeviceRespDTO deviceInfo = deviceService.getDeviceFromCache(message.getDeviceId()); + if (deviceInfo == null) { + log.error("[handle][设备信息({})不存在]", message.getDeviceId()); + return; + } + + // 2.1 根据方法构建主题 + String topic = buildTopicByMethod(message, deviceInfo.getProductKey(), deviceInfo.getDeviceName()); + if (StrUtil.isBlank(topic)) { + log.warn("[handle][未知的消息方法: {}]", message.getMethod()); + return; + } + + // 2.2 构建载荷 + byte[] payload = deviceMessageService.encodeDeviceMessage(message, deviceInfo.getProductKey(), deviceInfo.getDeviceName()); + + // 2.3 发布消息 + protocol.publishMessage(topic, new String(payload)); + } + + /** + * 根据消息方法和回复状态构建主题 + * + * @param message 设备消息 + * @param productKey 产品标识 + * @param deviceName 设备名称 + * @return 构建的主题，如果方法不支持返回 null + */ + private String buildTopicByMethod(IotDeviceMessage message, String productKey, String deviceName) { + // 1. 解析消息方法 + IotDeviceMessageMethodEnum methodEnum = IotDeviceMessageMethodEnum.of(message.getMethod()); + if (methodEnum == null) { + log.warn("[buildTopicByMethod][未知的消息方法: {}]", message.getMethod()); + return null; + } + + // 2. 判断是否回复消息 + boolean isReply = IotDeviceMessageUtils.isReplyMessage(message); + + // 3. 根据消息方法和回复状态，构建主题 + if (methodEnum == IotDeviceMessageMethodEnum.PROPERTY_POST && isReply) { + return IotMqttTopicUtils.buildPropertyPostReplyTopic(productKey, deviceName); + } + if (methodEnum == IotDeviceMessageMethodEnum.PROPERTY_SET && !isReply) { + return IotMqttTopicUtils.buildPropertySetTopic(productKey, deviceName); + } + + log.warn("[buildTopicByMethod][暂时不支持的下行消息: method={}, isReply={}]", + message.getMethod(), isReply); + return null; + } + +} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/router/IotEmqxUpstreamHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/router/IotEmqxUpstreamHandler.java new file mode 100644 index 0000000..81d8cbb --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/router/IotEmqxUpstreamHandler.java @@ -0,0 +1,60 @@ +package cn.iocoder.yudao.module.iot.gateway.protocol.emqx.router; + +import cn.hutool.core.util.StrUtil; +import cn.hutool.extra.spring.SpringUtil; +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; +import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.IotEmqxUpstreamProtocol; +import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService; +import io.vertx.mqtt.messages.MqttPublishMessage; +import lombok.extern.slf4j.Slf4j; + +/** + * IoT 网关 EMQX 上行消息处理器 + * + * @author 芋道源码 + */ +@Slf4j +public class IotEmqxUpstreamHandler { + + private final IotDeviceMessageService deviceMessageService; + + private final String serverId; + + public IotEmqxUpstreamHandler(IotEmqxUpstreamProtocol protocol) { + this.deviceMessageService = SpringUtil.getBean(IotDeviceMessageService.class); + this.serverId = protocol.getServerId(); + } + + /** + * 处理 MQTT 发布消息 + */ + public void handle(MqttPublishMessage mqttMessage) { + log.info("[handle][收到 MQTT 消息, topic: {}, payload: {}]", mqttMessage.topicName(), mqttMessage.payload()); + String topic = mqttMessage.topicName(); + byte[] payload = mqttMessage.payload().getBytes(); + try { + // 1. 解析主题，一次性获取所有信息 + String[] topicParts = topic.split("/"); + if (topicParts.length < 4 || StrUtil.hasBlank(topicParts[2], topicParts[3])) { + log.warn("[handle][topic({}) 格式不正确，无法解析有效的 productKey 和 deviceName]", topic); + return; + } + + String productKey = topicParts[2]; + String deviceName = topicParts[3]; + + // 3. 解码消息 + IotDeviceMessage message = deviceMessageService.decodeDeviceMessage(payload, productKey, deviceName); + if (message == null) { + log.warn("[handle][topic({}) payload({}) 消息解码失败]", topic, new String(payload)); + return; + } + + // 4. 发送消息到队列 + deviceMessageService.sendDeviceMessage(message, productKey, deviceName, serverId); + } catch (Exception e) { + log.error("[handle][topic({}) payload({}) 处理异常]", topic, new String(payload), e); + } + } + +} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttDownstreamSubscriber.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttDownstreamSubscriber.java deleted file mode 100644 index 861c3a5..0000000 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttDownstreamSubscriber.java +++ /dev/null @@ -1,66 +0,0 @@ -package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt; - -import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus; -import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber; -import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; -import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils; -import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router.IotMqttDownstreamHandler; -import jakarta.annotation.PostConstruct; -import lombok.extern.slf4j.Slf4j; - -/** - * IoT 网关 MQTT 订阅者：接收下行给设备的消息 - * - * @author 芋道源码 - */ -@Slf4j -public class IotMqttDownstreamSubscriber implements IotMessageSubscriber { - - private final IotMqttDownstreamHandler downstreamHandler; - private final IotMessageBus messageBus; - private final IotMqttUpstreamProtocol protocol; - - public IotMqttDownstreamSubscriber(IotMqttUpstreamProtocol protocol, IotMessageBus messageBus) { - this.protocol = protocol; - this.messageBus = messageBus; - this.downstreamHandler = new IotMqttDownstreamHandler(protocol); - } - - @PostConstruct - public void init() { - messageBus.register(this); - } - - @Override - public String getTopic() { - return IotDeviceMessageUtils.buildMessageBusGatewayDeviceMessageTopic(protocol.getServerId()); - } - - @Override - public String getGroup() { - // 保证点对点消费，需要保证独立的 Group，所以使用 Topic 作为 Group - return getTopic(); - } - - @Override - public void onMessage(IotDeviceMessage message) { - log.debug("[onMessage][接收到下行消息, messageId: {}, method: {}, deviceId: {}]", - message.getId(), message.getMethod(), message.getDeviceId()); - try { - // 1. 校验 - String method = message.getMethod(); - if (method == null) { - log.warn("[onMessage][消息方法为空, messageId: {}, deviceId: {}]", - message.getId(), message.getDeviceId()); - return; - } - - // 2. 处理下行消息 - downstreamHandler.handle(message); - } catch (Exception e) { - log.error("[onMessage][处理下行消息失败, messageId: {}, method: {}, deviceId: {}]", - message.getId(), message.getMethod(), message.getDeviceId(), e); - } - } - -} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttUpstreamProtocol.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttUpstreamProtocol.java deleted file mode 100644 index 3cdfa08..0000000 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttUpstreamProtocol.java +++ /dev/null @@ -1,430 +0,0 @@ -package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt; - -import cn.hutool.core.collection.CollUtil; -import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils; -import cn.iocoder.yudao.module.iot.gateway.config.IotGatewayProperties; -import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router.IotMqttHttpAuthHandler; -import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router.IotMqttUpstreamHandler; -import cn.iocoder.yudao.module.iot.gateway.util.IotMqttTopicUtils; -import io.netty.handler.codec.mqtt.MqttQoS; -import io.vertx.core.Vertx; -import io.vertx.core.buffer.Buffer; -import io.vertx.core.http.HttpServer; -import io.vertx.ext.web.Router; -import io.vertx.ext.web.handler.BodyHandler; -import io.vertx.mqtt.MqttClient; -import io.vertx.mqtt.MqttClientOptions; -import jakarta.annotation.PostConstruct; -import jakarta.annotation.PreDestroy; -import lombok.Getter; -import lombok.extern.slf4j.Slf4j; - -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.concurrent.atomic.AtomicInteger; - -/** - * IoT 网关 MQTT 协议：接收设备上行消息 - *

- * 1. MQTT 客户端：连接 EMQX，消费处理设备上行和下行消息 - * 2. HTTP 认证服务：为 EMQX 提供设备认证、连接、断开接口 - * - * @author 芋道源码 - */ -@Slf4j -public class IotMqttUpstreamProtocol { - - private final IotGatewayProperties.EmqxProperties emqxProperties; - - /** - * 服务运行状态标志 - */ - private volatile boolean isRunning = false; - - private Vertx vertx; - - @Getter - private final String serverId; - - // MQTT 客户端相关 - private MqttClient mqttClient; - private IotMqttUpstreamHandler upstreamHandler; - - // HTTP 认证服务相关 - private HttpServer httpAuthServer; - - public IotMqttUpstreamProtocol(IotGatewayProperties.EmqxProperties emqxProperties) { - this.emqxProperties = emqxProperties; - this.serverId = IotDeviceMessageUtils.generateServerId(emqxProperties.getMqttPort()); - } - - @PostConstruct - public void start() { - if (isRunning) { - log.warn("[start][MQTT 协议服务已经在运行中，请勿重复启动]"); - return; - } - log.info("[start][启动 MQTT 协议服务]"); - - try { - this.vertx = Vertx.vertx(); - - // 1. 启动 HTTP 认证服务 - startHttpAuthServer(); - - // 2. 启动 MQTT 客户端 - startMqttClient(); - - isRunning = true; - log.info("[start][MQTT 协议服务启动完成]"); - } catch (Exception e) { - log.error("[start][MQTT 协议服务启动失败]", e); - // 启动失败时清理资源 - stop(); - throw e; - } - } - - @PreDestroy - public void stop() { - if (!isRunning) { - log.warn("[stop][MQTT 协议服务已经停止，无需再次停止]"); - return; - } - log.info("[stop][停止 MQTT 协议服务]"); - - // 1. 停止 MQTT 客户端 - stopMqttClient(); - - // 2. 停止 HTTP 认证服务 - stopHttpAuthServer(); - - // 3. 关闭 Vertx 实例 - if (vertx != null) { - try { - vertx.close(); - log.debug("[stop][Vertx 实例已关闭]"); - } catch (Exception e) { - log.warn("[stop][关闭 Vertx 实例失败]", e); - } - } - - isRunning = false; - log.info("[stop][MQTT 协议服务已停止]"); - } - - /** - * 启动 HTTP 认证服务 - */ - private void startHttpAuthServer() { - log.info("[startHttpAuthServer][启动 HTTP 认证服务]"); - - // 1.1 创建路由 - Router router = Router.router(vertx); - router.route().handler(BodyHandler.create()); - // 1.2 创建认证处理器 - IotMqttHttpAuthHandler authHandler = new IotMqttHttpAuthHandler(this); - router.post(IotMqttTopicUtils.MQTT_AUTH_PATH).handler(authHandler::handleAuth); - router.post(IotMqttTopicUtils.MQTT_EVENT_PATH).handler(authHandler::handleEvent); - - // 2. 启动 HTTP 服务器 - int authPort = emqxProperties.getHttpPort(); - try { - httpAuthServer = vertx.createHttpServer() - .requestHandler(router) - .listen(authPort) - .result(); - log.info("[startHttpAuthServer][HTTP 认证服务启动成功, 端口: {}]", authPort); - } catch (Exception e) { - log.error("[startHttpAuthServer][HTTP 认证服务启动失败]", e); - throw e; - } - } - - /** - * 停止 HTTP 认证服务 - */ - private void stopHttpAuthServer() { - if (httpAuthServer == null) { - return; - } - try { - httpAuthServer.close().result(); - log.info("[stopHttpAuthServer][HTTP 认证服务已停止]"); - } catch (Exception e) { - log.error("[stopHttpAuthServer][HTTP 认证服务停止失败]", e); - } - } - - /** - * 启动 MQTT 客户端 - */ - private void startMqttClient() { - log.info("[startMqttClient][启动 MQTT 客户端]"); - - try { - // 1. 初始化消息处理器 - this.upstreamHandler = new IotMqttUpstreamHandler(this); - - // 2. 创建 MQTT 客户端 - log.info("[startMqttClient][使用 MQTT 客户端 ID: {}]", emqxProperties.getMqttClientId()); - createMqttClient(); - - // 3. 连接 MQTT Broker（同步等待首次连接结果） - boolean connected = connectMqttSync(); - if (!connected) { - throw new RuntimeException("首次连接 MQTT Broker 失败"); - } - - log.info("[startMqttClient][MQTT 客户端启动完成]"); - } catch (Exception e) { - log.error("[startMqttClient][MQTT 客户端启动失败]", e); - throw new RuntimeException("MQTT 客户端启动失败", e); - } - } - - /** - * 同步连接 MQTT Broker - * - * @return 是否连接成功 - */ - private boolean connectMqttSync() { - String host = emqxProperties.getMqttHost(); - Integer port = emqxProperties.getMqttPort(); - log.info("[connectMqttSync][开始连接 MQTT Broker, host: {}, port: {}]", host, port); - - // 使用计数器实现同步等待 - java.util.concurrent.CountDownLatch latch = new java.util.concurrent.CountDownLatch(1); - java.util.concurrent.atomic.AtomicBoolean success = new java.util.concurrent.atomic.AtomicBoolean(false); - - mqttClient.connect(port, host, connectResult -> { - if (connectResult.succeeded()) { - log.info("[connectMqttSync][MQTT 客户端连接成功, host: {}, port: {}]", host, port); - // 设置处理器 - setupMqttHandlers(); - // 订阅主题 - subscribeToTopics(); - success.set(true); - } else { - log.error("[connectMqttSync][连接 MQTT Broker 失败, host: {}, port: {}]", - host, port, connectResult.cause()); - // 首次连接失败，启动重连机制 - reconnectWithDelay(); - } - latch.countDown(); - }); - - try { - // 等待连接结果，最多等待10秒 - latch.await(10, java.util.concurrent.TimeUnit.SECONDS); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - log.error("[connectMqttSync][等待连接结果被中断]", e); - } - - return success.get(); - } - - /** - * 停止 MQTT 客户端 - */ - private void stopMqttClient() { - // 1. 取消 MQTT 主题订阅 - if (mqttClient != null && mqttClient.isConnected()) { - List topicList = emqxProperties.getMqttTopics(); - if (CollUtil.isNotEmpty(topicList)) { - for (String topic : topicList) { - try { - mqttClient.unsubscribe(topic); - log.debug("[stopMqttClient][取消订阅主题: {}]", topic); - } catch (Exception e) { - log.warn("[stopMqttClient][取消订阅主题({})异常]", topic, e); - } - } - } - } - - // 2. 关闭 MQTT 客户端 - if (mqttClient != null && mqttClient.isConnected()) { - try { - mqttClient.disconnect(); - log.info("[stopMqttClient][MQTT 客户端已断开]"); - } catch (Exception e) { - log.warn("[stopMqttClient][关闭 MQTT 客户端异常]", e); - } - } - } - - /** - * 连接 MQTT Broker 并订阅主题 - * - * @param isReconnect 是否为重连 - */ - private void connectMqtt(boolean isReconnect) { - // 1. 参数校验 - String host = emqxProperties.getMqttHost(); - Integer port = emqxProperties.getMqttPort(); - - if (isReconnect) { - log.info("[connectMqtt][开始重连 MQTT Broker, host: {}, port: {}]", host, port); - // 重连时重新创建客户端实例 - createMqttClient(); - } else { - log.info("[connectMqtt][开始连接 MQTT Broker, host: {}, port: {}]", host, port); - } - - // 2. 异步连接 - mqttClient.connect(port, host, connectResult -> { - if (!connectResult.succeeded()) { - log.error("[connectMqtt][连接 MQTT Broker 失败, host: {}, port: {}, isReconnect: {}]", - host, port, isReconnect, connectResult.cause()); - - // 首次连接失败或重连失败时，尝试重连 - if (!isReconnect) { - log.warn("[connectMqtt][首次连接失败，将开始重连机制]"); - } - reconnectWithDelay(); - return; - } - - if (isReconnect) { - log.info("[connectMqtt][MQTT 客户端重连成功, host: {}, port: {}]", host, port); - } else { - log.info("[connectMqtt][MQTT 客户端连接成功, host: {}, port: {}]", host, port); - } - - // 设置处理器 - setupMqttHandlers(); - // 订阅主题 - subscribeToTopics(); - }); - } - - /** - * 创建 MQTT 客户端 - */ - private void createMqttClient() { - MqttClientOptions options = new MqttClientOptions() - .setClientId(emqxProperties.getMqttClientId()) - .setUsername(emqxProperties.getMqttUsername()) - .setPassword(emqxProperties.getMqttPassword()) - .setSsl(emqxProperties.getMqttSsl()); - this.mqttClient = MqttClient.create(vertx, options); - } - - /** - * 设置 MQTT 处理器 - */ - private void setupMqttHandlers() { - // 由于 mqttClient 在 createMqttClient() 方法中已初始化，此处无需检查 - // 设置断开重连监听器 - mqttClient.closeHandler(closeEvent -> { - log.warn("[closeHandler][MQTT 连接已断开, 准备重连]"); - reconnectWithDelay(); - }); - - // 设置异常处理器 - mqttClient.exceptionHandler(exception -> { - log.error("[exceptionHandler][MQTT 客户端异常]", exception); - }); - - // 设置消息处理器 - // upstreamHandler 在 startMqttClient() 方法中已初始化，此处无需检查 - mqttClient.publishHandler(upstreamHandler::handle); - log.debug("[setupMqttHandlers][MQTT 消息处理器设置完成]"); - } - - /** - * 订阅设备上行消息主题 - */ - private void subscribeToTopics() { - List topicList = emqxProperties.getMqttTopics(); - if (CollUtil.isEmpty(topicList)) { - log.warn("[subscribeToTopics][订阅主题列表为空, 跳过订阅]"); - return; - } - if (mqttClient == null || !mqttClient.isConnected()) { - log.warn("[subscribeToTopics][MQTT 客户端未连接, 跳过订阅]"); - return; - } - - int qos = emqxProperties.getMqttQos(); - log.info("[subscribeToTopics][开始订阅主题, 共 {} 个, QoS: {}]", topicList.size(), qos); - - // 使用 AtomicInteger 替代数组，线程安全且更简洁 - AtomicInteger successCount = new AtomicInteger(0); - AtomicInteger failCount = new AtomicInteger(0); - - // 构建主题-QoS 映射，批量订阅 - Map topicQosMap = new HashMap<>(); - for (String topic : topicList) { - topicQosMap.put(topic, qos); - } - - // 批量订阅所有主题 - mqttClient.subscribe(topicQosMap, subscribeResult -> { - if (subscribeResult.succeeded()) { - // 批量订阅成功，记录所有主题为成功 - int successful = successCount.addAndGet(topicList.size()); - log.info("[subscribeToTopics][批量订阅主题成功, 共 {} 个主题, QoS: {}]", successful, qos); - for (String topic : topicList) { - log.debug("[subscribeToTopics][订阅主题成功, topic: {}, qos: {}]", topic, qos); - } - } else { - // 批量订阅失败，记录所有主题为失败 - int failed = failCount.addAndGet(topicList.size()); - log.error("[subscribeToTopics][批量订阅主题失败, 共 {} 个主题, 原因: {}]", - failed, subscribeResult.cause().getMessage(), subscribeResult.cause()); - for (String topic : topicList) { - log.error("[subscribeToTopics][订阅主题失败, topic: {}, qos: {}]", topic, qos); - } - } - - // 记录汇总日志 - log.info("[subscribeToTopics][主题订阅完成, 成功: {}, 失败: {}, 总计: {}]", - successCount.get(), failCount.get(), topicList.size()); - }); - } - - /** - * 延迟重连 - */ - private void reconnectWithDelay() { - long delay = emqxProperties.getReconnectDelayMs(); - vertx.setTimer(delay, timerId -> { - if (!isRunning) { - log.debug("[reconnectWithDelay][服务已停止, 取消重连]"); - return; - } - // 检查连接状态，如果已连接则无需重连 - if (mqttClient != null && mqttClient.isConnected()) { - log.debug("[reconnectWithDelay][MQTT 客户端已连接, 无需重连]"); - return; - } - log.info("[reconnectWithDelay][开始重连 MQTT Broker, 延迟: {} ms]", delay); - try { - connectMqtt(true); // 标记为重连 - } catch (Exception e) { - log.error("[reconnectWithDelay][重连失败, 将继续尝试]", e); - // 重连失败时，不需要重复调用，因为 connectMqtt(true) 内部已经处理了重连逻辑 - } - }); - } - - /** - * 发布消息到 MQTT Broker - * - * @param topic 主题 - * @param payload 消息内容 - */ - public void publishMessage(String topic, String payload) { - if (mqttClient == null || !mqttClient.isConnected()) { - log.warn("[publishMessage][MQTT 客户端未连接, 无法发布消息到 topic({})]", topic); - return; - } - MqttQoS qos = MqttQoS.valueOf(emqxProperties.getMqttQos()); - mqttClient.publish(topic, Buffer.buffer(payload), qos, false, false); - } - -} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttDownstreamHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttDownstreamHandler.java deleted file mode 100644 index 372184a..0000000 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttDownstreamHandler.java +++ /dev/null @@ -1,96 +0,0 @@ -package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router; - -import cn.hutool.core.util.StrUtil; -import cn.hutool.extra.spring.SpringUtil; -import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceRespDTO; -import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum; -import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; -import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils; -import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttUpstreamProtocol; -import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceService; -import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService; -import cn.iocoder.yudao.module.iot.gateway.util.IotMqttTopicUtils; -import lombok.extern.slf4j.Slf4j; - -/** - * IoT 网关 MQTT 下行消息处理器 - *

- * 从消息总线接收到下行消息，然后发布到 MQTT Broker，从而被设备所接收 - * - * @author 芋道源码 - */ -@Slf4j -public class IotMqttDownstreamHandler { - - private final IotMqttUpstreamProtocol protocol; - - private final IotDeviceService deviceService; - - private final IotDeviceMessageService deviceMessageService; - - public IotMqttDownstreamHandler(IotMqttUpstreamProtocol protocol) { - this.protocol = protocol; - this.deviceService = SpringUtil.getBean(IotDeviceService.class); - this.deviceMessageService = SpringUtil.getBean(IotDeviceMessageService.class); - } - - /** - * 处理下行消息 - * - * @param message 设备消息 - */ - public void handle(IotDeviceMessage message) { - // 1. 获取设备信息（使用缓存） - IotDeviceRespDTO deviceInfo = deviceService.getDeviceFromCache(message.getDeviceId()); - if (deviceInfo == null) { - log.error("[handle][设备信息({})不存在]", message.getDeviceId()); - return; - } - - // 2.1 根据方法构建主题 - String topic = buildTopicByMethod(message, deviceInfo.getProductKey(), deviceInfo.getDeviceName()); - if (StrUtil.isBlank(topic)) { - log.warn("[handle][未知的消息方法: {}]", message.getMethod()); - return; - } - - // 2.2 构建载荷 - byte[] payload = deviceMessageService.encodeDeviceMessage(message, deviceInfo.getProductKey(), deviceInfo.getDeviceName()); - - // 2.3 发布消息 - protocol.publishMessage(topic, new String(payload)); - } - - /** - * 根据消息方法和回复状态构建主题 - * - * @param message 设备消息 - * @param productKey 产品标识 - * @param deviceName 设备名称 - * @return 构建的主题，如果方法不支持返回 null - */ - private String buildTopicByMethod(IotDeviceMessage message, String productKey, String deviceName) { - // 1. 解析消息方法 - IotDeviceMessageMethodEnum methodEnum = IotDeviceMessageMethodEnum.of(message.getMethod()); - if (methodEnum == null) { - log.warn("[buildTopicByMethod][未知的消息方法: {}]", message.getMethod()); - return null; - } - - // 2. 判断是否回复消息 - boolean isReply = IotDeviceMessageUtils.isReplyMessage(message); - - // 3. 根据消息方法和回复状态，构建主题 - if (methodEnum == IotDeviceMessageMethodEnum.PROPERTY_POST && isReply) { - return IotMqttTopicUtils.buildPropertyPostReplyTopic(productKey, deviceName); - } - if (methodEnum == IotDeviceMessageMethodEnum.PROPERTY_SET && !isReply) { - return IotMqttTopicUtils.buildPropertySetTopic(productKey, deviceName); - } - - log.warn("[buildTopicByMethod][暂时不支持的下行消息: method={}, isReply={}]", - message.getMethod(), isReply); - return null; - } - -} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttHttpAuthHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttHttpAuthHandler.java deleted file mode 100644 index 3410fb3..0000000 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttHttpAuthHandler.java +++ /dev/null @@ -1,250 +0,0 @@ -package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router; - -import cn.hutool.core.util.BooleanUtil; -import cn.hutool.core.util.StrUtil; -import cn.hutool.extra.spring.SpringUtil; -import cn.iocoder.yudao.framework.common.pojo.CommonResult; -import cn.iocoder.yudao.module.iot.core.biz.IotDeviceCommonApi; -import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceAuthReqDTO; -import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; -import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils; -import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttUpstreamProtocol; -import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService; -import io.vertx.core.json.JsonObject; -import io.vertx.ext.web.RoutingContext; -import lombok.extern.slf4j.Slf4j; - -/** - * IoT 网关 MQTT HTTP 认证处理器 - *

- * 处理 EMQX 的认证请求和事件钩子，提供统一的错误处理和参数校验 - * - * @author 芋道源码 - */ -@Slf4j -public class IotMqttHttpAuthHandler { - - /** - * HTTP 成功状态码（EMQX 要求固定使用 200） - */ - private static final int SUCCESS_STATUS_CODE = 200; - - /** - * 认证允许结果 - */ - private static final String RESULT_ALLOW = "allow"; - /** - * 认证拒绝结果 - */ - private static final String RESULT_DENY = "deny"; - /** - * 认证忽略结果 - */ - private static final String RESULT_IGNORE = "ignore"; - - /** - * EMQX 事件类型常量 - */ - private static final String EVENT_CLIENT_CONNECTED = "client.connected"; - private static final String EVENT_CLIENT_DISCONNECTED = "client.disconnected"; - - private final IotMqttUpstreamProtocol protocol; - - private final IotDeviceMessageService deviceMessageService; - - private final IotDeviceCommonApi deviceApi; - - public IotMqttHttpAuthHandler(IotMqttUpstreamProtocol protocol) { - this.protocol = protocol; - this.deviceMessageService = SpringUtil.getBean(IotDeviceMessageService.class); - this.deviceApi = SpringUtil.getBean(IotDeviceCommonApi.class); - } - - /** - * EMQX 认证接口 - */ - public void handleAuth(RoutingContext context) { - try { - // 1. 参数校验 - JsonObject body = parseRequestBody(context); - if (body == null) { - return; - } - String clientId = body.getString("clientid"); - String username = body.getString("username"); - String password = body.getString("password"); - log.debug("[handleAuth][设备认证请求: clientId={}, username={}]", clientId, username); - if (StrUtil.hasEmpty(clientId, username, password)) { - log.info("[handleAuth][认证参数不完整: clientId={}, username={}]", clientId, username); - sendAuthResponse(context, RESULT_DENY); - return; - } - - // 2. 执行认证 - boolean authResult = performDeviceAuth(clientId, username, password); - log.info("[handleAuth][设备认证结果: {} -> {}]", username, authResult); - if (authResult) { - sendAuthResponse(context, RESULT_ALLOW); - } else { - sendAuthResponse(context, RESULT_DENY); - } - } catch (Exception e) { - log.error("[handleAuth][设备认证异常]", e); - sendAuthResponse(context, RESULT_IGNORE); - } - } - - /** - * EMQX 统一事件处理接口 - * 根据 EMQX 官方 Webhook 设计，统一处理所有客户端事件 - * 支持的事件类型：client.connected、client.disconnected 等 - */ - public void handleEvent(RoutingContext context) { - JsonObject body = null; - try { - // 1. 解析请求体 - body = parseRequestBody(context); - if (body == null) { - return; - } - String event = body.getString("event"); - String username = body.getString("username"); - log.debug("[handleEvent][收到事件: {} - {}]", event, username); - - // 2. 根据事件类型进行分发处理 - switch (event) { - case EVENT_CLIENT_CONNECTED: - handleClientConnected(body); - break; - case EVENT_CLIENT_DISCONNECTED: - handleClientDisconnected(body); - break; - default: - break; - } - - // EMQX Webhook 只需要 200 状态码，无需响应体 - context.response().setStatusCode(SUCCESS_STATUS_CODE).end(); - } catch (Exception e) { - log.error("[handleEvent][事件处理失败][body={}]", body != null ? body.encode() : "null", e); - // 即使处理失败，也返回 200 避免EMQX重试 - context.response().setStatusCode(SUCCESS_STATUS_CODE).end(); - } - } - - /** - * 处理客户端连接事件 - */ - private void handleClientConnected(JsonObject body) { - String username = body.getString("username"); - log.info("[handleClientConnected][设备上线: {}]", username); - handleDeviceStateChange(username, true); - } - - /** - * 处理客户端断开连接事件 - */ - private void handleClientDisconnected(JsonObject body) { - String username = body.getString("username"); - String reason = body.getString("reason"); - log.info("[handleClientDisconnected][设备下线: {} ({})]", username, reason); - handleDeviceStateChange(username, false); - } - - /** - * 解析请求体 - * - * @param context 路由上下文 - * @return 请求体JSON对象，解析失败时返回null - */ - private JsonObject parseRequestBody(RoutingContext context) { - String rawBody = null; - try { - rawBody = context.body().asString(); - JsonObject body = context.body().asJsonObject(); - if (body == null) { - log.info("[parseRequestBody][请求体为空][rawBody={}]", rawBody); - sendAuthResponse(context, RESULT_IGNORE); - return null; - } - return body; - } catch (Exception e) { - log.error("[parseRequestBody][解析请求体失败][rawBody={}]", rawBody, e); - sendAuthResponse(context, RESULT_IGNORE); - return null; - } - } - - /** - * 执行设备认证 - * - * @param clientId 客户端ID - * @param username 用户名 - * @param password 密码 - * @return 认证是否成功 - */ - private boolean performDeviceAuth(String clientId, String username, String password) { - try { - CommonResult result = deviceApi.authDevice(new IotDeviceAuthReqDTO() - .setClientId(clientId).setUsername(username).setPassword(password)); - result.checkError(); - return BooleanUtil.isTrue(result.getData()); - } catch (Exception e) { - log.error("[performDeviceAuth][认证接口调用失败: {}]", username, e); - throw e; - } - } - - /** - * 处理设备状态变化 - * - * @param username 用户名 - * @param online 是否在线 true 在线 false 离线 - */ - private void handleDeviceStateChange(String username, boolean online) { - // 1. 解析设备信息 - IotDeviceAuthUtils.DeviceInfo deviceInfo = IotDeviceAuthUtils.parseUsername(username); - if (deviceInfo == null) { - log.debug("[handleDeviceStateChange][跳过非设备连接: {}]", username); - return; - } - - try { - // 2. 构建设备状态消息 - IotDeviceMessage message = online ? IotDeviceMessage.buildStateOnline() - : IotDeviceMessage.buildStateOffline(); - - // 3. 发送设备状态消息 - deviceMessageService.sendDeviceMessage(message, - deviceInfo.getProductKey(), deviceInfo.getDeviceName(), protocol.getServerId()); - } catch (Exception e) { - log.error("[handleDeviceStateChange][发送设备状态消息失败: {}]", username, e); - } - } - - /** - * 发送 EMQX 认证响应 - * 根据 EMQX 官方文档要求，必须返回 JSON 格式响应 - * - * @param context 路由上下文 - * @param result 认证结果：allow、deny、ignore - */ - private void sendAuthResponse(RoutingContext context, String result) { - // 构建符合 EMQX 官方规范的响应 - JsonObject response = new JsonObject() - .put("result", result) - .put("is_superuser", false); - - // 可以根据业务需求添加客户端属性 - // response.put("client_attrs", new JsonObject().put("role", "device")); - - // 可以添加认证过期时间（可选） - // response.put("expire_at", System.currentTimeMillis() / 1000 + 3600); - - context.response() - .setStatusCode(SUCCESS_STATUS_CODE) - .putHeader("Content-Type", "application/json; charset=utf-8") - .end(response.encode()); - } - -} \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttUpstreamHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttUpstreamHandler.java deleted file mode 100644 index 47d0a2f..0000000 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttUpstreamHandler.java +++ /dev/null @@ -1,60 +0,0 @@ -package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router; - -import cn.hutool.core.util.StrUtil; -import cn.hutool.extra.spring.SpringUtil; -import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; -import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttUpstreamProtocol; -import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService; -import io.vertx.mqtt.messages.MqttPublishMessage; -import lombok.extern.slf4j.Slf4j; - -/** - * IoT 网关 MQTT 上行消息处理器 - * - * @author 芋道源码 - */ -@Slf4j -public class IotMqttUpstreamHandler { - - private final IotDeviceMessageService deviceMessageService; - - private final String serverId; - - public IotMqttUpstreamHandler(IotMqttUpstreamProtocol protocol) { - this.deviceMessageService = SpringUtil.getBean(IotDeviceMessageService.class); - this.serverId = protocol.getServerId(); - } - - /** - * 处理 MQTT 发布消息 - */ - public void handle(MqttPublishMessage mqttMessage) { - log.info("[handle][收到 MQTT 消息, topic: {}, payload: {}]", mqttMessage.topicName(), mqttMessage.payload()); - String topic = mqttMessage.topicName(); - byte[] payload = mqttMessage.payload().getBytes(); - try { - // 1. 解析主题，一次性获取所有信息 - String[] topicParts = topic.split("/"); - if (topicParts.length < 4 || StrUtil.hasBlank(topicParts[2], topicParts[3])) { - log.warn("[handle][topic({}) 格式不正确，无法解析有效的 productKey 和 deviceName]", topic); - return; - } - - String productKey = topicParts[2]; - String deviceName = topicParts[3]; - - // 3. 解码消息 - IotDeviceMessage message = deviceMessageService.decodeDeviceMessage(payload, productKey, deviceName); - if (message == null) { - log.warn("[handle][topic({}) payload({}) 消息解码失败]", topic, new String(payload)); - return; - } - - // 4. 发送消息到队列 - deviceMessageService.sendDeviceMessage(message, productKey, deviceName, serverId); - } catch (Exception e) { - log.error("[handle][topic({}) payload({}) 处理异常]", topic, new String(payload), e); - } - } - -} \ No newline at end of file -- libgit2 0.21.4