From e41d66ff9221b1dc1a96d2f5e5f5147bb057e640 Mon Sep 17 00:00:00 2001 From: YunaiV Date: Tue, 3 Feb 2026 12:50:02 +0800 Subject: [PATCH] feat（iot）：【协议改造】emqx 初步改造（20%）：调整包目录 --- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayConfiguration.java | 2 +- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/IotEmqxAuthEventProtocol.java | 2 +- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/IotEmqxDownstreamSubscriber.java | 30 ------------------------------ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/IotEmqxUpstreamProtocol.java | 2 +- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/downstream/IotEmqxDownstreamHandler.java | 77 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/downstream/IotEmqxDownstreamSubscriber.java | 42 ++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/upstream/IotEmqxAuthEventHandler.java | 275 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/upstream/IotEmqxUpstreamHandler.java | 60 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/router/IotEmqxAuthEventHandler.java | 275 ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/router/IotEmqxDownstreamHandler.java | 77 ----------------------------------------------------------------------------- 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/resources/application.yaml | 8 ++++---- yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/IotDirectDeviceEmqxProtocolIntegrationTest.java | 437 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 13 files changed, 898 insertions(+), 449 deletions(-) delete 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/handler/downstream/IotEmqxDownstreamHandler.java create mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/downstream/IotEmqxDownstreamSubscriber.java create mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/upstream/IotEmqxAuthEventHandler.java create mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/upstream/IotEmqxUpstreamHandler.java delete mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/router/IotEmqxAuthEventHandler.java delete mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/router/IotEmqxDownstreamHandler.java delete mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/router/IotEmqxUpstreamHandler.java create mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/IotDirectDeviceEmqxProtocolIntegrationTest.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 2115f76..aa57281 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 @@ -3,7 +3,7 @@ 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.IotProtocolManager; 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.handler.downstream.IotEmqxDownstreamSubscriber; import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.IotEmqxUpstreamProtocol; import cn.iocoder.yudao.module.iot.gateway.serialize.IotMessageSerializerManager; import io.vertx.core.Vertx; 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 index ce10cf7..5d3b5e3 100644 --- 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 @@ -2,7 +2,7 @@ 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.protocol.emqx.handler.upstream.IotEmqxAuthEventHandler; import cn.iocoder.yudao.module.iot.gateway.util.IotMqttTopicUtils; import io.vertx.core.Vertx; import io.vertx.core.http.HttpServer; 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 deleted file mode 100644 index 4b5bad2..0000000 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/IotEmqxDownstreamSubscriber.java +++ /dev/null @@ -1,30 +0,0 @@ -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.mq.message.IotDeviceMessage; -import cn.iocoder.yudao.module.iot.gateway.protocol.IotProtocol; -import cn.iocoder.yudao.module.iot.gateway.protocol.IotProtocolDownstreamSubscriber; -import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.router.IotEmqxDownstreamHandler; -import lombok.extern.slf4j.Slf4j; - -/** - * IoT 网关 EMQX 订阅者：接收下行给设备的消息 - * - * @author 芋道源码 - */ -@Slf4j -public class IotEmqxDownstreamSubscriber extends IotProtocolDownstreamSubscriber { - - private final IotEmqxDownstreamHandler downstreamHandler; - - public IotEmqxDownstreamSubscriber(IotEmqxUpstreamProtocol protocol, IotMessageBus messageBus) { - super(protocol, messageBus); - this.downstreamHandler = new IotEmqxDownstreamHandler(protocol); - } - - @Override - protected void handleMessage(IotDeviceMessage message) { - downstreamHandler.handle(message); - } - -} 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 index 47b2f16..5ebaa1f 100644 --- 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 @@ -6,7 +6,7 @@ import cn.iocoder.yudao.module.iot.core.enums.IotProtocolTypeEnum; 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.IotProtocol; -import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.router.IotEmqxUpstreamHandler; +import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.handler.upstream.IotEmqxUpstreamHandler; import io.netty.handler.codec.mqtt.MqttQoS; import io.vertx.core.Vertx; import io.vertx.core.buffer.Buffer; diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/downstream/IotEmqxDownstreamHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/downstream/IotEmqxDownstreamHandler.java new file mode 100644 index 0000000..a05fd11 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/downstream/IotEmqxDownstreamHandler.java @@ -0,0 +1,77 @@ +package cn.iocoder.yudao.module.iot.gateway.protocol.emqx.handler.downstream; + +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.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, payload); + } + + /** + * 根据消息方法和回复状态构建主题 + * + * @param message 设备消息 + * @param productKey 产品标识 + * @param deviceName 设备名称 + * @return 构建的主题，如果方法不支持返回 null + */ + private String buildTopicByMethod(IotDeviceMessage message, String productKey, String deviceName) { + // 1. 判断是否为回复消息 + boolean isReply = IotDeviceMessageUtils.isReplyMessage(message); + // 2. 根据消息方法类型构建对应的主题 + return IotMqttTopicUtils.buildTopicByMethod(message.getMethod(), productKey, deviceName, isReply); + } + +} \ 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/handler/downstream/IotEmqxDownstreamSubscriber.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/downstream/IotEmqxDownstreamSubscriber.java new file mode 100644 index 0000000..bcce471 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/downstream/IotEmqxDownstreamSubscriber.java @@ -0,0 +1,42 @@ +package cn.iocoder.yudao.module.iot.gateway.protocol.emqx.handler.downstream; + +import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus; +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; +import cn.iocoder.yudao.module.iot.gateway.protocol.IotProtocolDownstreamSubscriber; +import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.IotEmqxUpstreamProtocol; +import jakarta.annotation.PostConstruct; +import jakarta.annotation.PreDestroy; +import lombok.extern.slf4j.Slf4j; + +/** + * IoT 网关 EMQX 订阅者：接收下行给设备的消息 + * + * @author 芋道源码 + */ +@Slf4j +public class IotEmqxDownstreamSubscriber extends IotProtocolDownstreamSubscriber { + + private final IotEmqxDownstreamHandler downstreamHandler; + + public IotEmqxDownstreamSubscriber(IotEmqxUpstreamProtocol protocol, IotMessageBus messageBus) { + super(protocol, messageBus); + this.downstreamHandler = new IotEmqxDownstreamHandler(protocol); + } + + @PostConstruct + public void startSubscriber() { + // EMQX 模式下，由 Spring 管理 Bean 生命周期；需要显式启动订阅者，才能从消息总线消费下行消息并发布到 Broker + start(); + } + + @PreDestroy + public void stopSubscriber() { + stop(); + } + + @Override + protected void handleMessage(IotDeviceMessage message) { + downstreamHandler.handle(message); + } + +} diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/upstream/IotEmqxAuthEventHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/upstream/IotEmqxAuthEventHandler.java new file mode 100644 index 0000000..ae548cc --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/upstream/IotEmqxAuthEventHandler.java @@ -0,0 +1,275 @@ +package cn.iocoder.yudao.module.iot.gateway.protocol.emqx.handler.upstream; + +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.topic.IotDeviceIdentity; +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 = handleDeviceAuth(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 = parseEventRequestBody(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); + } + + /** + * 解析认证接口请求体 + *

+ * 认证接口解析失败时返回 JSON 格式响应（包含 result 字段） + * + * @param context 路由上下文 + * @return 请求体JSON对象，解析失败时返回null + */ + private JsonObject parseRequestBody(RoutingContext context) { + try { + JsonObject body = context.body().asJsonObject(); + if (body == null) { + log.info("[parseRequestBody][请求体为空]"); + sendAuthResponse(context, RESULT_IGNORE); + return null; + } + return body; + } catch (Exception e) { + log.error("[parseRequestBody][body({}) 解析请求体失败]", context.body().asString(), e); + sendAuthResponse(context, RESULT_IGNORE); + return null; + } + } + + /** + * 解析事件接口请求体 + *

+ * 事件接口解析失败时仅返回 200 状态码，无响应体（符合 EMQX Webhook 规范） + * + * @param context 路由上下文 + * @return 请求体JSON对象，解析失败时返回null + */ + private JsonObject parseEventRequestBody(RoutingContext context) { + try { + JsonObject body = context.body().asJsonObject(); + if (body == null) { + log.info("[parseEventRequestBody][请求体为空]"); + context.response().setStatusCode(SUCCESS_STATUS_CODE).end(); + return null; + } + return body; + } catch (Exception e) { + log.error("[parseEventRequestBody][body({}) 解析请求体失败]", context.body().asString(), e); + context.response().setStatusCode(SUCCESS_STATUS_CODE).end(); + return null; + } + } + + /** + * 执行设备认证 + * + * @param clientId 客户端ID + * @param username 用户名 + * @param password 密码 + * @return 认证是否成功 + */ + private boolean handleDeviceAuth(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("[handleDeviceAuth][设备({}) 认证接口调用失败]", username, e); + throw e; + } + } + + /** + * 处理设备状态变化 + * + * @param username 用户名 + * @param online 是否在线 true 在线 false 离线 + */ + private void handleDeviceStateChange(String username, boolean online) { + // 1. 解析设备信息 + IotDeviceIdentity deviceInfo = IotDeviceAuthUtils.parseUsername(username); + if (deviceInfo == null) { + log.debug("[handleDeviceStateChange][跳过非设备({})连接]", username); + return; + } + + try { + // 2. 构建设备状态消息 + IotDeviceMessage message = online ? IotDeviceMessage.buildStateUpdateOnline() + : 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/handler/upstream/IotEmqxUpstreamHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/upstream/IotEmqxUpstreamHandler.java new file mode 100644 index 0000000..5ff8d12 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/upstream/IotEmqxUpstreamHandler.java @@ -0,0 +1,60 @@ +package cn.iocoder.yudao.module.iot.gateway.protocol.emqx.handler.upstream; + +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/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 deleted file mode 100644 index 6b6694f..0000000 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/router/IotEmqxAuthEventHandler.java +++ /dev/null @@ -1,275 +0,0 @@ -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.topic.IotDeviceIdentity; -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 = handleDeviceAuth(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 = parseEventRequestBody(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); - } - - /** - * 解析认证接口请求体 - *

- * 认证接口解析失败时返回 JSON 格式响应（包含 result 字段） - * - * @param context 路由上下文 - * @return 请求体JSON对象，解析失败时返回null - */ - private JsonObject parseRequestBody(RoutingContext context) { - try { - JsonObject body = context.body().asJsonObject(); - if (body == null) { - log.info("[parseRequestBody][请求体为空]"); - sendAuthResponse(context, RESULT_IGNORE); - return null; - } - return body; - } catch (Exception e) { - log.error("[parseRequestBody][body({}) 解析请求体失败]", context.body().asString(), e); - sendAuthResponse(context, RESULT_IGNORE); - return null; - } - } - - /** - * 解析事件接口请求体 - *

- * 事件接口解析失败时仅返回 200 状态码，无响应体（符合 EMQX Webhook 规范） - * - * @param context 路由上下文 - * @return 请求体JSON对象，解析失败时返回null - */ - private JsonObject parseEventRequestBody(RoutingContext context) { - try { - JsonObject body = context.body().asJsonObject(); - if (body == null) { - log.info("[parseEventRequestBody][请求体为空]"); - context.response().setStatusCode(SUCCESS_STATUS_CODE).end(); - return null; - } - return body; - } catch (Exception e) { - log.error("[parseEventRequestBody][body({}) 解析请求体失败]", context.body().asString(), e); - context.response().setStatusCode(SUCCESS_STATUS_CODE).end(); - return null; - } - } - - /** - * 执行设备认证 - * - * @param clientId 客户端ID - * @param username 用户名 - * @param password 密码 - * @return 认证是否成功 - */ - private boolean handleDeviceAuth(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("[handleDeviceAuth][设备({}) 认证接口调用失败]", username, e); - throw e; - } - } - - /** - * 处理设备状态变化 - * - * @param username 用户名 - * @param online 是否在线 true 在线 false 离线 - */ - private void handleDeviceStateChange(String username, boolean online) { - // 1. 解析设备信息 - IotDeviceIdentity deviceInfo = IotDeviceAuthUtils.parseUsername(username); - if (deviceInfo == null) { - log.debug("[handleDeviceStateChange][跳过非设备({})连接]", username); - return; - } - - try { - // 2. 构建设备状态消息 - IotDeviceMessage message = online ? IotDeviceMessage.buildStateUpdateOnline() - : 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 deleted file mode 100644 index 06632b3..0000000 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/router/IotEmqxDownstreamHandler.java +++ /dev/null @@ -1,77 +0,0 @@ -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.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, payload); - } - - /** - * 根据消息方法和回复状态构建主题 - * - * @param message 设备消息 - * @param productKey 产品标识 - * @param deviceName 设备名称 - * @return 构建的主题，如果方法不支持返回 null - */ - private String buildTopicByMethod(IotDeviceMessage message, String productKey, String deviceName) { - // 1. 判断是否为回复消息 - boolean isReply = IotDeviceMessageUtils.isReplyMessage(message); - // 2. 根据消息方法类型构建对应的主题 - return IotMqttTopicUtils.buildTopicByMethod(message.getMethod(), productKey, deviceName, isReply); - } - -} \ 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 deleted file mode 100644 index 81d8cbb..0000000 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/router/IotEmqxUpstreamHandler.java +++ /dev/null @@ -1,60 +0,0 @@ -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/resources/application.yaml b/yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application.yaml index d24a631..4916c0d 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application.yaml +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application.yaml @@ -92,7 +92,7 @@ yudao: # 针对引入的 WebSocket 组件的配置 # ==================================== - id: websocket-json - enabled: true + enabled: false protocol: websocket port: 8094 serialize: json @@ -117,9 +117,9 @@ yudao: # 针对引入的 MQTT 组件的配置 # ==================================== - id: mqtt-json - enabled: true + enabled: false protocol: mqtt - port: 1883 + port: 1884 serialize: json mqtt: max-message-size: 8192 # 最大消息大小（字节） @@ -132,7 +132,7 @@ yudao: # 针对引入的 EMQX 组件的配置 # ==================================== emqx: - enabled: false + enabled: true http-port: 8090 # MQTT HTTP 服务端口 mqtt-host: 127.0.0.1 # MQTT Broker 地址 mqtt-port: 1883 # MQTT Broker 端口 diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/IotDirectDeviceEmqxProtocolIntegrationTest.java b/yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/IotDirectDeviceEmqxProtocolIntegrationTest.java new file mode 100644 index 0000000..a2e8591 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/IotDirectDeviceEmqxProtocolIntegrationTest.java @@ -0,0 +1,437 @@ +package cn.iocoder.yudao.module.iot.gateway.protocol.emqx; + +import cn.hutool.core.map.MapUtil; +import cn.hutool.http.HttpResponse; +import cn.hutool.http.HttpUtil; +import cn.iocoder.yudao.framework.common.util.json.JsonUtils; +import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceAuthReqDTO; +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.topic.event.IotDeviceEventPostReqDTO; +import cn.iocoder.yudao.module.iot.core.topic.property.IotDevicePropertyPostReqDTO; +import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils; +import cn.iocoder.yudao.module.iot.gateway.codec.IotDeviceMessageCodec; +import cn.iocoder.yudao.module.iot.gateway.codec.alink.IotAlinkDeviceMessageCodec; +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 lombok.extern.slf4j.Slf4j; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Disabled; +import org.junit.jupiter.api.Test; + +import java.util.concurrent.TimeUnit; + +/** + * IoT 直连设备 EMQX 协议集成测试（手动测试） + * + *

测试场景：直连设备（IotProductDeviceTypeEnum 的 DIRECT 类型）通过 EMQX Broker 连接平台 + * + *

EMQX 协议架构： + * 

+ *     +--------+       MQTT        +-------------+       HTTP Hook        +---------+
+ *     | 设备   | ----------------> | EMQX Broker | --------------------> | 网关    |
+ *     +--------+                   +-------------+                        +---------+
+ *         |                              |                                     |
+ *         | 1. 连接认证                   | 2. 调用 /mqtt/auth                   |
+ *         | 3. 发布消息                   | 4. 调用 /mqtt/event (上线/下线)       |
+ *         |                              | 5. 网关订阅 EMQX 消息                 |
+ *         |                              |                                     |
+ *
+ * + *

测试分类： + *

+ * + *

使用步骤： + *

    + *
  1. 启动 EMQX Broker（MQTT 端口 1883）
    2. + *
  2. 启动 yudao-module-iot-gateway 服务（HTTP 端口 18083）
    3. + *
  3. 配置 EMQX HTTP 认证插件指向网关的 /mqtt/auth 接口
    4. + *
  4. 配置 EMQX Webhook 插件指向网关的 /mqtt/event 接口
    5. + *
  5. 运行测试方法
    6. + *
+ * + * @author 芋道源码 + */ +@Slf4j +@Disabled +@SuppressWarnings("HttpUrlsUsage") +public class IotDirectDeviceEmqxProtocolIntegrationTest { + + private static final String SERVER_HOST = "127.0.0.1"; + /** + * EMQX 认证事件 HTTP 接口端口（网关提供给 EMQX Server 调用） + */ + private static final int HTTP_PORT = 18083; + /** + * EMQX Broker MQTT 端口（设备连接 EMQX） + */ + private static final int MQTT_PORT = 1883; + private static final int TIMEOUT_SECONDS = 10; + + private static Vertx vertx; + + // ===================== 编解码器（EMQX 使用 Alink 协议） ===================== + + private static final IotDeviceMessageCodec CODEC = new IotAlinkDeviceMessageCodec(); + + // ===================== 直连设备信息（根据实际情况修改，从 iot_device 表查询） ===================== + + private static final String PRODUCT_KEY = "4aymZgOTOOCrDKRT"; + private static final String DEVICE_NAME = "small"; + private static final String DEVICE_SECRET = "0baa4c2ecc104ae1a26b4070c218bdf3"; + + @BeforeAll + public static void setUp() { + vertx = Vertx.vertx(); + } + + @AfterAll + public static void tearDown() { + if (vertx != null) { + vertx.close(); + } + } + + // ================================================================================== + // 第一部分：模拟设备连接 EMQX Broker + // ================================================================================== + + /** + * 设备连接测试：模拟设备连接 EMQX Broker + *

+ * 当设备连接 EMQX 时，EMQX 会自动调用网关的 /mqtt/auth 接口进行认证 + */ + @Test + public void testDeviceConnect() throws Exception { + // 1. 构建认证信息 + IotDeviceAuthReqDTO authInfo = IotDeviceAuthUtils.getAuthInfo(PRODUCT_KEY, DEVICE_NAME, DEVICE_SECRET); + log.info("[testDeviceConnect][认证信息: clientId={}, username={}, password={}]", + authInfo.getClientId(), authInfo.getUsername(), authInfo.getPassword()); + + // 2. 创建客户端并连接 EMQX Broker + MqttClient client = createClient(authInfo); + try { + client.connect(MQTT_PORT, SERVER_HOST) + .toCompletionStage().toCompletableFuture().get(TIMEOUT_SECONDS, TimeUnit.SECONDS); + log.info("[testDeviceConnect][连接 EMQX Broker 成功，客户端 ID: {}]", client.clientId()); + log.info("[testDeviceConnect][EMQX 会自动调用网关的 /mqtt/auth 接口进行认证]"); + log.info("[testDeviceConnect][EMQX 会自动调用网关的 /mqtt/event 接口通知设备上线]"); + } finally { + disconnect(client); + log.info("[testDeviceConnect][EMQX 会自动调用网关的 /mqtt/event 接口通知设备下线]"); + } + } + + /** + * 属性上报测试：设备通过 EMQX Broker 发布属性消息 + *

+ * 消息流程：设备 -> EMQX Broker -> 网关（订阅 EMQX 消息） + */ + @Test + public void testPropertyPost() throws Exception { + // 1. 连接 EMQX Broker + MqttClient client = connectToEmqx(); + log.info("[testPropertyPost][连接 EMQX Broker 成功]"); + + try { + // 2.1 构建属性上报消息 + IotDeviceMessage request = IotDeviceMessage.requestOf( + IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod(), + IotDevicePropertyPostReqDTO.of(MapUtil.builder() + .put("width", 1) + .put("height", "2") + .build())); + + // 2.2 发布消息到 EMQX Broker + String topic = String.format("/sys/%s/%s/thing/property/post", PRODUCT_KEY, DEVICE_NAME); + publish(client, topic, request); + log.info("[testPropertyPost][属性上报消息已发送到 EMQX Broker]"); + log.info("[testPropertyPost][网关会通过订阅 EMQX 接收此消息]"); + + // 2.3 等待消息处理 + Thread.sleep(2000); + log.info("[testPropertyPost][请检查网关日志确认消息是否被正确处理]"); + } finally { + disconnect(client); + } + } + + /** + * 事件上报测试：设备通过 EMQX Broker 发布事件消息 + *

+ * 消息流程：设备 -> EMQX Broker -> 网关（订阅 EMQX 消息） + */ + @Test + public void testEventPost() throws Exception { + // 1. 连接 EMQX Broker + MqttClient client = connectToEmqx(); + log.info("[testEventPost][连接 EMQX Broker 成功]"); + + try { + // 2.1 构建事件上报消息 + IotDeviceMessage request = IotDeviceMessage.requestOf( + IotDeviceMessageMethodEnum.EVENT_POST.getMethod(), + IotDeviceEventPostReqDTO.of( + "eat", + MapUtil.builder().put("rice", 3).build(), + System.currentTimeMillis())); + + // 2.2 发布消息到 EMQX Broker + String topic = String.format("/sys/%s/%s/thing/event/post", PRODUCT_KEY, DEVICE_NAME); + publish(client, topic, request); + log.info("[testEventPost][事件上报消息已发送到 EMQX Broker]"); + log.info("[testEventPost][网关会通过订阅 EMQX 接收此消息]"); + + // 2.3 等待消息处理 + Thread.sleep(2000); + log.info("[testEventPost][请检查网关日志确认消息是否被正确处理]"); + } finally { + disconnect(client); + } + } + + /** + * 订阅下行消息测试：设备订阅服务端下发的消息 + *

+ * 消息流程：网关 -> EMQX Broker -> 设备 + */ + @Test + public void testSubscribe() throws Exception { + // 1. 连接 EMQX Broker + MqttClient client = connectToEmqx(); + log.info("[testSubscribe][连接 EMQX Broker 成功]"); + + try { + // 2. 设置消息处理器 + client.publishHandler(message -> log.info("[testSubscribe][收到下行消息: topic={}, payload={}]", + message.topicName(), message.payload().toString())); + + // 3. 订阅下行主题 + String topic = String.format("/sys/%s/%s/thing/service/#", PRODUCT_KEY, DEVICE_NAME); + log.info("[testSubscribe][订阅主题: {}]", topic); + subscribe(client, topic); + log.info("[testSubscribe][订阅成功，等待下行消息... (30秒后自动断开)]"); + log.info("[testSubscribe][网关下发的消息会通过 EMQX Broker 转发给设备]"); + + // 4. 保持连接 30 秒等待消息 + Thread.sleep(30000); + } finally { + disconnect(client); + } + } + + // ================================================================================== + // 第二部分：模拟 EMQX Server 调用网关 HTTP Hook 接口 + // 说明：这些接口是 EMQX Server 自动调用的，这里只是用于单独测试接口功能 + // ================================================================================== + + /** + * 认证接口测试：模拟 EMQX Server 调用 /mqtt/auth 接口 + *

+ * 注意：正常情况下此接口由 EMQX HTTP 认证插件自动调用，这里只是测试接口本身 + */ + @Test + public void testEmqxAuthHook() { + // 1.1 构建请求 + String url = String.format("http://%s:%d/mqtt/auth", SERVER_HOST, HTTP_PORT); + IotDeviceAuthReqDTO authInfo = IotDeviceAuthUtils.getAuthInfo(PRODUCT_KEY, DEVICE_NAME, DEVICE_SECRET); + // 1.2 EMQX 认证请求格式 + String payload = JsonUtils.toJsonString(MapUtil.builder() + .put("clientid", authInfo.getClientId()) + .put("username", authInfo.getUsername()) + .put("password", authInfo.getPassword()) + .build()); + // 1.3 输出请求 + log.info("[testEmqxAuthHook][模拟 EMQX Server 调用认证接口]"); + log.info("[testEmqxAuthHook][请求 URL: {}]", url); + log.info("[testEmqxAuthHook][请求体: {}]", payload); + + // 2.1 发送请求 + try (HttpResponse httpResponse = HttpUtil.createPost(url) + .header("Content-Type", "application/json") + .body(payload) + .execute()) { + // 2.2 输出结果 + log.info("[testEmqxAuthHook][响应状态码: {}]", httpResponse.getStatus()); + log.info("[testEmqxAuthHook][响应体: {}]", httpResponse.body()); + log.info("[testEmqxAuthHook][认证结果: result=allow 表示认证成功, result=deny 表示认证失败]"); + } + } + + /** + * 认证失败测试：模拟 EMQX Server 调用 /mqtt/auth 接口（错误密码） + */ + @Test + public void testEmqxAuthHookFailed() { + // 1.1 构建请求 + String url = String.format("http://%s:%d/mqtt/auth", SERVER_HOST, HTTP_PORT); + // 1.2 使用错误的密码 + String payload = JsonUtils.toJsonString(MapUtil.builder() + .put("clientid", PRODUCT_KEY + "." + DEVICE_NAME) + .put("username", DEVICE_NAME + "&" + PRODUCT_KEY) + .put("password", "wrong_password") + .build()); + // 1.3 输出请求 + log.info("[testEmqxAuthHookFailed][模拟 EMQX Server 调用认证接口（错误密码）]"); + log.info("[testEmqxAuthHookFailed][请求 URL: {}]", url); + log.info("[testEmqxAuthHookFailed][请求体: {}]", payload); + + // 2.1 发送请求 + try (HttpResponse httpResponse = HttpUtil.createPost(url) + .header("Content-Type", "application/json") + .body(payload) + .execute()) { + // 2.2 输出结果 + log.info("[testEmqxAuthHookFailed][响应状态码: {}]", httpResponse.getStatus()); + log.info("[testEmqxAuthHookFailed][响应体: {}]", httpResponse.body()); + log.info("[testEmqxAuthHookFailed][预期结果: result=deny]"); + } + } + + /** + * 设备上线事件测试：模拟 EMQX Server Webhook 调用 /mqtt/event 接口 + *

+ * 注意：正常情况下此接口由 EMQX Webhook 插件自动调用，这里只是测试接口本身 + */ + @Test + public void testEmqxClientConnectedHook() { + // 1.1 构建请求 + String url = String.format("http://%s:%d/mqtt/event", SERVER_HOST, HTTP_PORT); + IotDeviceAuthReqDTO authInfo = IotDeviceAuthUtils.getAuthInfo(PRODUCT_KEY, DEVICE_NAME, DEVICE_SECRET); + // 1.2 EMQX Webhook client.connected 事件格式 + String payload = JsonUtils.toJsonString(MapUtil.builder() + .put("event", "client.connected") + .put("clientid", authInfo.getClientId()) + .put("username", authInfo.getUsername()) + .put("peername", "127.0.0.1:12345") + .put("connected_at", System.currentTimeMillis()) + .build()); + // 1.3 输出请求 + log.info("[testEmqxClientConnectedHook][模拟 EMQX Server Webhook 调用设备上线事件]"); + log.info("[testEmqxClientConnectedHook][请求 URL: {}]", url); + log.info("[testEmqxClientConnectedHook][请求体: {}]", payload); + + // 2.1 发送请求 + try (HttpResponse httpResponse = HttpUtil.createPost(url) + .header("Content-Type", "application/json") + .body(payload) + .execute()) { + // 2.2 输出结果 + log.info("[testEmqxClientConnectedHook][响应状态码: {}]", httpResponse.getStatus()); + log.info("[testEmqxClientConnectedHook][响应体: {}]", httpResponse.body()); + log.info("[testEmqxClientConnectedHook][预期结果: 状态码 200，设备状态更新为在线]"); + } + } + + /** + * 设备下线事件测试：模拟 EMQX Server Webhook 调用 /mqtt/event 接口 + *

+ * 注意：正常情况下此接口由 EMQX Webhook 插件自动调用，这里只是测试接口本身 + */ + @Test + public void testEmqxClientDisconnectedHook() { + // 1.1 构建请求 + String url = String.format("http://%s:%d/mqtt/event", SERVER_HOST, HTTP_PORT); + IotDeviceAuthReqDTO authInfo = IotDeviceAuthUtils.getAuthInfo(PRODUCT_KEY, DEVICE_NAME, DEVICE_SECRET); + // 1.2 EMQX Webhook client.disconnected 事件格式 + String payload = JsonUtils.toJsonString(MapUtil.builder() + .put("event", "client.disconnected") + .put("clientid", authInfo.getClientId()) + .put("username", authInfo.getUsername()) + .put("reason", "normal") + .put("disconnected_at", System.currentTimeMillis()) + .build()); + // 1.3 输出请求 + log.info("[testEmqxClientDisconnectedHook][模拟 EMQX Server Webhook 调用设备下线事件]"); + log.info("[testEmqxClientDisconnectedHook][请求 URL: {}]", url); + log.info("[testEmqxClientDisconnectedHook][请求体: {}]", payload); + + // 2.1 发送请求 + try (HttpResponse httpResponse = HttpUtil.createPost(url) + .header("Content-Type", "application/json") + .body(payload) + .execute()) { + // 2.2 输出结果 + log.info("[testEmqxClientDisconnectedHook][响应状态码: {}]", httpResponse.getStatus()); + log.info("[testEmqxClientDisconnectedHook][响应体: {}]", httpResponse.body()); + log.info("[testEmqxClientDisconnectedHook][预期结果: 状态码 200，设备状态更新为离线]"); + } + } + + // ===================== 辅助方法 ===================== + + /** + * 创建 MQTT 客户端 + * + * @param authInfo 认证信息 + * @return MQTT 客户端 + */ + private MqttClient createClient(IotDeviceAuthReqDTO authInfo) { + MqttClientOptions options = new MqttClientOptions() + .setClientId(authInfo.getClientId()) + .setUsername(authInfo.getUsername()) + .setPassword(authInfo.getPassword()) + .setCleanSession(true) + .setKeepAliveInterval(60); + return MqttClient.create(vertx, options); + } + + /** + * 连接 EMQX Broker 并认证设备 + * + * @return 已认证的 MQTT 客户端 + */ + private MqttClient connectToEmqx() throws Exception { + IotDeviceAuthReqDTO authInfo = IotDeviceAuthUtils.getAuthInfo(PRODUCT_KEY, DEVICE_NAME, DEVICE_SECRET); + MqttClient client = createClient(authInfo); + client.connect(MQTT_PORT, SERVER_HOST) + .toCompletionStage().toCompletableFuture().get(TIMEOUT_SECONDS, TimeUnit.SECONDS); + return client; + } + + /** + * 订阅主题 + * + * @param client MQTT 客户端 + * @param topic 主题 + */ + private void subscribe(MqttClient client, String topic) throws Exception { + client.subscribe(topic, MqttQoS.AT_LEAST_ONCE.value()) + .toCompletionStage().toCompletableFuture().get(TIMEOUT_SECONDS, TimeUnit.SECONDS); + log.info("[subscribe][订阅主题成功: {}]", topic); + } + + /** + * 发布消息 + * + * @param client MQTT 客户端 + * @param topic 发布主题 + * @param request 请求消息 + */ + private void publish(MqttClient client, String topic, IotDeviceMessage request) throws Exception { + byte[] payload = CODEC.encode(request); + log.info("[publish][发送消息: topic={}, payload={}]", topic, new String(payload)); + client.publish(topic, Buffer.buffer(payload), MqttQoS.AT_LEAST_ONCE, false, false) + .toCompletionStage().toCompletableFuture().get(TIMEOUT_SECONDS, TimeUnit.SECONDS); + log.info("[publish][消息发布成功]"); + } + + /** + * 断开连接 + * + * @param client MQTT 客户端 + */ + private void disconnect(MqttClient client) throws Exception { + client.disconnect() + .toCompletionStage().toCompletableFuture().get(TIMEOUT_SECONDS, TimeUnit.SECONDS); + log.info("[disconnect][断开连接成功]"); + } + +} -- libgit2 0.21.4