protocolHandlers) {
- this.deviceMessageService = deviceMessageService;
- this.connectionManager = connectionManager;
- this.protocolHandlers = protocolHandlers;
- this.serverId = protocol.getServerId();
- }
-
- @Override
- public void handle(NetSocket socket) {
- String clientId = IdUtil.simpleUUID();
- log.debug("[handle][设备连接，客户端 ID: {}，地址: {}]", clientId, socket.remoteAddress());
-
- // 设置异常和关闭处理器
- socket.exceptionHandler(ex -> {
- log.warn("[handle][连接异常，客户端 ID: {}，地址: {}]", clientId, socket.remoteAddress(), ex);
- cleanupConnection(socket);
- });
-
- socket.closeHandler(v -> {
- log.debug("[handle][连接关闭，客户端 ID: {}，地址: {}]", clientId, socket.remoteAddress());
- cleanupConnection(socket);
- });
-
- // 设置消息处理器
- socket.handler(buffer -> {
- try {
- processMessage(clientId, buffer, socket);
- } catch (Exception e) {
- log.error("[handle][消息处理失败，断开连接，客户端 ID: {}，地址: {}，错误: {}]",
- clientId, socket.remoteAddress(), e.getMessage(), e);
- cleanupConnection(socket);
- socket.close();
- }
- });
- }
-
- /**
- * 处理消息
- *
- * 流程：
- * 1. 检测消息格式类型（JSON/Binary/JT808）
- * 2. 解码消息
- * 3. 查找协议处理器
- * 4. 判断是否为认证消息
- * 5. 路由到协议处理器处理
- *
- * @param clientId 客户端 ID
- * @param buffer 消息
- * @param socket 网络连接
- * @throws Exception 消息解码失败时抛出异常
- */
- private void processMessage(String clientId, Buffer buffer, NetSocket socket) throws Exception {
- // 1. 基础检查
- if (buffer == null || buffer.length() == 0) {
- return;
- }
-
- // 2. 获取消息格式类型
- String codecType = getMessageCodecType(buffer, socket);
-
- // 3. 解码消息
- IotDeviceMessage message;
- try {
- message = deviceMessageService.decodeDeviceMessage(buffer.getBytes(), codecType);
- if (message == null) {
- throw new Exception("解码后消息为空");
- }
- } catch (Exception e) {
- // 消息格式错误时抛出异常，由上层处理连接断开
- throw new Exception("消息解码失败: " + e.getMessage(), e);
- }
-
- // 4. 查找协议处理器
- ProtocolHandler handler = findProtocolHandler(message, codecType);
- if (handler == null) {
- log.warn("[processMessage][未找到协议处理器，codecType: {}, method: {}]",
- codecType, message.getMethod());
- return;
- }
-
- // 5. 判断是否为认证消息
- if (isAuthenticationMessage(message)) {
- handleAuthenticationWithProtocol(clientId, message, codecType, socket, handler);
- } else {
- handleBusinessWithProtocol(clientId, message, codecType, socket, handler);
- }
- }
-
- /**
- * 使用协议处理器处理认证消息
- *
- * 认证结果处理：
- * - SUCCESS：注册连接，发送上线消息
- * - PENDING：等待后续认证步骤（如 JT808 注册后等待鉴权）
- * - FAILURE：认证失败，不做处理（协议处理器已发送失败响应）
- *
- * @param clientId 客户端 ID
- * @param message 认证消息
- * @param codecType 消息编解码类型
- * @param socket 网络连接
- * @param handler 协议处理器
- */
- private void handleAuthenticationWithProtocol(String clientId, IotDeviceMessage message,
- String codecType, NetSocket socket,
- ProtocolHandler handler) {
- try {
- // 委托给协议处理器
- AuthResult result = handler.handleAuthentication(clientId, message, codecType, socket);
-
- // 根据认证结果处理
- if (result.isSuccess()) {
- // 认证成功：注册连接并发送上线消息
- registerConnection(socket, result.getDevice(), clientId, codecType);
- sendOnlineMessage(result.getDevice());
-
- log.info("[handleAuthentication][认证成功，设备: {}, 协议: {}]",
- result.getDevice().getDeviceName(), handler.getProtocolType());
-
- } else if (result.isPending()) {
- // 认证待定：等待后续认证步骤（如 JT808 注册后等待鉴权）
- log.info("[handleAuthentication][认证待定，设备: {}, 协议: {}, 消息: {}]",
- result.getDevice() != null ? result.getDevice().getDeviceName() : "unknown",
- handler.getProtocolType(), result.getMessage());
-
- } else {
- // 认证失败：协议处理器已发送失败响应，这里只记录日志
- log.warn("[handleAuthentication][认证失败，clientId: {}, 协议: {}, 原因: {}]",
- clientId, handler.getProtocolType(), result.getMessage());
- }
-
- } catch (Exception e) {
- log.error("[handleAuthentication][认证异常，clientId: {}, 协议: {}]",
- clientId, handler.getProtocolType(), e);
- handler.sendResponse(socket, message, false, "认证异常", codecType);
- }
- }
-
- /**
- * 使用协议处理器处理业务消息
- *
- * 前置条件：设备已认证
- *
- * @param clientId 客户端 ID
- * @param message 业务消息
- * @param codecType 消息编解码类型
- * @param socket 网络连接
- * @param handler 协议处理器
- */
- private void handleBusinessWithProtocol(String clientId, IotDeviceMessage message,
- String codecType, NetSocket socket,
- ProtocolHandler handler) {
- try {
- // 1. 检查认证状态
- if (connectionManager.isNotAuthenticated(socket)) {
- log.warn("[handleBusinessMessage][设备未认证，clientId: {}]", clientId);
- handler.sendResponse(socket, message, false, "请先进行认证", codecType);
- return;
- }
-
- // 2. 获取连接信息
- IotTcpConnectionManager.ConnectionInfo connectionInfo = connectionManager.getConnectionInfo(socket);
- if (connectionInfo == null) {
- log.error("[handleBusinessMessage][连接信息不存在，clientId: {}]", clientId);
- return;
- }
-
- // 3. 委托给协议处理器处理业务消息
- handler.handleBusinessMessage(
- clientId,
- message,
- codecType,
- socket,
- connectionInfo.getProductKey(),
- connectionInfo.getDeviceName(),
- serverId
- );
-
- } catch (Exception e) {
- log.error("[handleBusinessMessage][业务消息处理异常，clientId: {}]", clientId, e);
- }
- }
-
- /**
- * 获取消息编解码类型
- *
- * 检测优先级：
- * 1. 如果已认证，使用缓存的编解码类型
- * 2. 未认证时，通过消息格式自动检测：
- * - JT808：首尾标识符 0x7e
- * - Binary：魔术字 0x7E
- * - JSON：默认
- *
- * @param buffer 消息
- * @param socket 网络连接
- * @return 消息编解码类型
- */
- private String getMessageCodecType(Buffer buffer, NetSocket socket) {
- // 1. 如果已认证，优先使用缓存的编解码类型
- IotTcpConnectionManager.ConnectionInfo connectionInfo = connectionManager.getConnectionInfo(socket);
- if (connectionInfo != null && connectionInfo.isAuthenticated() &&
- StrUtil.isNotBlank(connectionInfo.getCodecType())) {
- return connectionInfo.getCodecType();
- }
-
- // 2. 未认证时检测消息格式类型
- byte[] data = buffer.getBytes();
-
- // 2.1 检测是否为 JT808 格式（首尾标识符 0x7e）
- if (IotJt808DeviceMessageCodec.isJt808Format(data)) {
- return CODEC_TYPE_JT808;
- }
-
- // 2.2 检测是否为自定义二进制格式（魔术字 0x7E）
- if (IotTcpBinaryDeviceMessageCodec.isBinaryFormatQuick(data)) {
- return CODEC_TYPE_BINARY;
- }
-
- // 2.3 默认为 JSON 格式
- return CODEC_TYPE_JSON;
- }
-
- /**
- * 查找协议处理器
- *
- * 遍历所有协议处理器，返回第一个能处理该消息的处理器
- *
- * @param message 设备消息
- * @param codecType 消息编解码类型
- * @return 协议处理器，未找到返回 null
- */
- private ProtocolHandler findProtocolHandler(IotDeviceMessage message, String codecType) {
- return protocolHandlers.stream()
- .filter(handler -> handler.canHandle(message, codecType))
- .findFirst()
- .orElse(null);
- }
-
- /**
- * 判断是否为认证消息
- *
- * 认证消息包括：
- * - 标准认证：auth
- * - JT808 注册：jt808.terminal.register
- * - JT808 鉴权：jt808.terminal.auth
- *
- * @param message 设备消息
- * @return true-是认证消息，false-不是
- */
- private boolean isAuthenticationMessage(IotDeviceMessage message) {
- String method = message.getMethod();
- return "auth".equals(method)
- || "jt808.terminal.register".equals(method)
- || "jt808.terminal.auth".equals(method);
- }
-
- /**
- * 注册连接信息
- *
- * 将设备连接信息注册到连接管理器，包括：
- * - 设备 ID
- * - 产品 Key
- * - 设备名称
- * - 客户端 ID
- * - 编解码类型
- * - 认证状态
- *
- * @param socket 网络连接
- * @param device 设备信息
- * @param clientId 客户端 ID
- * @param codecType 消息编解码类型
- */
- private void registerConnection(NetSocket socket, IotDeviceRespDTO device,
- String clientId, String codecType) {
- IotTcpConnectionManager.ConnectionInfo connectionInfo = new IotTcpConnectionManager.ConnectionInfo()
- .setDeviceId(device.getId())
- .setProductKey(device.getProductKey())
- .setDeviceName(device.getDeviceName())
- .setClientId(clientId)
- .setCodecType(codecType)
- .setAuthenticated(true);
-
- // 注册连接（如果设备已有其他连接，会自动断开旧连接）
- connectionManager.registerConnection(socket, device.getId(), connectionInfo);
- }
-
- /**
- * 发送设备上线消息
- *
- * 设备认证成功后，发送上线消息到消息总线，通知业务层设备已上线
- *
- * @param device 设备信息
- */
- private void sendOnlineMessage(IotDeviceRespDTO device) {
- try {
- IotDeviceMessage onlineMessage = IotDeviceMessage.buildStateUpdateOnline();
- deviceMessageService.sendDeviceMessage(onlineMessage, device.getProductKey(),
- device.getDeviceName(), serverId);
-
- log.debug("[sendOnlineMessage][发送上线消息成功，设备: {}]", device.getDeviceName());
-
- } catch (Exception e) {
- log.error("[sendOnlineMessage][发送上线消息失败，设备: {}]", device.getDeviceName(), e);
- }
- }
-
- /**
- * 清理连接
- *
- * 连接关闭或异常时，清理连接信息并发送离线消息
- *
- * @param socket 网络连接
- */
- private void cleanupConnection(NetSocket socket) {
- try {
- // 1. 发送离线消息（如果已认证）
- IotTcpConnectionManager.ConnectionInfo connectionInfo = connectionManager.getConnectionInfo(socket);
- if (connectionInfo != null && connectionInfo.isAuthenticated()) {
- IotDeviceMessage offlineMessage = IotDeviceMessage.buildStateOffline();
- deviceMessageService.sendDeviceMessage(offlineMessage, connectionInfo.getProductKey(),
- connectionInfo.getDeviceName(), serverId);
-
- log.debug("[cleanupConnection][发送离线消息成功，设备: {}]", connectionInfo.getDeviceName());
- }
-
- // 2. 注销连接
- connectionManager.unregisterConnection(socket);
-
- } catch (Exception e) {
- log.error("[cleanupConnection][清理连接失败]", e);
- }
- }
-
-}
+package com.viewsh.module.iot.gateway.protocol.tcp.router;
+
+import cn.hutool.core.util.IdUtil;
+import cn.hutool.core.util.StrUtil;
+import com.viewsh.module.iot.core.biz.dto.IotDeviceRespDTO;
+import com.viewsh.module.iot.core.mq.message.IotDeviceMessage;
+import com.viewsh.module.iot.gateway.codec.jt808.IotJt808DeviceMessageCodec;
+import com.viewsh.module.iot.gateway.codec.tcp.IotTcpBinaryDeviceMessageCodec;
+import com.viewsh.module.iot.gateway.codec.tcp.IotTcpJsonDeviceMessageCodec;
+import com.viewsh.module.iot.gateway.protocol.tcp.IotTcpUpstreamProtocol;
+import com.viewsh.module.iot.gateway.protocol.tcp.handler.AuthResult;
+import com.viewsh.module.iot.gateway.protocol.tcp.handler.ProtocolHandler;
+import com.viewsh.module.iot.gateway.protocol.tcp.manager.IotTcpConnectionManager;
+import com.viewsh.module.iot.gateway.service.device.message.IotDeviceMessageService;
+import io.vertx.core.Handler;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.net.NetSocket;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.List;
+
+/**
+ * TCP 上行消息处理器（重构版）
+ *
+ * 职责：
+ * 1. 管理 TCP 连接生命周期（连接建立、异常、关闭）
+ * 2. 检测消息格式类型（JSON/Binary/JT808）
+ * 3. 解码设备消息
+ * 4. 路由到对应的协议处理器
+ * 5. 管理设备认证状态
+ * 6. 发送设备上线/离线消息
+ *
+ * 设计原则：
+ * - 主处理器只负责路由，不包含协议特定逻辑
+ * - 协议处理器通过 Spring 自动注入，实现插件化
+ * - 认证成功后，统一注册连接和发送上线消息
+ * - 支持多步认证（如 JT808 的注册+鉴权）
+ *
+ * @author 芋道源码
+ */
+@Slf4j
+public class IotTcpUpstreamHandler implements Handler {
+
+ private static final String CODEC_TYPE_JSON = IotTcpJsonDeviceMessageCodec.TYPE;
+ private static final String CODEC_TYPE_BINARY = IotTcpBinaryDeviceMessageCodec.TYPE;
+ private static final String CODEC_TYPE_JT808 = IotJt808DeviceMessageCodec.TYPE;
+
+ private final IotDeviceMessageService deviceMessageService;
+ private final IotTcpConnectionManager connectionManager;
+ private final List protocolHandlers;
+ private final String serverId;
+
+ public IotTcpUpstreamHandler(IotTcpUpstreamProtocol protocol,
+ IotDeviceMessageService deviceMessageService,
+ IotTcpConnectionManager connectionManager,
+ List protocolHandlers) {
+ this.deviceMessageService = deviceMessageService;
+ this.connectionManager = connectionManager;
+ this.protocolHandlers = protocolHandlers;
+ this.serverId = protocol.getServerId();
+ }
+
+ @Override
+ public void handle(NetSocket socket) {
+ String clientId = IdUtil.simpleUUID();
+ log.debug("[handle][设备连接，客户端 ID: {}，地址: {}]", clientId, socket.remoteAddress());
+
+ // 设置异常和关闭处理器
+ socket.exceptionHandler(ex -> {
+ log.warn("[handle][连接异常，客户端 ID: {}，地址: {}]", clientId, socket.remoteAddress(), ex);
+ cleanupConnection(socket);
+ });
+
+ socket.closeHandler(v -> {
+ log.debug("[handle][连接关闭，客户端 ID: {}，地址: {}]", clientId, socket.remoteAddress());
+ cleanupConnection(socket);
+ });
+
+ // 设置消息处理器
+ socket.handler(buffer -> {
+ try {
+ processMessage(clientId, buffer, socket);
+ } catch (Exception e) {
+ log.error("[handle][消息处理失败，断开连接，客户端 ID: {}，地址: {}，错误: {}]",
+ clientId, socket.remoteAddress(), e.getMessage(), e);
+ cleanupConnection(socket);
+ socket.close();
+ }
+ });
+ }
+
+ /**
+ * 处理消息
+ *
+ * 流程：
+ * 1. 检测消息格式类型（JSON/Binary/JT808）
+ * 2. 解码消息
+ * 3. 查找协议处理器
+ * 4. 判断是否为认证消息
+ * 5. 路由到协议处理器处理
+ *
+ * @param clientId 客户端 ID
+ * @param buffer 消息
+ * @param socket 网络连接
+ * @throws Exception 消息解码失败时抛出异常
+ */
+ private void processMessage(String clientId, Buffer buffer, NetSocket socket) throws Exception {
+ // 1. 基础检查
+ if (buffer == null || buffer.length() == 0) {
+ return;
+ }
+
+ // 2. 获取消息格式类型
+ String codecType = getMessageCodecType(buffer, socket);
+ if (codecType == null) {
+ log.warn("[processMessage][未知消息格式，断开连接，clientId: {}，数据开头: {}]",
+ clientId, buffer.length() > 20 ? buffer.getString(0, 20) : buffer.toString());
+ socket.close();
+ return;
+ }
+
+ // 3. 解码消息
+ IotDeviceMessage message;
+ try {
+ message = deviceMessageService.decodeDeviceMessage(buffer.getBytes(), codecType);
+ if (message == null) {
+ throw new Exception("解码后消息为空");
+ }
+ } catch (Exception e) {
+ // 消息格式错误时抛出异常，由上层处理连接断开
+ throw new Exception("消息解码失败: " + e.getMessage(), e);
+ }
+
+ // 4. 查找协议处理器
+ ProtocolHandler handler = findProtocolHandler(message, codecType);
+ if (handler == null) {
+ log.warn("[processMessage][未找到协议处理器，codecType: {}, method: {}]",
+ codecType, message.getMethod());
+ return;
+ }
+
+ // 5. 判断是否为认证消息
+ if (isAuthenticationMessage(message)) {
+ handleAuthenticationWithProtocol(clientId, message, codecType, socket, handler);
+ } else {
+ handleBusinessWithProtocol(clientId, message, codecType, socket, handler);
+ }
+ }
+
+ /**
+ * 使用协议处理器处理认证消息
+ *
+ * 认证结果处理：
+ * - SUCCESS：注册连接，发送上线消息
+ * - PENDING：等待后续认证步骤（如 JT808 注册后等待鉴权）
+ * - FAILURE：认证失败，不做处理（协议处理器已发送失败响应）
+ *
+ * @param clientId 客户端 ID
+ * @param message 认证消息
+ * @param codecType 消息编解码类型
+ * @param socket 网络连接
+ * @param handler 协议处理器
+ */
+ private void handleAuthenticationWithProtocol(String clientId, IotDeviceMessage message,
+ String codecType, NetSocket socket,
+ ProtocolHandler handler) {
+ try {
+ // 委托给协议处理器
+ AuthResult result = handler.handleAuthentication(clientId, message, codecType, socket);
+
+ // 根据认证结果处理
+ if (result.isSuccess()) {
+ // 认证成功：注册连接并发送上线消息
+ registerConnection(socket, result.getDevice(), clientId, codecType);
+ sendOnlineMessage(result.getDevice());
+
+ log.info("[handleAuthentication][认证成功，设备: {}, 协议: {}]",
+ result.getDevice().getDeviceName(), handler.getProtocolType());
+
+ } else if (result.isPending()) {
+ // 认证待定：等待后续认证步骤（如 JT808 注册后等待鉴权）
+ log.info("[handleAuthentication][认证待定，设备: {}, 协议: {}, 消息: {}]",
+ result.getDevice() != null ? result.getDevice().getDeviceName() : "unknown",
+ handler.getProtocolType(), result.getMessage());
+
+ } else {
+ // 认证失败：协议处理器已发送失败响应，这里只记录日志
+ log.warn("[handleAuthentication][认证失败，clientId: {}, 协议: {}, 原因: {}]",
+ clientId, handler.getProtocolType(), result.getMessage());
+ }
+
+ } catch (Exception e) {
+ log.error("[handleAuthentication][认证异常，clientId: {}, 协议: {}]",
+ clientId, handler.getProtocolType(), e);
+ handler.sendResponse(socket, message, false, "认证异常", codecType);
+ }
+ }
+
+ /**
+ * 使用协议处理器处理业务消息
+ *
+ * 前置条件：设备已认证
+ *
+ * @param clientId 客户端 ID
+ * @param message 业务消息
+ * @param codecType 消息编解码类型
+ * @param socket 网络连接
+ * @param handler 协议处理器
+ */
+ private void handleBusinessWithProtocol(String clientId, IotDeviceMessage message,
+ String codecType, NetSocket socket,
+ ProtocolHandler handler) {
+ try {
+ // 1. 检查认证状态
+ if (connectionManager.isNotAuthenticated(socket)) {
+ log.warn("[handleBusinessMessage][设备未认证，clientId: {}]", clientId);
+ handler.sendResponse(socket, message, false, "请先进行认证", codecType);
+ return;
+ }
+
+ // 2. 获取连接信息
+ IotTcpConnectionManager.ConnectionInfo connectionInfo = connectionManager.getConnectionInfo(socket);
+ if (connectionInfo == null) {
+ log.error("[handleBusinessMessage][连接信息不存在，clientId: {}]", clientId);
+ return;
+ }
+
+ // 3. 委托给协议处理器处理业务消息
+ handler.handleBusinessMessage(
+ clientId,
+ message,
+ codecType,
+ socket,
+ connectionInfo.getProductKey(),
+ connectionInfo.getDeviceName(),
+ serverId);
+
+ } catch (Exception e) {
+ log.error("[handleBusinessMessage][业务消息处理异常，clientId: {}]", clientId, e);
+ }
+ }
+
+ /**
+ * 获取消息编解码类型
+ *
+ * 检测优先级：
+ * 1. 如果已认证，使用缓存的编解码类型
+ * 2. 未认证时，通过消息格式自动检测：
+ * - JT808：首尾标识符 0x7e
+ * - Binary：魔术字 0x7E
+ * - JSON：默认
+ *
+ * @param buffer 消息
+ * @param socket 网络连接
+ * @return 消息编解码类型
+ */
+ private String getMessageCodecType(Buffer buffer, NetSocket socket) {
+ // 1. 如果已认证，优先使用缓存的编解码类型
+ IotTcpConnectionManager.ConnectionInfo connectionInfo = connectionManager.getConnectionInfo(socket);
+ if (connectionInfo != null && connectionInfo.isAuthenticated() &&
+ StrUtil.isNotBlank(connectionInfo.getCodecType())) {
+ return connectionInfo.getCodecType();
+ }
+
+ // 2. 未认证时检测消息格式类型
+ byte[] data = buffer.getBytes();
+
+ // 2.1 检测是否为 JT808 格式（首尾标识符 0x7e）
+ if (IotJt808DeviceMessageCodec.isJt808Format(data)) {
+ return CODEC_TYPE_JT808;
+ }
+
+ // 2.2 检测是否为自定义二进制格式（魔术字 0x7E）
+ if (IotTcpBinaryDeviceMessageCodec.isBinaryFormatQuick(data)) {
+ return CODEC_TYPE_BINARY;
+ }
+
+ // 2.3 检测是否为 JSON 格式（以 { 或 [ 开头）
+ String jsonStr = StrUtil.utf8Str(data).trim();
+ if (StrUtil.startWithAny(jsonStr, "{", "[")) {
+ return CODEC_TYPE_JSON;
+ }
+
+ // 2.4 未知格式
+ return null;
+ }
+
+ /**
+ * 查找协议处理器
+ *
+ * 遍历所有协议处理器，返回第一个能处理该消息的处理器
+ *
+ * @param message 设备消息
+ * @param codecType 消息编解码类型
+ * @return 协议处理器，未找到返回 null
+ */
+ private ProtocolHandler findProtocolHandler(IotDeviceMessage message, String codecType) {
+ return protocolHandlers.stream()
+ .filter(handler -> handler.canHandle(message, codecType))
+ .findFirst()
+ .orElse(null);
+ }
+
+ /**
+ * 判断是否为认证消息
+ *
+ * 认证消息包括：
+ * - 标准认证：auth
+ * - JT808 注册：jt808.terminal.register
+ * - JT808 鉴权：jt808.terminal.auth
+ *
+ * @param message 设备消息
+ * @return true-是认证消息，false-不是
+ */
+ private boolean isAuthenticationMessage(IotDeviceMessage message) {
+ String method = message.getMethod();
+ return "auth".equals(method)
+ || "jt808.terminal.register".equals(method)
+ || "jt808.terminal.auth".equals(method);
+ }
+
+ /**
+ * 注册连接信息
+ *
+ * 将设备连接信息注册到连接管理器，包括：
+ * - 设备 ID
+ * - 产品 Key
+ * - 设备名称
+ * - 客户端 ID
+ * - 编解码类型
+ * - 认证状态
+ *
+ * @param socket 网络连接
+ * @param device 设备信息
+ * @param clientId 客户端 ID
+ * @param codecType 消息编解码类型
+ */
+ private void registerConnection(NetSocket socket, IotDeviceRespDTO device,
+ String clientId, String codecType) {
+ IotTcpConnectionManager.ConnectionInfo connectionInfo = new IotTcpConnectionManager.ConnectionInfo()
+ .setDeviceId(device.getId())
+ .setProductKey(device.getProductKey())
+ .setDeviceName(device.getDeviceName())
+ .setClientId(clientId)
+ .setCodecType(codecType)
+ .setAuthenticated(true);
+
+ // 注册连接（如果设备已有其他连接，会自动断开旧连接）
+ connectionManager.registerConnection(socket, device.getId(), connectionInfo);
+ }
+
+ /**
+ * 发送设备上线消息
+ *
+ * 设备认证成功后，发送上线消息到消息总线，通知业务层设备已上线
+ *
+ * @param device 设备信息
+ */
+ private void sendOnlineMessage(IotDeviceRespDTO device) {
+ try {
+ IotDeviceMessage onlineMessage = IotDeviceMessage.buildStateUpdateOnline();
+ deviceMessageService.sendDeviceMessage(onlineMessage, device.getProductKey(),
+ device.getDeviceName(), serverId);
+
+ log.debug("[sendOnlineMessage][发送上线消息成功，设备: {}]", device.getDeviceName());
+
+ } catch (Exception e) {
+ log.error("[sendOnlineMessage][发送上线消息失败，设备: {}]", device.getDeviceName(), e);
+ }
+ }
+
+ /**
+ * 清理连接
+ *
+ * 连接关闭或异常时，清理连接信息并发送离线消息
+ *
+ * @param socket 网络连接
+ */
+ private void cleanupConnection(NetSocket socket) {
+ try {
+ // 1. 发送离线消息（如果已认证）
+ IotTcpConnectionManager.ConnectionInfo connectionInfo = connectionManager.getConnectionInfo(socket);
+ if (connectionInfo != null && connectionInfo.isAuthenticated()) {
+ IotDeviceMessage offlineMessage = IotDeviceMessage.buildStateOffline();
+ deviceMessageService.sendDeviceMessage(offlineMessage, connectionInfo.getProductKey(),
+ connectionInfo.getDeviceName(), serverId);
+
+ log.debug("[cleanupConnection][发送离线消息成功，设备: {}]", connectionInfo.getDeviceName());
+ }
+
+ // 2. 注销连接
+ connectionManager.unregisterConnection(socket);
+
+ } catch (Exception e) {
+ log.error("[cleanupConnection][清理连接失败]", e);
+ }
+ }
+
+}
diff --git a/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/api/device/IoTDeviceApiImpl.java b/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/api/device/IoTDeviceApiImpl.java
index 58fed2d..19e2798 100644
--- a/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/api/device/IoTDeviceApiImpl.java
+++ b/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/api/device/IoTDeviceApiImpl.java
@@ -1,75 +1,75 @@
-package com.viewsh.module.iot.api.device;
-
-import com.viewsh.framework.common.enums.RpcConstants;
-import com.viewsh.framework.common.pojo.CommonResult;
-import com.viewsh.framework.common.util.object.BeanUtils;
-import com.viewsh.module.iot.core.biz.IotDeviceCommonApi;
-import com.viewsh.module.iot.core.biz.dto.IotDeviceAuthReqDTO;
-import com.viewsh.module.iot.core.biz.dto.IotDeviceGetReqDTO;
-import com.viewsh.module.iot.core.biz.dto.IotDeviceRespDTO;
-import com.viewsh.module.iot.dal.dataobject.device.IotDeviceDO;
-import com.viewsh.module.iot.dal.dataobject.product.IotProductDO;
-import com.viewsh.module.iot.service.device.IotDeviceService;
-import com.viewsh.module.iot.service.product.IotProductService;
-import jakarta.annotation.Resource;
-import jakarta.annotation.security.PermitAll;
-import org.springframework.context.annotation.Primary;
-import org.springframework.validation.annotation.Validated;
-import org.springframework.web.bind.annotation.PostMapping;
-import org.springframework.web.bind.annotation.RequestBody;
-import org.springframework.web.bind.annotation.RestController;
-
-import static com.viewsh.framework.common.pojo.CommonResult.success;
-
-/**
- * IoT 设备 API 实现类
- *
- * @author haohao
- */
-@RestController
-@Validated
-@Primary // 保证优先匹配，因为 viewsh-iot-gateway 也有 IotDeviceCommonApi 的实现，并且也可能会被 biz 引入
-public class IoTDeviceApiImpl implements IotDeviceCommonApi {
-
- @Resource
- private IotDeviceService deviceService;
- @Resource
- private IotProductService productService;
-
- @Override
- @PostMapping(RpcConstants.RPC_API_PREFIX + "/iot/device/auth")
- @PermitAll
- public CommonResult authDevice(@RequestBody IotDeviceAuthReqDTO authReqDTO) {
- return success(deviceService.authDevice(authReqDTO));
- }
-
- @Override
- @PostMapping(RpcConstants.RPC_API_PREFIX + "/iot/device/get") // 特殊：方便调用，暂时使用 POST，实际更推荐 GET
- @PermitAll
- public CommonResult getDevice(@RequestBody IotDeviceGetReqDTO getReqDTO) {
- IotDeviceDO device;
-
- // 查询优先级：id > (productKey + deviceName) > deviceName
- if (getReqDTO.getId() != null) {
- // 通过设备 ID 查询
- device = deviceService.getDeviceFromCache(getReqDTO.getId());
- } else if (getReqDTO.getProductKey() != null && getReqDTO.getDeviceName() != null) {
- // 通过 productKey + deviceName 查询
- device = deviceService.getDeviceFromCache(getReqDTO.getProductKey(), getReqDTO.getDeviceName());
- } else if (getReqDTO.getDeviceName() != null) {
- // 仅通过 deviceName 查询（用于 JT808 等协议，终端手机号应该是全局唯一的）
- device = deviceService.getDeviceFromCacheByDeviceName(getReqDTO.getDeviceName());
- } else {
- device = null;
- }
-
- return success(BeanUtils.toBean(device, IotDeviceRespDTO.class, deviceDTO -> {
- IotProductDO product = productService.getProductFromCache(deviceDTO.getProductId());
- if (product != null) {
- deviceDTO.setCodecType(product.getCodecType());
- deviceDTO.setProductAuthType(product.getAuthType());
- }
- }));
- }
-
+package com.viewsh.module.iot.api.device;
+
+import com.viewsh.framework.common.enums.RpcConstants;
+import com.viewsh.framework.common.pojo.CommonResult;
+import com.viewsh.framework.common.util.object.BeanUtils;
+import com.viewsh.module.iot.core.biz.IotDeviceCommonApi;
+import com.viewsh.module.iot.core.biz.dto.IotDeviceAuthReqDTO;
+import com.viewsh.module.iot.core.biz.dto.IotDeviceGetReqDTO;
+import com.viewsh.module.iot.core.biz.dto.IotDeviceRespDTO;
+import com.viewsh.module.iot.dal.dataobject.device.IotDeviceDO;
+import com.viewsh.module.iot.dal.dataobject.product.IotProductDO;
+import com.viewsh.module.iot.service.device.IotDeviceService;
+import com.viewsh.module.iot.service.product.IotProductService;
+import jakarta.annotation.Resource;
+import jakarta.annotation.security.PermitAll;
+import org.springframework.context.annotation.Primary;
+import org.springframework.validation.annotation.Validated;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RestController;
+
+import static com.viewsh.framework.common.pojo.CommonResult.success;
+
+/**
+ * IoT 设备 API 实现类（供IOT GATEWAY使用）
+ *
+ * @author haohao
+ */
+@RestController
+@Validated
+@Primary // 保证优先匹配，因为 viewsh-iot-gateway 也有 IotDeviceCommonApi 的实现，并且也可能会被 biz 引入
+public class IoTDeviceApiImpl implements IotDeviceCommonApi {
+
+ @Resource
+ private IotDeviceService deviceService;
+ @Resource
+ private IotProductService productService;
+
+ @Override
+ @PostMapping(RpcConstants.RPC_API_PREFIX + "/iot/device/auth")
+ @PermitAll
+ public CommonResult authDevice(@RequestBody IotDeviceAuthReqDTO authReqDTO) {
+ return success(deviceService.authDevice(authReqDTO));
+ }
+
+ @Override
+ @PostMapping(RpcConstants.RPC_API_PREFIX + "/iot/device/get") // 特殊：方便调用，暂时使用 POST，实际更推荐 GET
+ @PermitAll
+ public CommonResult getDevice(@RequestBody IotDeviceGetReqDTO getReqDTO) {
+ IotDeviceDO device;
+
+ // 查询优先级：id > (productKey + deviceName) > deviceName
+ if (getReqDTO.getId() != null) {
+ // 通过设备 ID 查询
+ device = deviceService.getDeviceFromCache(getReqDTO.getId());
+ } else if (getReqDTO.getProductKey() != null && getReqDTO.getDeviceName() != null) {
+ // 通过 productKey + deviceName 查询
+ device = deviceService.getDeviceFromCache(getReqDTO.getProductKey(), getReqDTO.getDeviceName());
+ } else if (getReqDTO.getDeviceName() != null) {
+ // 仅通过 deviceName 查询（用于 JT808 等协议，终端手机号应该是全局唯一的）
+ device = deviceService.getDeviceFromCacheByDeviceName(getReqDTO.getDeviceName());
+ } else {
+ device = null;
+ }
+
+ return success(BeanUtils.toBean(device, IotDeviceRespDTO.class, deviceDTO -> {
+ IotProductDO product = productService.getProductFromCache(deviceDTO.getProductId());
+ if (product != null) {
+ deviceDTO.setCodecType(product.getCodecType());
+ deviceDTO.setProductAuthType(product.getAuthType());
+ }
+ }));
+ }
+
}
\ No newline at end of file
diff --git a/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/api/device/IotDeviceControlApiImpl.java b/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/api/device/IotDeviceControlApiImpl.java
new file mode 100644
index 0000000..c72af38
--- /dev/null
+++ b/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/api/device/IotDeviceControlApiImpl.java
@@ -0,0 +1,123 @@
+package com.viewsh.module.iot.api.device;
+
+import cn.hutool.core.map.MapUtil;
+import com.viewsh.framework.common.pojo.CommonResult;
+import com.viewsh.module.iot.api.device.dto.IotDeviceServiceInvokeReqDTO;
+import com.viewsh.module.iot.api.device.dto.IotDeviceServiceInvokeRespDTO;
+import com.viewsh.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import com.viewsh.module.iot.core.enums.IotDeviceStateEnum;
+import com.viewsh.module.iot.core.mq.message.IotDeviceMessage;
+import com.viewsh.module.iot.dal.dataobject.device.IotDeviceDO;
+import com.viewsh.module.iot.service.device.IotDeviceService;
+import com.viewsh.module.iot.service.device.message.IotDeviceMessageService;
+import io.swagger.v3.oas.annotations.Operation;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.annotation.Primary;
+import org.springframework.validation.annotation.Validated;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.time.LocalDateTime;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static com.viewsh.framework.common.pojo.CommonResult.success;
+
+/**
+ * IoT 设备控制 API 实现类
+ *
+ * 提供 RPC 接口供其他模块调用 IoT 设备服务
+ *
+ * @author lzh
+ */
+@RestController
+@Validated
+@Primary
+@Slf4j
+public class IotDeviceControlApiImpl implements IotDeviceControlApi {
+
+ @Resource
+ private IotDeviceService deviceService;
+
+ @Resource
+ private IotDeviceMessageService deviceMessageService;
+
+ @Override
+ @PostMapping(PREFIX + "/invoke-service")
+ @Operation(summary = "调用设备服务")
+ public CommonResult invokeService(@RequestBody IotDeviceServiceInvokeReqDTO reqDTO) {
+ try {
+ // 1. 获取设备信息
+ IotDeviceDO device = deviceService.getDeviceFromCache(reqDTO.getDeviceId());
+ if (device == null) {
+ return success(IotDeviceServiceInvokeRespDTO.builder()
+ .success(false)
+ .code(404)
+ .errorMsg("设备不存在")
+ .responseTime(LocalDateTime.now())
+ .build());
+ }
+
+ // 2. 检查设备是否在线
+ if (!IotDeviceStateEnum.ONLINE.getState().equals(device.getState())) {
+ return success(IotDeviceServiceInvokeRespDTO.builder()
+ .success(false)
+ .code(400)
+ .errorMsg("设备不在线，当前状态: " + device.getState())
+ .responseTime(LocalDateTime.now())
+ .build());
+ }
+
+ // 3. 构建服务调用消息
+ @SuppressWarnings("unchecked")
+ Map params = (Map) (Map) MapUtil.builder()
+ .put("identifier", reqDTO.getIdentifier())
+ .put("params", reqDTO.getParams() != null ? reqDTO.getParams() : new HashMap<>())
+ .build();
+
+ IotDeviceMessage message = IotDeviceMessage.requestOf(
+ IotDeviceMessageMethodEnum.SERVICE_INVOKE.getMethod(), params);
+
+ // 4. 发送消息
+ IotDeviceMessage result = deviceMessageService.sendDeviceMessage(message, device);
+
+ // 5. 返回结果（异步模式，立即返回消息ID）
+ return success(IotDeviceServiceInvokeRespDTO.builder()
+ .messageId(result.getId())
+ .success(true)
+ .data(null)
+ .responseTime(LocalDateTime.now())
+ .build());
+
+ } catch (Exception e) {
+ log.error("[invokeService] 设备服务调用失败: deviceId={}, identifier={}",
+ reqDTO.getDeviceId(), reqDTO.getIdentifier(), e);
+ return success(IotDeviceServiceInvokeRespDTO.builder()
+ .success(false)
+ .code(500)
+ .errorMsg(e.getMessage())
+ .responseTime(LocalDateTime.now())
+ .build());
+ }
+ }
+
+ @Override
+ @PostMapping(PREFIX + "/invoke-service-batch")
+ @Operation(summary = "批量调用设备服务")
+ public CommonResult> invokeServiceBatch(
+ @RequestBody List reqDTOList) {
+
+ List results = new java.util.ArrayList<>();
+ for (IotDeviceServiceInvokeReqDTO reqDTO : reqDTOList) {
+ CommonResult result = invokeService(reqDTO);
+ results.add(result.getData());
+ }
+
+ return success(results);
+ }
+
+}
+
diff --git a/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/api/device/IotDevicePropertyQueryApiImpl.java b/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/api/device/IotDevicePropertyQueryApiImpl.java
new file mode 100644
index 0000000..f5c9047
--- /dev/null
+++ b/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/api/device/IotDevicePropertyQueryApiImpl.java
@@ -0,0 +1,175 @@
+package com.viewsh.module.iot.api.device;
+
+import com.viewsh.framework.common.pojo.CommonResult;
+import com.viewsh.module.iot.api.device.dto.property.DevicePropertyBatchQueryReqDTO;
+import com.viewsh.module.iot.api.device.dto.property.DevicePropertyHistoryQueryReqDTO;
+import com.viewsh.module.iot.api.device.dto.property.DevicePropertyHistoryRespDTO;
+import com.viewsh.module.iot.api.device.dto.property.DevicePropertyRespDTO;
+import com.viewsh.module.iot.controller.admin.device.vo.property.IotDevicePropertyHistoryListReqVO;
+import com.viewsh.module.iot.controller.admin.device.vo.property.IotDevicePropertyRespVO;
+import com.viewsh.module.iot.dal.dataobject.device.IotDeviceDO;
+import com.viewsh.module.iot.dal.dataobject.device.IotDevicePropertyDO;
+import com.viewsh.module.iot.service.device.IotDeviceService;
+import com.viewsh.module.iot.service.device.property.IotDevicePropertyService;
+import io.swagger.v3.oas.annotations.Operation;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.annotation.Primary;
+import org.springframework.validation.annotation.Validated;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static com.viewsh.framework.common.pojo.CommonResult.success;
+
+/**
+ * IoT 设备属性查询 API 实现类
+ *
+ * 提供 RPC 接口供其他模块（如 Ops 模块）查询设备属性
+ *
+ * @author lzh
+ */
+@RestController
+@Validated
+@Primary
+@Slf4j
+public class IotDevicePropertyQueryApiImpl implements IotDevicePropertyQueryApi {
+
+ @Resource
+ private IotDeviceService deviceService;
+
+ @Resource
+ private IotDevicePropertyService devicePropertyService;
+
+ @Override
+ @GetMapping(PREFIX + "/get")
+ @Operation(summary = "获取设备单个属性")
+ public CommonResult getProperty(@RequestParam("deviceId") Long deviceId,
+ @RequestParam("identifier") String identifier) {
+ try {
+ Map properties = devicePropertyService.getLatestDeviceProperties(deviceId);
+ IotDevicePropertyDO property = properties.get(identifier);
+
+ if (property == null) {
+ return success(DevicePropertyRespDTO.builder()
+ .identifier(identifier)
+ .value(null)
+ .updateTime(null)
+ .build());
+ }
+
+ return success(DevicePropertyRespDTO.builder()
+ .identifier(identifier)
+ .value(property.getValue())
+ .updateTime(property.getUpdateTime() != null ?
+ property.getUpdateTime().atZone(java.time.ZoneId.systemDefault()).toInstant().toEpochMilli() : null)
+ .build());
+ } catch (Exception e) {
+ log.error("[getProperty] 获取设备属性失败: deviceId={}, identifier={}", deviceId, identifier, e);
+ return success(DevicePropertyRespDTO.builder()
+ .identifier(identifier)
+ .value(null)
+ .updateTime(null)
+ .build());
+ }
+ }
+
+ @Override
+ @GetMapping(PREFIX + "/get-latest")
+ @Operation(summary = "获取设备最新属性")
+ public CommonResult> getLatestProperties(@RequestParam("deviceId") Long deviceId) {
+ try {
+ // 验证设备是否存在
+ IotDeviceDO device = deviceService.getDeviceFromCache(deviceId);
+ if (device == null) {
+ log.warn("[getLatestProperties] 设备不存在: deviceId={}", deviceId);
+ return success(Map.of());
+ }
+
+ Map properties = devicePropertyService.getLatestDeviceProperties(deviceId);
+ Map result = properties.entrySet().stream()
+ .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().getValue()));
+
+ return success(result);
+ } catch (Exception e) {
+ log.error("[getLatestProperties] 获取设备最新属性失败: deviceId={}", deviceId, e);
+ return success(Map.of());
+ }
+ }
+
+ @Override
+ @PostMapping(PREFIX + "/batch-get")
+ @Operation(summary = "批量获取多个设备的指定属性")
+ public CommonResult>> batchGetProperties(
+ @RequestBody DevicePropertyBatchQueryReqDTO reqDTO) {
+ try {
+ Map> result = reqDTO.getDeviceIds().stream()
+ .collect(Collectors.toMap(
+ deviceId -> deviceId,
+ deviceId -> {
+ Map properties = devicePropertyService.getLatestDeviceProperties(deviceId);
+
+ if (reqDTO.getIdentifiers() == null || reqDTO.getIdentifiers().isEmpty()) {
+ // 返回所有属性
+ return properties.entrySet().stream()
+ .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().getValue()));
+ } else {
+ // 只返回指定的属性
+ return reqDTO.getIdentifiers().stream()
+ .filter(properties::containsKey)
+ .collect(Collectors.toMap(
+ identifier -> identifier,
+ identifier -> properties.get(identifier).getValue()));
+ }
+ }
+ ));
+
+ return success(result);
+ } catch (Exception e) {
+ log.error("[batchGetProperties] 批量获取设备属性失败: deviceIds={}", reqDTO.getDeviceIds(), e);
+ return success(Map.of());
+ }
+ }
+
+ @Override
+ @PostMapping(PREFIX + "/history")
+ @Operation(summary = "查询设备属性历史数据")
+ public CommonResult> getPropertyHistory(
+ @RequestBody DevicePropertyHistoryQueryReqDTO reqDTO) {
+ try {
+ // 构建查询请求
+ IotDevicePropertyHistoryListReqVO listReqVO = new IotDevicePropertyHistoryListReqVO();
+ listReqVO.setDeviceId(reqDTO.getDeviceId());
+ listReqVO.setIdentifier(reqDTO.getIdentifier());
+ if (reqDTO.getStartTime() != null && reqDTO.getEndTime() != null) {
+ listReqVO.setTimes(new LocalDateTime[]{reqDTO.getStartTime(), reqDTO.getEndTime()});
+ }
+ // listReqVO.setPageSize(reqDTO.getLimit()); // 暂不支持分页
+
+ List historyList = devicePropertyService.getHistoryDevicePropertyList(listReqVO);
+
+ List result = historyList.stream()
+ .map(vo -> DevicePropertyHistoryRespDTO.builder()
+ .identifier(vo.getIdentifier())
+ .value(vo.getValue())
+ .timestamp(vo.getUpdateTime())
+ .build())
+ .collect(Collectors.toList());
+
+ return success(result);
+ } catch (Exception e) {
+ log.error("[getPropertyHistory] 查询设备属性历史失败: deviceId={}, identifier={}",
+ reqDTO.getDeviceId(), reqDTO.getIdentifier(), e);
+ return success(List.of());
+ }
+ }
+
+}
diff --git a/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/api/device/IotDeviceStatusQueryApiImpl.java b/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/api/device/IotDeviceStatusQueryApiImpl.java
new file mode 100644
index 0000000..52d9be2
--- /dev/null
+++ b/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/api/device/IotDeviceStatusQueryApiImpl.java
@@ -0,0 +1,66 @@
+package com.viewsh.module.iot.api.device;
+
+import com.viewsh.framework.common.pojo.CommonResult;
+import com.viewsh.module.iot.api.device.dto.status.DeviceStatusRespDTO;
+import com.viewsh.module.iot.core.enums.IotDeviceStateEnum;
+import com.viewsh.module.iot.dal.dataobject.device.IotDeviceDO;
+import com.viewsh.module.iot.service.device.IotDeviceService;
+import io.swagger.v3.oas.annotations.Operation;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.annotation.Primary;
+import org.springframework.validation.annotation.Validated;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+import static com.viewsh.framework.common.pojo.CommonResult.success;
+
+/**
+ * IoT 设备状态查询 API 实现类
+ *
+ * 提供 RPC 接口供其他模块查询设备状态
+ *
+ * @author lzh
+ */
+@RestController
+@Validated
+@Primary
+@Slf4j
+public class IotDeviceStatusQueryApiImpl implements IotDeviceStatusQueryApi {
+
+ @Resource
+ private IotDeviceService deviceService;
+
+ @Override
+ @GetMapping(PREFIX + "/is-online")
+ @Operation(summary = "检查设备是否在线")
+ public CommonResult isDeviceOnline(@RequestParam("deviceId") Long deviceId) {
+ IotDeviceDO device = deviceService.getDeviceFromCache(deviceId);
+ if (device == null) {
+ return success(false);
+ }
+ return success(IotDeviceStateEnum.ONLINE.getState().equals(device.getState()));
+ }
+
+ @Override
+ @GetMapping(PREFIX + "/get-status")
+ @Operation(summary = "获取设备状态")
+ public CommonResult getDeviceStatus(@RequestParam("deviceId") Long deviceId) {
+ IotDeviceDO device = deviceService.getDeviceFromCache(deviceId);
+ if (device == null) {
+ return success(DeviceStatusRespDTO.builder()
+ .deviceId(deviceId)
+ .status(IotDeviceStateEnum.OFFLINE.getState())
+ .build());
+ }
+ return success(DeviceStatusRespDTO.builder()
+ .deviceId(device.getId())
+ .deviceCode(device.getSerialNumber())
+ .status(device.getState())
+ .statusChangeTime(IotDeviceStateEnum.ONLINE.getState().equals(device.getState()) ?
+ device.getOnlineTime() : device.getOfflineTime())
+ .build());
+ }
+
+}
diff --git a/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/service/device/IotDeviceServiceImpl.java b/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/service/device/IotDeviceServiceImpl.java
index ad7ef73..4067757 100644
--- a/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/service/device/IotDeviceServiceImpl.java
+++ b/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/service/device/IotDeviceServiceImpl.java
@@ -1,591 +1,664 @@
-package com.viewsh.module.iot.service.device;
-
-import cn.hutool.core.collection.CollUtil;
-import cn.hutool.core.util.IdUtil;
-import cn.hutool.core.util.ObjUtil;
-import cn.hutool.core.util.StrUtil;
-import cn.hutool.extra.spring.SpringUtil;
-import com.viewsh.framework.common.exception.ServiceException;
-import com.viewsh.framework.common.pojo.PageResult;
-import com.viewsh.framework.common.util.object.BeanUtils;
-import com.viewsh.framework.common.util.validation.ValidationUtils;
-import com.viewsh.framework.tenant.core.aop.TenantIgnore;
-import com.viewsh.framework.tenant.core.util.TenantUtils;
-import com.viewsh.module.iot.controller.admin.device.vo.device.*;
-import com.viewsh.module.iot.core.biz.dto.IotDeviceAuthReqDTO;
-import com.viewsh.module.iot.core.enums.IotAuthTypeEnum;
-import com.viewsh.module.iot.core.enums.IotDeviceStateEnum;
-import com.viewsh.module.iot.core.util.IotDeviceAuthUtils;
-import com.viewsh.module.iot.dal.dataobject.device.IotDeviceDO;
-import com.viewsh.module.iot.dal.dataobject.device.IotDeviceGroupDO;
-import com.viewsh.module.iot.dal.dataobject.product.IotProductDO;
-import com.viewsh.module.iot.dal.mysql.device.IotDeviceMapper;
-import com.viewsh.module.iot.dal.redis.RedisKeyConstants;
-import com.viewsh.module.iot.enums.product.IotProductDeviceTypeEnum;
-import com.viewsh.module.iot.service.product.IotProductService;
-import jakarta.annotation.Resource;
-import jakarta.validation.ConstraintViolationException;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.cache.annotation.CacheEvict;
-import org.springframework.cache.annotation.Cacheable;
-import org.springframework.cache.annotation.Caching;
-import org.springframework.context.annotation.Lazy;
-import org.springframework.stereotype.Service;
-import org.springframework.transaction.annotation.Transactional;
-import org.springframework.validation.annotation.Validated;
-
-import javax.annotation.Nullable;
-import java.time.LocalDateTime;
-import java.util.*;
-
-import static com.viewsh.framework.common.exception.util.ServiceExceptionUtil.exception;
-import static com.viewsh.framework.common.util.collection.CollectionUtils.convertList;
-import static com.viewsh.module.iot.enums.ErrorCodeConstants.*;
-
-/**
- * IoT 设备 Service 实现类
- *
- * @author 芋道源码
- */
-@Service
-@Validated
-@Slf4j
-public class IotDeviceServiceImpl implements IotDeviceService {
-
- @Resource
- private IotDeviceMapper deviceMapper;
-
- @Resource
- @Lazy // 延迟加载，解决循环依赖
- private IotProductService productService;
- @Resource
- @Lazy // 延迟加载，解决循环依赖
- private IotDeviceGroupService deviceGroupService;
-
- @Override
- public Long createDevice(IotDeviceSaveReqVO createReqVO) {
- // 1.1 校验产品是否存在
- IotProductDO product = productService.getProduct(createReqVO.getProductId());
- if (product == null) {
- throw exception(PRODUCT_NOT_EXISTS);
- }
- // 1.2 统一校验
- validateCreateDeviceParam(product.getProductKey(), createReqVO.getDeviceName(),
- createReqVO.getGatewayId(), product);
- // 1.3 校验分组存在
- deviceGroupService.validateDeviceGroupExists(createReqVO.getGroupIds());
- // 1.4 校验设备序列号全局唯一
- validateSerialNumberUnique(createReqVO.getSerialNumber(), null);
-
- // 2. 插入到数据库
- IotDeviceDO device = BeanUtils.toBean(createReqVO, IotDeviceDO.class);
- initDevice(device, product);
- deviceMapper.insert(device);
- return device.getId();
- }
-
- private void validateCreateDeviceParam(String productKey, String deviceName,
- Long gatewayId, IotProductDO product) {
- // 校验设备名称在同一产品下是否唯一
- TenantUtils.executeIgnore(() -> {
- if (deviceMapper.selectByProductKeyAndDeviceName(productKey, deviceName) != null) {
- throw exception(DEVICE_NAME_EXISTS);
- }
- });
- // 校验父设备是否为合法网关
- if (IotProductDeviceTypeEnum.isGatewaySub(product.getDeviceType())
- && gatewayId != null) {
- validateGatewayDeviceExists(gatewayId);
- }
- }
-
- /**
- * 校验设备序列号全局唯一性
- *
- * @param serialNumber 设备序列号
- * @param excludeId 排除的设备编号（用于更新时排除自身）
- */
- private void validateSerialNumberUnique(String serialNumber, Long excludeId) {
- if (StrUtil.isBlank(serialNumber)) {
- return;
- }
- IotDeviceDO existDevice = deviceMapper.selectBySerialNumber(serialNumber);
- if (existDevice != null && ObjUtil.notEqual(existDevice.getId(), excludeId)) {
- throw exception(DEVICE_SERIAL_NUMBER_EXISTS);
- }
- }
-
- private void initDevice(IotDeviceDO device, IotProductDO product) {
- device.setProductId(product.getId()).setProductKey(product.getProductKey())
- .setDeviceType(product.getDeviceType());
- // 生成密钥
- device.setDeviceSecret(generateDeviceSecret());
- // 设置设备状态为未激活
- device.setState(IotDeviceStateEnum.INACTIVE.getState());
- }
-
- @Override
- public void updateDevice(IotDeviceSaveReqVO updateReqVO) {
- updateReqVO.setDeviceName(null).setProductId(null); // 不允许更新
- // 1.1 校验存在
- IotDeviceDO device = validateDeviceExists(updateReqVO.getId());
- // 1.2 校验父设备是否为合法网关
- if (IotProductDeviceTypeEnum.isGatewaySub(device.getDeviceType())
- && updateReqVO.getGatewayId() != null) {
- validateGatewayDeviceExists(updateReqVO.getGatewayId());
- }
- // 1.3 校验分组存在
- deviceGroupService.validateDeviceGroupExists(updateReqVO.getGroupIds());
- // 1.4 校验设备序列号全局唯一
- validateSerialNumberUnique(updateReqVO.getSerialNumber(), updateReqVO.getId());
-
- // 2. 更新到数据库
- IotDeviceDO updateObj = BeanUtils.toBean(updateReqVO, IotDeviceDO.class);
- deviceMapper.updateById(updateObj);
-
- // 3. 清空对应缓存
- deleteDeviceCache(device);
- }
-
- @Override
- @Transactional(rollbackFor = Exception.class)
- public void updateDeviceGroup(IotDeviceUpdateGroupReqVO updateReqVO) {
- // 1.1 校验设备存在
- List devices = deviceMapper.selectByIds(updateReqVO.getIds());
- if (CollUtil.isEmpty(devices)) {
- return;
- }
- // 1.2 校验分组存在
- deviceGroupService.validateDeviceGroupExists(updateReqVO.getGroupIds());
-
- // 3. 更新设备分组
- deviceMapper.updateBatch(convertList(devices, device -> new IotDeviceDO()
- .setId(device.getId()).setGroupIds(updateReqVO.getGroupIds())));
-
- // 4. 清空对应缓存
- deleteDeviceCache(devices);
- }
-
- @Override
- public void deleteDevice(Long id) {
- // 1.1 校验存在
- IotDeviceDO device = validateDeviceExists(id);
- // 1.2 如果是网关设备，检查是否有子设备
- if (device.getGatewayId() != null && deviceMapper.selectCountByGatewayId(id) > 0) {
- throw exception(DEVICE_HAS_CHILDREN);
- }
-
- // 2. 删除设备
- deviceMapper.deleteById(id);
-
- // 3. 清空对应缓存
- deleteDeviceCache(device);
- }
-
- @Override
- @Transactional(rollbackFor = Exception.class)
- public void deleteDeviceList(Collection ids) {
- // 1.1 校验存在
- if (CollUtil.isEmpty(ids)) {
- return;
- }
- List devices = deviceMapper.selectByIds(ids);
- if (CollUtil.isEmpty(devices)) {
- return;
- }
- // 1.2 校验网关设备是否存在
- for (IotDeviceDO device : devices) {
- if (device.getGatewayId() != null && deviceMapper.selectCountByGatewayId(device.getId()) > 0) {
- throw exception(DEVICE_HAS_CHILDREN);
- }
- }
-
- // 2. 删除设备
- deviceMapper.deleteByIds(ids);
-
- // 3. 清空对应缓存
- deleteDeviceCache(devices);
- }
-
- @Override
- public IotDeviceDO validateDeviceExists(Long id) {
- IotDeviceDO device = deviceMapper.selectById(id);
- if (device == null) {
- throw exception(DEVICE_NOT_EXISTS);
- }
- return device;
- }
-
- @Override
- public IotDeviceDO validateDeviceExistsFromCache(Long id) {
- IotDeviceDO device = getSelf().getDeviceFromCache(id);
- if (device == null) {
- throw exception(DEVICE_NOT_EXISTS);
- }
- return device;
- }
-
- /**
- * 校验网关设备是否存在
- *
- * @param id 设备 ID
- */
- private void validateGatewayDeviceExists(Long id) {
- IotDeviceDO device = deviceMapper.selectById(id);
- if (device == null) {
- throw exception(DEVICE_GATEWAY_NOT_EXISTS);
- }
- if (!IotProductDeviceTypeEnum.isGateway(device.getDeviceType())) {
- throw exception(DEVICE_NOT_GATEWAY);
- }
- }
-
- @Override
- public IotDeviceDO getDevice(Long id) {
- return deviceMapper.selectById(id);
- }
-
- @Override
- @Cacheable(value = RedisKeyConstants.DEVICE, key = "#id", unless = "#result == null")
- @TenantIgnore // 忽略租户信息
- public IotDeviceDO getDeviceFromCache(Long id) {
- return deviceMapper.selectById(id);
- }
-
- @Override
- @Cacheable(value = RedisKeyConstants.DEVICE, key = "#productKey + '_' + #deviceName", unless = "#result == null")
- @TenantIgnore // 忽略租户信息，跨租户 productKey + deviceName 是唯一的
- public IotDeviceDO getDeviceFromCache(String productKey, String deviceName) {
- return deviceMapper.selectByProductKeyAndDeviceName(productKey, deviceName);
- }
-
- @Override
- @Cacheable(value = RedisKeyConstants.DEVICE, key = "'deviceName_' + #deviceName", unless = "#result == null")
- @TenantIgnore // 忽略租户信息，用于 JT808 等协议，终端手机号应该是全局唯一的
- public IotDeviceDO getDeviceFromCacheByDeviceName(String deviceName) {
- return deviceMapper.selectByDeviceName(deviceName);
- }
-
- @Override
- public PageResult getDevicePage(IotDevicePageReqVO pageReqVO) {
- return deviceMapper.selectPage(pageReqVO);
- }
-
- @Override
- public List getDeviceListByCondition(@Nullable Integer deviceType, @Nullable Long productId) {
- return deviceMapper.selectListByCondition(deviceType, productId);
- }
-
- @Override
- public List getDeviceListByState(Integer state) {
- return deviceMapper.selectListByState(state);
- }
-
- @Override
- public List getDeviceListByProductId(Long productId) {
- return deviceMapper.selectListByProductId(productId);
- }
-
- @Override
- public void updateDeviceState(IotDeviceDO device, Integer state) {
- // 1. 更新状态和时间
- IotDeviceDO updateObj = new IotDeviceDO().setId(device.getId()).setState(state);
- if (device.getOnlineTime() == null
- && Objects.equals(state, IotDeviceStateEnum.ONLINE.getState())) {
- updateObj.setActiveTime(LocalDateTime.now());
- }
- if (Objects.equals(state, IotDeviceStateEnum.ONLINE.getState())) {
- updateObj.setOnlineTime(LocalDateTime.now());
- } else if (Objects.equals(state, IotDeviceStateEnum.OFFLINE.getState())) {
- updateObj.setOfflineTime(LocalDateTime.now());
- }
- deviceMapper.updateById(updateObj);
-
- // 2. 清空对应缓存
- deleteDeviceCache(device);
- }
-
- @Override
- public void updateDeviceState(Long id, Integer state) {
- // 校验存在
- IotDeviceDO device = validateDeviceExists(id);
- // 执行更新
- updateDeviceState(device, state);
- }
-
- @Override
- public Long getDeviceCountByProductId(Long productId) {
- return deviceMapper.selectCountByProductId(productId);
- }
-
- @Override
- public Long getDeviceCountByGroupId(Long groupId) {
- return deviceMapper.selectCountByGroupId(groupId);
- }
-
- /**
- * 生成 deviceSecret
- *
- * @return 生成的 deviceSecret
- */
- private String generateDeviceSecret() {
- return IdUtil.fastSimpleUUID();
- }
-
- @Override
- @Transactional(rollbackFor = Exception.class) // 添加事务，异常则回滚所有导入
- public IotDeviceImportRespVO importDevice(List importDevices, boolean updateSupport) {
- // 1. 参数校验
- if (CollUtil.isEmpty(importDevices)) {
- throw exception(DEVICE_IMPORT_LIST_IS_EMPTY);
- }
-
- // 2. 遍历，逐个创建 or 更新
- IotDeviceImportRespVO respVO = IotDeviceImportRespVO.builder().createDeviceNames(new ArrayList<>())
- .updateDeviceNames(new ArrayList<>()).failureDeviceNames(new LinkedHashMap<>()).build();
- importDevices.forEach(importDevice -> {
- try {
- // 2.1.1 校验字段是否符合要求
- try {
- ValidationUtils.validate(importDevice);
- } catch (ConstraintViolationException ex) {
- respVO.getFailureDeviceNames().put(importDevice.getDeviceName(), ex.getMessage());
- return;
- }
- // 2.1.2 校验产品是否存在
- IotProductDO product = productService.validateProductExists(importDevice.getProductKey());
- // 2.1.3 校验父设备是否存在
- Long gatewayId = null;
- if (StrUtil.isNotEmpty(importDevice.getParentDeviceName())) {
- IotDeviceDO gatewayDevice = deviceMapper.selectByDeviceName(importDevice.getParentDeviceName());
- if (gatewayDevice == null) {
- throw exception(DEVICE_GATEWAY_NOT_EXISTS);
- }
- if (!IotProductDeviceTypeEnum.isGateway(gatewayDevice.getDeviceType())) {
- throw exception(DEVICE_NOT_GATEWAY);
- }
- gatewayId = gatewayDevice.getId();
- }
- // 2.1.4 校验设备分组是否存在
- Set groupIds = new HashSet<>();
- if (StrUtil.isNotEmpty(importDevice.getGroupNames())) {
- String[] groupNames = importDevice.getGroupNames().split(",");
- for (String groupName : groupNames) {
- IotDeviceGroupDO group = deviceGroupService.getDeviceGroupByName(groupName);
- if (group == null) {
- throw exception(DEVICE_GROUP_NOT_EXISTS);
- }
- groupIds.add(group.getId());
- }
- }
-
- // 2.2.1 判断如果不存在，在进行插入
- IotDeviceDO existDevice = deviceMapper.selectByDeviceName(importDevice.getDeviceName());
- if (existDevice == null) {
- createDevice(new IotDeviceSaveReqVO()
- .setDeviceName(importDevice.getDeviceName())
- .setProductId(product.getId()).setGatewayId(gatewayId).setGroupIds(groupIds)
- .setLocationType(importDevice.getLocationType()));
- respVO.getCreateDeviceNames().add(importDevice.getDeviceName());
- return;
- }
- // 2.2.2 如果存在，判断是否允许更新
- if (updateSupport) {
- throw exception(DEVICE_KEY_EXISTS);
- }
- updateDevice(new IotDeviceSaveReqVO().setId(existDevice.getId())
- .setGatewayId(gatewayId).setGroupIds(groupIds).setLocationType(importDevice.getLocationType()));
- respVO.getUpdateDeviceNames().add(importDevice.getDeviceName());
- } catch (ServiceException ex) {
- respVO.getFailureDeviceNames().put(importDevice.getDeviceName(), ex.getMessage());
- }
- });
- return respVO;
- }
-
- @Override
- public IotDeviceAuthInfoRespVO getDeviceAuthInfo(Long id) {
- IotDeviceDO device = validateDeviceExists(id);
-
- // 获取生效的认证类型
- String effectiveAuthType = device.getAuthType();
- String targetSecret = device.getDeviceSecret();
-
- if (StrUtil.isEmpty(effectiveAuthType)) {
- IotProductDO product = productService.getProductFromCache(device.getProductId());
- if (product != null) {
- effectiveAuthType = product.getAuthType();
- // 如果是产品级一型一密，需要获取 ProductSecret
- if (IotAuthTypeEnum.PRODUCT_SECRET.getType().equals(effectiveAuthType)) {
- targetSecret = product.getProductSecret();
- }
- }
- } else if (IotAuthTypeEnum.PRODUCT_SECRET.getType().equals(effectiveAuthType)) {
- // 如果设备级强制设为一型一密（罕见但逻辑上兼容）
- IotProductDO product = productService.getProductFromCache(device.getProductId());
- if (product != null) {
- targetSecret = product.getProductSecret();
- }
- }
-
- // 使用 IotDeviceAuthUtils 生成认证信息
- IotDeviceAuthUtils.AuthInfo authInfo = IotDeviceAuthUtils.getAuthInfo(
- device.getProductKey(), device.getDeviceName(), targetSecret);
- return BeanUtils.toBean(authInfo, IotDeviceAuthInfoRespVO.class);
- }
-
- private void deleteDeviceCache(IotDeviceDO device) {
- // 保证 Spring AOP 触发
- getSelf().deleteDeviceCache0(device);
- }
-
- private void deleteDeviceCache(List devices) {
- devices.forEach(this::deleteDeviceCache);
- }
-
- @SuppressWarnings("unused")
- @Caching(evict = {
- @CacheEvict(value = RedisKeyConstants.DEVICE, key = "#device.id"),
- @CacheEvict(value = RedisKeyConstants.DEVICE, key = "#device.productKey + '_' + #device.deviceName")
- })
- public void deleteDeviceCache0(IotDeviceDO device) {
- }
-
- @Override
- public Long getDeviceCount(LocalDateTime createTime) {
- return deviceMapper.selectCountByCreateTime(createTime);
- }
-
- @Override
- public Map getDeviceCountMapByProductId() {
- return deviceMapper.selectDeviceCountMapByProductId();
- }
-
- @Override
- public Map getDeviceCountMapByState() {
- return deviceMapper.selectDeviceCountGroupByState();
- }
-
- @Override
- public List getDeviceListByProductKeyAndNames(String productKey, List deviceNames) {
- if (StrUtil.isBlank(productKey) || CollUtil.isEmpty(deviceNames)) {
- return Collections.emptyList();
- }
- return deviceMapper.selectByProductKeyAndDeviceNames(productKey, deviceNames);
- }
-
- @Override
- public boolean authDevice(IotDeviceAuthReqDTO authReqDTO) {
- // 1. 校验设备是否存在
- IotDeviceAuthUtils.DeviceInfo deviceInfo = IotDeviceAuthUtils.parseUsername(authReqDTO.getUsername());
- if (deviceInfo == null) {
- log.error("[authDevice][认证失败，username({}) 格式不正确]", authReqDTO.getUsername());
- return false;
- }
- String deviceName = deviceInfo.getDeviceName();
- String productKey = deviceInfo.getProductKey();
- IotDeviceDO device = getSelf().getDeviceFromCache(productKey, deviceName);
-
- // 1.1 处理动态注册 (Dynamic Registration)
- if (device == null) {
- // 获取产品信息
- IotProductDO product = productService.getProductByProductKey(productKey);
- if (product != null && IotAuthTypeEnum.DYNAMIC.getType().equals(product.getAuthType())) {
- // 自动创建设备
- IotDeviceSaveReqVO createReq = new IotDeviceSaveReqVO()
- .setProductId(product.getId())
- .setDeviceName(deviceName);
- try {
- // TODO: 考虑并发问题，这里简单处理
- getSelf().createDevice(createReq);
- // 重新获取设备
- device = getSelf().getDeviceFromCache(productKey, deviceName);
- log.info("[authDevice][动态注册设备成功: {}/{}]", productKey, deviceName);
- } catch (Exception e) {
- log.error("[authDevice][动态注册设备失败: {}/{}]", productKey, deviceName, e);
- return false;
- }
- }
- }
-
- if (device == null) {
- log.warn("[authDevice][设备({}/{}) 不存在]", productKey, deviceName);
- return false;
- }
-
- // 2. 校验密码
- // 2.1 获取生效的认证类型
- String effectiveAuthType = device.getAuthType();
- if (StrUtil.isEmpty(effectiveAuthType)) {
- IotProductDO product = productService.getProductFromCache(device.getProductId());
- if (product != null) {
- effectiveAuthType = product.getAuthType();
- }
- }
-
- // 2.2 根据类型校验
- String targetSecret = device.getDeviceSecret(); // 默认为一机一密
-
- if (IotAuthTypeEnum.PRODUCT_SECRET.getType().equals(effectiveAuthType)) {
- // 一型一密：使用 ProductSecret
- IotProductDO product = productService.getProductFromCache(device.getProductId());
- if (product == null || StrUtil.isEmpty(product.getProductSecret())) {
- log.error("[authDevice][一型一密认证失败，ProductSecret 为空: {}]", productKey);
- return false;
- }
- targetSecret = product.getProductSecret();
- } else if (IotAuthTypeEnum.NONE.getType().equals(effectiveAuthType)) {
- // 免鉴权：直接通过
- return true;
- } else if (IotAuthTypeEnum.DYNAMIC.getType().equals(effectiveAuthType)) {
- // 动态注册后，通常转为一机一密或一型一密，这里假设动态注册使用一型一密校验
- // 或者是免鉴权。具体策略视业务而定。这里暂定为一型一密。
- IotProductDO product = productService.getProductFromCache(device.getProductId());
- if (product != null) {
- targetSecret = product.getProductSecret();
- }
- }
-
- IotDeviceAuthUtils.AuthInfo authInfo = IotDeviceAuthUtils.getAuthInfo(productKey, deviceName, targetSecret);
- if (ObjUtil.notEqual(authInfo.getPassword(), authReqDTO.getPassword())) {
- log.error("[authDevice][设备({}/{}) 密码不正确]", productKey, deviceName);
- return false;
- }
- return true;
- }
-
- @Override
- public List validateDeviceListExists(Collection ids) {
- List devices = getDeviceList(ids);
- if (devices.size() != ids.size()) {
- throw exception(DEVICE_NOT_EXISTS);
- }
- return devices;
- }
-
- @Override
- public List getDeviceList(Collection ids) {
- if (CollUtil.isEmpty(ids)) {
- return Collections.emptyList();
- }
- return deviceMapper.selectByIds(ids);
- }
-
- @Override
- public void updateDeviceFirmware(Long deviceId, Long firmwareId) {
- // 1. 校验设备是否存在
- IotDeviceDO device = validateDeviceExists(deviceId);
-
- // 2. 更新设备固件版本
- IotDeviceDO updateObj = new IotDeviceDO().setId(deviceId).setFirmwareId(firmwareId);
- deviceMapper.updateById(updateObj);
-
- // 3. 清空对应缓存
- deleteDeviceCache(device);
- }
-
- private IotDeviceServiceImpl getSelf() {
- return SpringUtil.getBean(getClass());
- }
-
-}
+package com.viewsh.module.iot.service.device;
+
+import cn.hutool.core.collection.CollUtil;
+import cn.hutool.core.util.IdUtil;
+import cn.hutool.core.util.ObjUtil;
+import cn.hutool.core.util.StrUtil;
+import cn.hutool.extra.spring.SpringUtil;
+import com.viewsh.framework.common.exception.ServiceException;
+import com.viewsh.framework.common.pojo.PageResult;
+import com.viewsh.framework.common.util.object.BeanUtils;
+import com.viewsh.framework.common.util.validation.ValidationUtils;
+import com.viewsh.framework.tenant.core.aop.TenantIgnore;
+import com.viewsh.framework.tenant.core.util.TenantUtils;
+import com.viewsh.module.iot.core.integration.event.DeviceStatusChangedEvent;
+import com.viewsh.module.iot.controller.admin.device.vo.device.*;
+import com.viewsh.module.iot.core.biz.dto.IotDeviceAuthReqDTO;
+import com.viewsh.module.iot.core.enums.IotAuthTypeEnum;
+import com.viewsh.module.iot.core.enums.IotDeviceStateEnum;
+import com.viewsh.module.iot.core.integration.publisher.IntegrationEventPublisher;
+import com.viewsh.module.iot.core.util.IotDeviceAuthUtils;
+import com.viewsh.module.iot.dal.dataobject.device.IotDeviceDO;
+import com.viewsh.module.iot.dal.dataobject.device.IotDeviceGroupDO;
+import com.viewsh.module.iot.dal.dataobject.product.IotProductDO;
+import com.viewsh.module.iot.dal.mysql.device.IotDeviceMapper;
+import com.viewsh.module.iot.dal.redis.RedisKeyConstants;
+import com.viewsh.module.iot.enums.product.IotProductDeviceTypeEnum;
+import com.viewsh.module.iot.service.product.IotProductService;
+import jakarta.annotation.Resource;
+import jakarta.validation.ConstraintViolationException;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.cache.annotation.CacheEvict;
+import org.springframework.cache.annotation.Cacheable;
+import org.springframework.cache.annotation.Caching;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+import org.springframework.validation.annotation.Validated;
+
+import javax.annotation.Nullable;
+import java.time.LocalDateTime;
+import java.util.*;
+
+import static com.viewsh.framework.common.exception.util.ServiceExceptionUtil.exception;
+import static com.viewsh.framework.common.util.collection.CollectionUtils.convertList;
+import static com.viewsh.module.iot.enums.ErrorCodeConstants.*;
+
+/**
+ * IoT 设备 Service 实现类
+ *
+ * @author 芋道源码
+ */
+@Service
+@Validated
+@Slf4j
+public class IotDeviceServiceImpl implements IotDeviceService {
+
+ @Resource
+ private IotDeviceMapper deviceMapper;
+
+ @Resource
+ @Lazy // 延迟加载，解决循环依赖
+ private IotProductService productService;
+ @Resource
+ @Lazy // 延迟加载，解决循环依赖
+ private IotDeviceGroupService deviceGroupService;
+
+ @Resource
+ private IntegrationEventPublisher integrationEventPublisher;
+
+ @Override
+ public Long createDevice(IotDeviceSaveReqVO createReqVO) {
+ // 1.1 校验产品是否存在
+ IotProductDO product = productService.getProduct(createReqVO.getProductId());
+ if (product == null) {
+ throw exception(PRODUCT_NOT_EXISTS);
+ }
+ // 1.2 统一校验
+ validateCreateDeviceParam(product.getProductKey(), createReqVO.getDeviceName(),
+ createReqVO.getGatewayId(), product);
+ // 1.3 校验分组存在
+ deviceGroupService.validateDeviceGroupExists(createReqVO.getGroupIds());
+ // 1.4 校验设备序列号全局唯一
+ validateSerialNumberUnique(createReqVO.getSerialNumber(), null);
+
+ // 2. 插入到数据库
+ IotDeviceDO device = BeanUtils.toBean(createReqVO, IotDeviceDO.class);
+ initDevice(device, product);
+ deviceMapper.insert(device);
+ return device.getId();
+ }
+
+ private void validateCreateDeviceParam(String productKey, String deviceName,
+ Long gatewayId, IotProductDO product) {
+ // 校验设备名称在同一产品下是否唯一
+ TenantUtils.executeIgnore(() -> {
+ if (deviceMapper.selectByProductKeyAndDeviceName(productKey, deviceName) != null) {
+ throw exception(DEVICE_NAME_EXISTS);
+ }
+ });
+ // 校验父设备是否为合法网关
+ if (IotProductDeviceTypeEnum.isGatewaySub(product.getDeviceType())
+ && gatewayId != null) {
+ validateGatewayDeviceExists(gatewayId);
+ }
+ }
+
+ /**
+ * 校验设备序列号全局唯一性
+ *
+ * @param serialNumber 设备序列号
+ * @param excludeId 排除的设备编号（用于更新时排除自身）
+ */
+ private void validateSerialNumberUnique(String serialNumber, Long excludeId) {
+ if (StrUtil.isBlank(serialNumber)) {
+ return;
+ }
+ IotDeviceDO existDevice = deviceMapper.selectBySerialNumber(serialNumber);
+ if (existDevice != null && ObjUtil.notEqual(existDevice.getId(), excludeId)) {
+ throw exception(DEVICE_SERIAL_NUMBER_EXISTS);
+ }
+ }
+
+ private void initDevice(IotDeviceDO device, IotProductDO product) {
+ device.setProductId(product.getId()).setProductKey(product.getProductKey())
+ .setDeviceType(product.getDeviceType());
+ // 生成密钥
+ device.setDeviceSecret(generateDeviceSecret());
+ // 设置设备状态为未激活
+ device.setState(IotDeviceStateEnum.INACTIVE.getState());
+ }
+
+ @Override
+ public void updateDevice(IotDeviceSaveReqVO updateReqVO) {
+ updateReqVO.setDeviceName(null).setProductId(null); // 不允许更新
+ // 1.1 校验存在
+ IotDeviceDO device = validateDeviceExists(updateReqVO.getId());
+ // 1.2 校验父设备是否为合法网关
+ if (IotProductDeviceTypeEnum.isGatewaySub(device.getDeviceType())
+ && updateReqVO.getGatewayId() != null) {
+ validateGatewayDeviceExists(updateReqVO.getGatewayId());
+ }
+ // 1.3 校验分组存在
+ deviceGroupService.validateDeviceGroupExists(updateReqVO.getGroupIds());
+ // 1.4 校验设备序列号全局唯一
+ validateSerialNumberUnique(updateReqVO.getSerialNumber(), updateReqVO.getId());
+
+ // 2. 更新到数据库
+ IotDeviceDO updateObj = BeanUtils.toBean(updateReqVO, IotDeviceDO.class);
+ deviceMapper.updateById(updateObj);
+
+ // 3. 清空对应缓存
+ deleteDeviceCache(device);
+ }
+
+ @Override
+ @Transactional(rollbackFor = Exception.class)
+ public void updateDeviceGroup(IotDeviceUpdateGroupReqVO updateReqVO) {
+ // 1.1 校验设备存在
+ List devices = deviceMapper.selectByIds(updateReqVO.getIds());
+ if (CollUtil.isEmpty(devices)) {
+ return;
+ }
+ // 1.2 校验分组存在
+ deviceGroupService.validateDeviceGroupExists(updateReqVO.getGroupIds());
+
+ // 3. 更新设备分组
+ deviceMapper.updateBatch(convertList(devices, device -> new IotDeviceDO()
+ .setId(device.getId()).setGroupIds(updateReqVO.getGroupIds())));
+
+ // 4. 清空对应缓存
+ deleteDeviceCache(devices);
+ }
+
+ @Override
+ public void deleteDevice(Long id) {
+ // 1.1 校验存在
+ IotDeviceDO device = validateDeviceExists(id);
+ // 1.2 如果是网关设备，检查是否有子设备
+ if (device.getGatewayId() != null && deviceMapper.selectCountByGatewayId(id) > 0) {
+ throw exception(DEVICE_HAS_CHILDREN);
+ }
+
+ // 2. 删除设备
+ deviceMapper.deleteById(id);
+
+ // 3. 清空对应缓存
+ deleteDeviceCache(device);
+ }
+
+ @Override
+ @Transactional(rollbackFor = Exception.class)
+ public void deleteDeviceList(Collection ids) {
+ // 1.1 校验存在
+ if (CollUtil.isEmpty(ids)) {
+ return;
+ }
+ List devices = deviceMapper.selectByIds(ids);
+ if (CollUtil.isEmpty(devices)) {
+ return;
+ }
+ // 1.2 校验网关设备是否存在
+ for (IotDeviceDO device : devices) {
+ if (device.getGatewayId() != null && deviceMapper.selectCountByGatewayId(device.getId()) > 0) {
+ throw exception(DEVICE_HAS_CHILDREN);
+ }
+ }
+
+ // 2. 删除设备
+ deviceMapper.deleteByIds(ids);
+
+ // 3. 清空对应缓存
+ deleteDeviceCache(devices);
+ }
+
+ @Override
+ public IotDeviceDO validateDeviceExists(Long id) {
+ IotDeviceDO device = deviceMapper.selectById(id);
+ if (device == null) {
+ throw exception(DEVICE_NOT_EXISTS);
+ }
+ return device;
+ }
+
+ @Override
+ public IotDeviceDO validateDeviceExistsFromCache(Long id) {
+ IotDeviceDO device = getSelf().getDeviceFromCache(id);
+ if (device == null) {
+ throw exception(DEVICE_NOT_EXISTS);
+ }
+ return device;
+ }
+
+ /**
+ * 校验网关设备是否存在
+ *
+ * @param id 设备 ID
+ */
+ private void validateGatewayDeviceExists(Long id) {
+ IotDeviceDO device = deviceMapper.selectById(id);
+ if (device == null) {
+ throw exception(DEVICE_GATEWAY_NOT_EXISTS);
+ }
+ if (!IotProductDeviceTypeEnum.isGateway(device.getDeviceType())) {
+ throw exception(DEVICE_NOT_GATEWAY);
+ }
+ }
+
+ @Override
+ public IotDeviceDO getDevice(Long id) {
+ return deviceMapper.selectById(id);
+ }
+
+ @Override
+ @Cacheable(value = RedisKeyConstants.DEVICE, key = "#id", unless = "#result == null")
+ @TenantIgnore // 忽略租户信息
+ public IotDeviceDO getDeviceFromCache(Long id) {
+ return deviceMapper.selectById(id);
+ }
+
+ @Override
+ @Cacheable(value = RedisKeyConstants.DEVICE, key = "#productKey + '_' + #deviceName", unless = "#result == null")
+ @TenantIgnore // 忽略租户信息，跨租户 productKey + deviceName 是唯一的
+ public IotDeviceDO getDeviceFromCache(String productKey, String deviceName) {
+ return deviceMapper.selectByProductKeyAndDeviceName(productKey, deviceName);
+ }
+
+ @Override
+ @Cacheable(value = RedisKeyConstants.DEVICE, key = "'deviceName_' + #deviceName", unless = "#result == null")
+ @TenantIgnore // 忽略租户信息，用于 JT808 等协议，终端手机号应该是全局唯一的
+ public IotDeviceDO getDeviceFromCacheByDeviceName(String deviceName) {
+ return deviceMapper.selectByDeviceName(deviceName);
+ }
+
+ @Override
+ public PageResult getDevicePage(IotDevicePageReqVO pageReqVO) {
+ return deviceMapper.selectPage(pageReqVO);
+ }
+
+ @Override
+ public List getDeviceListByCondition(@Nullable Integer deviceType, @Nullable Long productId) {
+ return deviceMapper.selectListByCondition(deviceType, productId);
+ }
+
+ @Override
+ public List getDeviceListByState(Integer state) {
+ return deviceMapper.selectListByState(state);
+ }
+
+ @Override
+ public List getDeviceListByProductId(Long productId) {
+ return deviceMapper.selectListByProductId(productId);
+ }
+
+ @Override
+ public void updateDeviceState(IotDeviceDO device, Integer state) {
+ // 记录旧状态用于事件发布
+ Integer oldState = device.getState();
+
+ // 1. 更新状态和时间
+ IotDeviceDO updateObj = new IotDeviceDO().setId(device.getId()).setState(state);
+ if (device.getOnlineTime() == null
+ && Objects.equals(state, IotDeviceStateEnum.ONLINE.getState())) {
+ updateObj.setActiveTime(LocalDateTime.now());
+ }
+ if (Objects.equals(state, IotDeviceStateEnum.ONLINE.getState())) {
+ updateObj.setOnlineTime(LocalDateTime.now());
+ } else if (Objects.equals(state, IotDeviceStateEnum.OFFLINE.getState())) {
+ updateObj.setOfflineTime(LocalDateTime.now());
+ }
+ deviceMapper.updateById(updateObj);
+
+ // 2. 清空对应缓存
+ deleteDeviceCache(device);
+
+ // 3. 发布状态变更事件到跨模块事件总线
+ publishStatusChangedEvent(device, oldState, state);
+ }
+
+ /**
+ * 发布设备状态变更事件到 IntegrationEventBus
+ *
+ * @param device 设备信息
+ * @param oldState 旧状态
+ * @param newState 新状态
+ */
+ private void publishStatusChangedEvent(IotDeviceDO device, Integer oldState, Integer newState) {
+ // 只有状态真正发生变化时才发布事件
+ if (Objects.equals(oldState, newState)) {
+ return;
+ }
+
+ try {
+ // 获取产品信息
+ String productKey = "unknown";
+ IotProductDO product = productService.getProductFromCache(device.getProductId());
+ if (product != null) {
+ productKey = product.getProductKey();
+ }
+
+ // 确定状态变更原因
+ String reason = getStateChangeReason(oldState, newState);
+
+ DeviceStatusChangedEvent event = DeviceStatusChangedEvent.builder()
+ .deviceId(device.getId())
+ .deviceName(device.getDeviceName())
+ .productId(device.getProductId())
+ .productKey(productKey)
+ .tenantId(device.getTenantId())
+ .oldStatus(oldState)
+ .newStatus(newState)
+ .reason(reason)
+ .eventTime(LocalDateTime.now())
+ .build();
+
+ integrationEventPublisher.publishStatusChanged(event);
+ log.debug("[publishStatusChangedEvent] 跨模块状态变更事件已发布: eventId={}, deviceId={}, productKey={}, {} -> {}",
+ event.getEventId(), device.getId(), productKey, oldState, newState);
+ } catch (Exception e) {
+ log.error("[publishStatusChangedEvent] 跨模块状态变更事件发布失败: deviceId={}", device.getId(), e);
+ }
+ }
+
+ /**
+ * 根据状态变化获取变更原因
+ *
+ * @param oldState 旧状态
+ * @param newState 新状态
+ * @return 变更原因
+ */
+ private String getStateChangeReason(Integer oldState, Integer newState) {
+ if (Objects.equals(newState, IotDeviceStateEnum.ONLINE.getState())) {
+ return "设备上线";
+ } else if (Objects.equals(newState, IotDeviceStateEnum.OFFLINE.getState())) {
+ return "设备离线";
+ } else if (Objects.equals(newState, IotDeviceStateEnum.INACTIVE.getState())) {
+ return "设备停用";
+ }
+ return "状态变更";
+ }
+
+ @Override
+ public void updateDeviceState(Long id, Integer state) {
+ // 校验存在
+ IotDeviceDO device = validateDeviceExists(id);
+ // 执行更新
+ updateDeviceState(device, state);
+ }
+
+ @Override
+ public Long getDeviceCountByProductId(Long productId) {
+ return deviceMapper.selectCountByProductId(productId);
+ }
+
+ @Override
+ public Long getDeviceCountByGroupId(Long groupId) {
+ return deviceMapper.selectCountByGroupId(groupId);
+ }
+
+ /**
+ * 生成 deviceSecret
+ *
+ * @return 生成的 deviceSecret
+ */
+ private String generateDeviceSecret() {
+ return IdUtil.fastSimpleUUID();
+ }
+
+ @Override
+ @Transactional(rollbackFor = Exception.class) // 添加事务，异常则回滚所有导入
+ public IotDeviceImportRespVO importDevice(List importDevices, boolean updateSupport) {
+ // 1. 参数校验
+ if (CollUtil.isEmpty(importDevices)) {
+ throw exception(DEVICE_IMPORT_LIST_IS_EMPTY);
+ }
+
+ // 2. 遍历，逐个创建 or 更新
+ IotDeviceImportRespVO respVO = IotDeviceImportRespVO.builder().createDeviceNames(new ArrayList<>())
+ .updateDeviceNames(new ArrayList<>()).failureDeviceNames(new LinkedHashMap<>()).build();
+ importDevices.forEach(importDevice -> {
+ try {
+ // 2.1.1 校验字段是否符合要求
+ try {
+ ValidationUtils.validate(importDevice);
+ } catch (ConstraintViolationException ex) {
+ respVO.getFailureDeviceNames().put(importDevice.getDeviceName(), ex.getMessage());
+ return;
+ }
+ // 2.1.2 校验产品是否存在
+ IotProductDO product = productService.validateProductExists(importDevice.getProductKey());
+ // 2.1.3 校验父设备是否存在
+ Long gatewayId = null;
+ if (StrUtil.isNotEmpty(importDevice.getParentDeviceName())) {
+ IotDeviceDO gatewayDevice = deviceMapper.selectByDeviceName(importDevice.getParentDeviceName());
+ if (gatewayDevice == null) {
+ throw exception(DEVICE_GATEWAY_NOT_EXISTS);
+ }
+ if (!IotProductDeviceTypeEnum.isGateway(gatewayDevice.getDeviceType())) {
+ throw exception(DEVICE_NOT_GATEWAY);
+ }
+ gatewayId = gatewayDevice.getId();
+ }
+ // 2.1.4 校验设备分组是否存在
+ Set groupIds = new HashSet<>();
+ if (StrUtil.isNotEmpty(importDevice.getGroupNames())) {
+ String[] groupNames = importDevice.getGroupNames().split(",");
+ for (String groupName : groupNames) {
+ IotDeviceGroupDO group = deviceGroupService.getDeviceGroupByName(groupName);
+ if (group == null) {
+ throw exception(DEVICE_GROUP_NOT_EXISTS);
+ }
+ groupIds.add(group.getId());
+ }
+ }
+
+ // 2.2.1 判断如果不存在，在进行插入
+ IotDeviceDO existDevice = deviceMapper.selectByDeviceName(importDevice.getDeviceName());
+ if (existDevice == null) {
+ createDevice(new IotDeviceSaveReqVO()
+ .setDeviceName(importDevice.getDeviceName())
+ .setProductId(product.getId()).setGatewayId(gatewayId).setGroupIds(groupIds)
+ .setLocationType(importDevice.getLocationType()));
+ respVO.getCreateDeviceNames().add(importDevice.getDeviceName());
+ return;
+ }
+ // 2.2.2 如果存在，判断是否允许更新
+ if (updateSupport) {
+ throw exception(DEVICE_KEY_EXISTS);
+ }
+ updateDevice(new IotDeviceSaveReqVO().setId(existDevice.getId())
+ .setGatewayId(gatewayId).setGroupIds(groupIds).setLocationType(importDevice.getLocationType()));
+ respVO.getUpdateDeviceNames().add(importDevice.getDeviceName());
+ } catch (ServiceException ex) {
+ respVO.getFailureDeviceNames().put(importDevice.getDeviceName(), ex.getMessage());
+ }
+ });
+ return respVO;
+ }
+
+ @Override
+ public IotDeviceAuthInfoRespVO getDeviceAuthInfo(Long id) {
+ IotDeviceDO device = validateDeviceExists(id);
+
+ // 获取生效的认证类型
+ String effectiveAuthType = device.getAuthType();
+ String targetSecret = device.getDeviceSecret();
+
+ if (StrUtil.isEmpty(effectiveAuthType)) {
+ IotProductDO product = productService.getProductFromCache(device.getProductId());
+ if (product != null) {
+ effectiveAuthType = product.getAuthType();
+ // 如果是产品级一型一密，需要获取 ProductSecret
+ if (IotAuthTypeEnum.PRODUCT_SECRET.getType().equals(effectiveAuthType)) {
+ targetSecret = product.getProductSecret();
+ }
+ }
+ } else if (IotAuthTypeEnum.PRODUCT_SECRET.getType().equals(effectiveAuthType)) {
+ // 如果设备级强制设为一型一密（罕见但逻辑上兼容）
+ IotProductDO product = productService.getProductFromCache(device.getProductId());
+ if (product != null) {
+ targetSecret = product.getProductSecret();
+ }
+ }
+
+ // 使用 IotDeviceAuthUtils 生成认证信息
+ IotDeviceAuthUtils.AuthInfo authInfo = IotDeviceAuthUtils.getAuthInfo(
+ device.getProductKey(), device.getDeviceName(), targetSecret);
+ return BeanUtils.toBean(authInfo, IotDeviceAuthInfoRespVO.class);
+ }
+
+ private void deleteDeviceCache(IotDeviceDO device) {
+ // 保证 Spring AOP 触发
+ getSelf().deleteDeviceCache0(device);
+ }
+
+ private void deleteDeviceCache(List devices) {
+ devices.forEach(this::deleteDeviceCache);
+ }
+
+ @SuppressWarnings("unused")
+ @Caching(evict = {
+ @CacheEvict(value = RedisKeyConstants.DEVICE, key = "#device.id"),
+ @CacheEvict(value = RedisKeyConstants.DEVICE, key = "#device.productKey + '_' + #device.deviceName")
+ })
+ public void deleteDeviceCache0(IotDeviceDO device) {
+ }
+
+ @Override
+ public Long getDeviceCount(LocalDateTime createTime) {
+ return deviceMapper.selectCountByCreateTime(createTime);
+ }
+
+ @Override
+ public Map getDeviceCountMapByProductId() {
+ return deviceMapper.selectDeviceCountMapByProductId();
+ }
+
+ @Override
+ public Map getDeviceCountMapByState() {
+ return deviceMapper.selectDeviceCountGroupByState();
+ }
+
+ @Override
+ public List getDeviceListByProductKeyAndNames(String productKey, List deviceNames) {
+ if (StrUtil.isBlank(productKey) || CollUtil.isEmpty(deviceNames)) {
+ return Collections.emptyList();
+ }
+ return deviceMapper.selectByProductKeyAndDeviceNames(productKey, deviceNames);
+ }
+
+ @Override
+ public boolean authDevice(IotDeviceAuthReqDTO authReqDTO) {
+ // 1. 校验设备是否存在
+ IotDeviceAuthUtils.DeviceInfo deviceInfo = IotDeviceAuthUtils.parseUsername(authReqDTO.getUsername());
+ if (deviceInfo == null) {
+ log.error("[authDevice][认证失败，username({}) 格式不正确]", authReqDTO.getUsername());
+ return false;
+ }
+ String deviceName = deviceInfo.getDeviceName();
+ String productKey = deviceInfo.getProductKey();
+ IotDeviceDO device = getSelf().getDeviceFromCache(productKey, deviceName);
+
+ // 1.1 处理动态注册 (Dynamic Registration)
+ if (device == null) {
+ // 获取产品信息
+ IotProductDO product = productService.getProductByProductKey(productKey);
+ if (product != null && IotAuthTypeEnum.DYNAMIC.getType().equals(product.getAuthType())) {
+ // 自动创建设备
+ IotDeviceSaveReqVO createReq = new IotDeviceSaveReqVO()
+ .setProductId(product.getId())
+ .setDeviceName(deviceName);
+ try {
+ // TODO: 考虑并发问题，这里简单处理
+ getSelf().createDevice(createReq);
+ // 重新获取设备
+ device = getSelf().getDeviceFromCache(productKey, deviceName);
+ log.info("[authDevice][动态注册设备成功: {}/{}]", productKey, deviceName);
+ } catch (Exception e) {
+ log.error("[authDevice][动态注册设备失败: {}/{}]", productKey, deviceName, e);
+ return false;
+ }
+ }
+ }
+
+ if (device == null) {
+ log.warn("[authDevice][设备({}/{}) 不存在]", productKey, deviceName);
+ return false;
+ }
+
+ // 2. 校验密码
+ // 2.1 获取生效的认证类型
+ String effectiveAuthType = device.getAuthType();
+ if (StrUtil.isEmpty(effectiveAuthType)) {
+ IotProductDO product = productService.getProductFromCache(device.getProductId());
+ if (product != null) {
+ effectiveAuthType = product.getAuthType();
+ }
+ }
+
+ // 2.2 根据类型校验
+ String targetSecret = device.getDeviceSecret(); // 默认为一机一密
+
+ if (IotAuthTypeEnum.PRODUCT_SECRET.getType().equals(effectiveAuthType)) {
+ // 一型一密：使用 ProductSecret
+ IotProductDO product = productService.getProductFromCache(device.getProductId());
+ if (product == null || StrUtil.isEmpty(product.getProductSecret())) {
+ log.error("[authDevice][一型一密认证失败，ProductSecret 为空: {}]", productKey);
+ return false;
+ }
+ targetSecret = product.getProductSecret();
+ } else if (IotAuthTypeEnum.NONE.getType().equals(effectiveAuthType)) {
+ // 免鉴权：直接通过
+ return true;
+ } else if (IotAuthTypeEnum.DYNAMIC.getType().equals(effectiveAuthType)) {
+ // 动态注册后，通常转为一机一密或一型一密，这里假设动态注册使用一型一密校验
+ // 或者是免鉴权。具体策略视业务而定。这里暂定为一型一密。
+ IotProductDO product = productService.getProductFromCache(device.getProductId());
+ if (product != null) {
+ targetSecret = product.getProductSecret();
+ }
+ }
+
+ IotDeviceAuthUtils.AuthInfo authInfo = IotDeviceAuthUtils.getAuthInfo(productKey, deviceName, targetSecret);
+ if (ObjUtil.notEqual(authInfo.getPassword(), authReqDTO.getPassword())) {
+ log.error("[authDevice][设备({}/{}) 密码不正确]", productKey, deviceName);
+ return false;
+ }
+ return true;
+ }
+
+ @Override
+ public List validateDeviceListExists(Collection ids) {
+ List devices = getDeviceList(ids);
+ if (devices.size() != ids.size()) {
+ throw exception(DEVICE_NOT_EXISTS);
+ }
+ return devices;
+ }
+
+ @Override
+ public List getDeviceList(Collection ids) {
+ if (CollUtil.isEmpty(ids)) {
+ return Collections.emptyList();
+ }
+ return deviceMapper.selectByIds(ids);
+ }
+
+ @Override
+ public void updateDeviceFirmware(Long deviceId, Long firmwareId) {
+ // 1. 校验设备是否存在
+ IotDeviceDO device = validateDeviceExists(deviceId);
+
+ // 2. 更新设备固件版本
+ IotDeviceDO updateObj = new IotDeviceDO().setId(deviceId).setFirmwareId(firmwareId);
+ deviceMapper.updateById(updateObj);
+
+ // 3. 清空对应缓存
+ deleteDeviceCache(device);
+ }
+
+ private IotDeviceServiceImpl getSelf() {
+ return SpringUtil.getBean(getClass());
+ }
+
+}
diff --git a/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/service/device/property/IotDevicePropertyServiceImpl.java b/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/service/device/property/IotDevicePropertyServiceImpl.java
index ef6c4ef..b9b02e1 100644
--- a/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/service/device/property/IotDevicePropertyServiceImpl.java
+++ b/viewsh-module-iot/viewsh-module-iot-server/src/main/java/com/viewsh/module/iot/service/device/property/IotDevicePropertyServiceImpl.java
@@ -1,209 +1,274 @@
-package com.viewsh.module.iot.service.device.property;
-
-import cn.hutool.core.collection.CollUtil;
-import cn.hutool.core.date.LocalDateTimeUtil;
-import cn.hutool.core.map.MapUtil;
-import cn.hutool.core.util.StrUtil;
-import com.viewsh.framework.common.util.json.JsonUtils;
-import com.viewsh.framework.common.util.object.ObjectUtils;
-import com.viewsh.module.iot.controller.admin.device.vo.property.IotDevicePropertyHistoryListReqVO;
-import com.viewsh.module.iot.controller.admin.device.vo.property.IotDevicePropertyRespVO;
-import com.viewsh.module.iot.core.mq.message.IotDeviceMessage;
-import com.viewsh.module.iot.dal.dataobject.device.IotDeviceDO;
-import com.viewsh.module.iot.dal.dataobject.device.IotDevicePropertyDO;
-import com.viewsh.module.iot.dal.dataobject.product.IotProductDO;
-import com.viewsh.module.iot.dal.dataobject.thingmodel.IotThingModelDO;
-import com.viewsh.module.iot.dal.dataobject.thingmodel.model.dataType.ThingModelDateOrTextDataSpecs;
-import com.viewsh.module.iot.dal.redis.device.DevicePropertyRedisDAO;
-import com.viewsh.module.iot.dal.redis.device.DeviceReportTimeRedisDAO;
-import com.viewsh.module.iot.dal.redis.device.DeviceServerIdRedisDAO;
-import com.viewsh.module.iot.dal.tdengine.IotDevicePropertyMapper;
-import com.viewsh.module.iot.enums.thingmodel.IotDataSpecsDataTypeEnum;
-import com.viewsh.module.iot.enums.thingmodel.IotThingModelTypeEnum;
-import com.viewsh.module.iot.framework.tdengine.core.TDengineTableField;
-import com.viewsh.module.iot.service.product.IotProductService;
-import com.viewsh.module.iot.service.thingmodel.IotThingModelService;
-import jakarta.annotation.Resource;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.context.annotation.Lazy;
-import org.springframework.scheduling.annotation.Async;
-import org.springframework.stereotype.Service;
-
-import java.time.LocalDateTime;
-import java.util.*;
-
-import static com.viewsh.framework.common.util.collection.CollectionUtils.*;
-
-/**
- * IoT 设备【属性】数据 Service 实现类
- *
- * @author 芋道源码
- */
-@Service
-@Slf4j
-public class IotDevicePropertyServiceImpl implements IotDevicePropertyService {
-
- /**
- * 物模型的数据类型，与 TDengine 数据类型的映射关系
- *
- * @see TDEngine 数据类型
- */
- private static final Map TYPE_MAPPING = MapUtil.builder()
- .put(IotDataSpecsDataTypeEnum.INT.getDataType(), TDengineTableField.TYPE_INT)
- .put(IotDataSpecsDataTypeEnum.FLOAT.getDataType(), TDengineTableField.TYPE_FLOAT)
- .put(IotDataSpecsDataTypeEnum.DOUBLE.getDataType(), TDengineTableField.TYPE_DOUBLE)
- .put(IotDataSpecsDataTypeEnum.ENUM.getDataType(), TDengineTableField.TYPE_TINYINT)
- .put(IotDataSpecsDataTypeEnum.BOOL.getDataType(), TDengineTableField.TYPE_TINYINT)
- .put(IotDataSpecsDataTypeEnum.TEXT.getDataType(), TDengineTableField.TYPE_VARCHAR)
- .put(IotDataSpecsDataTypeEnum.DATE.getDataType(), TDengineTableField.TYPE_TIMESTAMP)
- .put(IotDataSpecsDataTypeEnum.STRUCT.getDataType(), TDengineTableField.TYPE_VARCHAR)
- .put(IotDataSpecsDataTypeEnum.ARRAY.getDataType(), TDengineTableField.TYPE_VARCHAR)
- .build();
-
- @Resource
- private IotThingModelService thingModelService;
- @Resource
- @Lazy // 延迟加载，解决循环依赖
- private IotProductService productService;
-
- @Resource
- private DevicePropertyRedisDAO deviceDataRedisDAO;
- @Resource
- private DeviceReportTimeRedisDAO deviceReportTimeRedisDAO;
- @Resource
- private DeviceServerIdRedisDAO deviceServerIdRedisDAO;
-
- @Resource
- private IotDevicePropertyMapper devicePropertyMapper;
-
- // ========== 设备属性相关操作 ==========
-
- @Override
- public void defineDevicePropertyData(Long productId) {
- // 1.1 查询产品和物模型
- IotProductDO product = productService.validateProductExists(productId);
- List thingModels = filterList(thingModelService.getThingModelListByProductId(productId),
- thingModel -> IotThingModelTypeEnum.PROPERTY.getType().equals(thingModel.getType()));
- // 1.2 解析 DB 里的字段
- List oldFields = new ArrayList<>();
- try {
- oldFields.addAll(devicePropertyMapper.getProductPropertySTableFieldList(product.getId()));
- } catch (Exception e) {
- if (!e.getMessage().contains("Table does not exist")) {
- throw e;
- }
- }
-
- // 2.1 情况一：如果是新增的时候，需要创建表
- List newFields = buildTableFieldList(thingModels);
- if (CollUtil.isEmpty(oldFields)) {
- if (CollUtil.isEmpty(newFields)) {
- log.info("[defineDevicePropertyData][productId({}) 没有需要定义的属性]", productId);
- return;
- }
- devicePropertyMapper.createProductPropertySTable(product.getId(), newFields);
- return;
- }
- // 2.2 情况二：如果是修改的时候，需要更新表
- devicePropertyMapper.alterProductPropertySTable(product.getId(), oldFields, newFields);
- }
-
- private List buildTableFieldList(List thingModels) {
- return convertList(thingModels, thingModel -> {
- TDengineTableField field = new TDengineTableField(
- StrUtil.toUnderlineCase(thingModel.getIdentifier()), // TDengine 字段默认都是小写
- TYPE_MAPPING.get(thingModel.getProperty().getDataType()));
- String dataType = thingModel.getProperty().getDataType();
- if (Objects.equals(dataType, IotDataSpecsDataTypeEnum.TEXT.getDataType())) {
- field.setLength(((ThingModelDateOrTextDataSpecs) thingModel.getProperty().getDataSpecs()).getLength());
- } else if (ObjectUtils.equalsAny(dataType, IotDataSpecsDataTypeEnum.STRUCT.getDataType(),
- IotDataSpecsDataTypeEnum.ARRAY.getDataType())) {
- field.setLength(TDengineTableField.LENGTH_VARCHAR);
- }
- return field;
- });
- }
-
- @Override
- public void saveDeviceProperty(IotDeviceDO device, IotDeviceMessage message) {
- if (!(message.getParams() instanceof Map)) {
- log.error("[saveDeviceProperty][消息内容({}) 的 data 类型不正确]", message);
- return;
- }
-
- // 1. 根据物模型，拼接合法的属性
- // TODO @芋艿：【待定 004】赋能后，属性到底以 thingModel 为准（ik），还是 db 的表结构为准（tl）？
- List thingModels = thingModelService.getThingModelListByProductIdFromCache(device.getProductId());
- Map properties = new HashMap<>();
- ((Map) message.getParams()).forEach((key, value) -> {
- IotThingModelDO thingModel = CollUtil.findOne(thingModels, o -> o.getIdentifier().equals(key));
- if (thingModel == null || thingModel.getProperty() == null) {
- log.error("[saveDeviceProperty][消息({}) 的属性({}) 不存在]", message, key);
- return;
- }
- if (ObjectUtils.equalsAny(thingModel.getProperty().getDataType(),
- IotDataSpecsDataTypeEnum.STRUCT.getDataType(), IotDataSpecsDataTypeEnum.ARRAY.getDataType())) {
- // 特殊：STRUCT 和 ARRAY 类型，在 TDengine 里，有没对应数据类型，只能通过 JSON 来存储
- properties.put((String) key, JsonUtils.toJsonString(value));
- } else {
- properties.put((String) key, value);
- }
- });
- if (CollUtil.isEmpty(properties)) {
- log.error("[saveDeviceProperty][消息({}) 没有合法的属性]", message);
- return;
- }
-
- // 2.1 保存设备属性【数据】
- devicePropertyMapper.insert(device, properties, LocalDateTimeUtil.toEpochMilli(message.getReportTime()));
-
- // 2.2 保存设备属性【日志】
- Map properties2 = convertMap(properties.entrySet(), Map.Entry::getKey, entry ->
- IotDevicePropertyDO.builder().value(entry.getValue()).updateTime(message.getReportTime()).build());
- deviceDataRedisDAO.putAll(device.getId(), properties2);
- }
-
- @Override
- public Map getLatestDeviceProperties(Long deviceId) {
- return deviceDataRedisDAO.get(deviceId);
- }
-
- @Override
- public List getHistoryDevicePropertyList(IotDevicePropertyHistoryListReqVO listReqVO) {
- try {
- return devicePropertyMapper.selectListByHistory(listReqVO);
- } catch (Exception exception) {
- if (exception.getMessage().contains("Table does not exist")) {
- return Collections.emptyList();
- }
- throw exception;
- }
- }
-
- // ========== 设备时间相关操作 ==========
-
- @Override
- public Set getDeviceIdListByReportTime(LocalDateTime maxReportTime) {
- return deviceReportTimeRedisDAO.range(maxReportTime);
- }
-
- @Override
- @Async
- public void updateDeviceReportTimeAsync(Long id, LocalDateTime reportTime) {
- deviceReportTimeRedisDAO.update(id, reportTime);
- }
-
- @Override
- public void updateDeviceServerIdAsync(Long id, String serverId) {
- if (StrUtil.isEmpty(serverId)) {
- return;
- }
- deviceServerIdRedisDAO.update(id, serverId);
- }
-
- @Override
- public String getDeviceServerId(Long id) {
- return deviceServerIdRedisDAO.get(id);
- }
-
+package com.viewsh.module.iot.service.device.property;
+
+import cn.hutool.core.collection.CollUtil;
+import cn.hutool.core.date.LocalDateTimeUtil;
+import cn.hutool.core.map.MapUtil;
+import cn.hutool.core.util.StrUtil;
+import com.viewsh.framework.common.util.json.JsonUtils;
+import com.viewsh.framework.common.util.object.ObjectUtils;
+import com.viewsh.framework.mq.redis.core.RedisMQTemplate;
+import com.viewsh.module.iot.core.integration.event.DevicePropertyChangedEvent;
+import com.viewsh.module.iot.controller.admin.device.vo.property.IotDevicePropertyHistoryListReqVO;
+import com.viewsh.module.iot.controller.admin.device.vo.property.IotDevicePropertyRespVO;
+import com.viewsh.module.iot.core.integration.publisher.IntegrationEventPublisher;
+import com.viewsh.module.iot.core.mq.message.IotDeviceMessage;
+import com.viewsh.module.iot.dal.dataobject.device.IotDeviceDO;
+import com.viewsh.module.iot.dal.dataobject.device.IotDevicePropertyDO;
+import com.viewsh.module.iot.dal.dataobject.product.IotProductDO;
+import com.viewsh.module.iot.dal.dataobject.thingmodel.IotThingModelDO;
+import com.viewsh.module.iot.dal.dataobject.thingmodel.model.dataType.ThingModelDateOrTextDataSpecs;
+import com.viewsh.module.iot.dal.redis.device.DevicePropertyRedisDAO;
+import com.viewsh.module.iot.dal.redis.device.DeviceReportTimeRedisDAO;
+import com.viewsh.module.iot.dal.redis.device.DeviceServerIdRedisDAO;
+import com.viewsh.module.iot.dal.tdengine.IotDevicePropertyMapper;
+import com.viewsh.module.iot.enums.thingmodel.IotDataSpecsDataTypeEnum;
+import com.viewsh.module.iot.enums.thingmodel.IotThingModelTypeEnum;
+import com.viewsh.module.iot.framework.tdengine.core.TDengineTableField;
+import com.viewsh.module.iot.service.product.IotProductService;
+import com.viewsh.module.iot.service.thingmodel.IotThingModelService;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.stereotype.Service;
+
+import java.time.LocalDateTime;
+import java.util.*;
+
+import static com.viewsh.framework.common.util.collection.CollectionUtils.*;
+
+/**
+ * IoT 设备【属性】数据 Service 实现类
+ *
+ * @author 芋道源码
+ */
+@Service
+@Slf4j
+public class IotDevicePropertyServiceImpl implements IotDevicePropertyService {
+
+ /**
+ * 物模型的数据类型，与 TDengine 数据类型的映射关系
+ *
+ * @see TDEngine 数据类型
+ */
+ private static final Map TYPE_MAPPING = MapUtil.builder()
+ .put(IotDataSpecsDataTypeEnum.INT.getDataType(), TDengineTableField.TYPE_INT)
+ .put(IotDataSpecsDataTypeEnum.FLOAT.getDataType(), TDengineTableField.TYPE_FLOAT)
+ .put(IotDataSpecsDataTypeEnum.DOUBLE.getDataType(), TDengineTableField.TYPE_DOUBLE)
+ .put(IotDataSpecsDataTypeEnum.ENUM.getDataType(), TDengineTableField.TYPE_TINYINT)
+ .put(IotDataSpecsDataTypeEnum.BOOL.getDataType(), TDengineTableField.TYPE_TINYINT)
+ .put(IotDataSpecsDataTypeEnum.TEXT.getDataType(), TDengineTableField.TYPE_VARCHAR)
+ .put(IotDataSpecsDataTypeEnum.DATE.getDataType(), TDengineTableField.TYPE_TIMESTAMP)
+ .put(IotDataSpecsDataTypeEnum.STRUCT.getDataType(), TDengineTableField.TYPE_VARCHAR)
+ .put(IotDataSpecsDataTypeEnum.ARRAY.getDataType(), TDengineTableField.TYPE_VARCHAR)
+ .build();
+
+ @Resource
+ private IotThingModelService thingModelService;
+ @Resource
+ @Lazy // 延迟加载，解决循环依赖
+ private IotProductService productService;
+
+ @Resource
+ private DevicePropertyRedisDAO deviceDataRedisDAO;
+ @Resource
+ private DeviceReportTimeRedisDAO deviceReportTimeRedisDAO;
+ @Resource
+ private DeviceServerIdRedisDAO deviceServerIdRedisDAO;
+
+ @Resource
+ private IotDevicePropertyMapper devicePropertyMapper;
+
+ @Resource
+ private RedisMQTemplate redisMQTemplate;
+
+ @Resource
+ private IntegrationEventPublisher integrationEventPublisher;
+
+ // ========== 设备属性相关操作 ==========
+
+ @Override
+ public void defineDevicePropertyData(Long productId) {
+ // 1.1 查询产品和物模型
+ IotProductDO product = productService.validateProductExists(productId);
+ List thingModels = filterList(thingModelService.getThingModelListByProductId(productId),
+ thingModel -> IotThingModelTypeEnum.PROPERTY.getType().equals(thingModel.getType()));
+ // 1.2 解析 DB 里的字段
+ List oldFields = new ArrayList<>();
+ try {
+ oldFields.addAll(devicePropertyMapper.getProductPropertySTableFieldList(product.getId()));
+ } catch (Exception e) {
+ if (!e.getMessage().contains("Table does not exist")) {
+ throw e;
+ }
+ }
+
+ // 2.1 情况一：如果是新增的时候，需要创建表
+ List newFields = buildTableFieldList(thingModels);
+ if (CollUtil.isEmpty(oldFields)) {
+ if (CollUtil.isEmpty(newFields)) {
+ log.info("[defineDevicePropertyData][productId({}) 没有需要定义的属性]", productId);
+ return;
+ }
+ devicePropertyMapper.createProductPropertySTable(product.getId(), newFields);
+ return;
+ }
+ // 2.2 情况二：如果是修改的时候，需要更新表
+ devicePropertyMapper.alterProductPropertySTable(product.getId(), oldFields, newFields);
+ }
+
+ private List buildTableFieldList(List thingModels) {
+ return convertList(thingModels, thingModel -> {
+ TDengineTableField field = new TDengineTableField(
+ StrUtil.toUnderlineCase(thingModel.getIdentifier()), // TDengine 字段默认都是小写
+ TYPE_MAPPING.get(thingModel.getProperty().getDataType()));
+ String dataType = thingModel.getProperty().getDataType();
+ if (Objects.equals(dataType, IotDataSpecsDataTypeEnum.TEXT.getDataType())) {
+ field.setLength(((ThingModelDateOrTextDataSpecs) thingModel.getProperty().getDataSpecs()).getLength());
+ } else if (ObjectUtils.equalsAny(dataType, IotDataSpecsDataTypeEnum.STRUCT.getDataType(),
+ IotDataSpecsDataTypeEnum.ARRAY.getDataType())) {
+ field.setLength(TDengineTableField.LENGTH_VARCHAR);
+ }
+ return field;
+ });
+ }
+
+ @Override
+ public void saveDeviceProperty(IotDeviceDO device, IotDeviceMessage message) {
+ if (!(message.getParams() instanceof Map)) {
+ log.error("[saveDeviceProperty][消息内容({}) 的 data 类型不正确]", message);
+ return;
+ }
+
+ // 1. 根据物模型，拼接合法的属性
+ // TODO @芋艿：【待定 004】赋能后，属性到底以 thingModel 为准（ik），还是 db 的表结构为准（tl）？
+ List thingModels = thingModelService.getThingModelListByProductIdFromCache(device.getProductId());
+ Map properties = new HashMap<>();
+ ((Map) message.getParams()).forEach((key, value) -> {
+ IotThingModelDO thingModel = CollUtil.findOne(thingModels, o -> o.getIdentifier().equals(key));
+ if (thingModel == null || thingModel.getProperty() == null) {
+ log.error("[saveDeviceProperty][消息({}) 的属性({}) 不存在]", message, key);
+ return;
+ }
+ if (ObjectUtils.equalsAny(thingModel.getProperty().getDataType(),
+ IotDataSpecsDataTypeEnum.STRUCT.getDataType(), IotDataSpecsDataTypeEnum.ARRAY.getDataType())) {
+ // 特殊：STRUCT 和 ARRAY 类型，在 TDengine 里，有没对应数据类型，只能通过 JSON 来存储
+ properties.put((String) key, JsonUtils.toJsonString(value));
+ } else {
+ properties.put((String) key, value);
+ }
+ });
+ if (CollUtil.isEmpty(properties)) {
+ log.error("[saveDeviceProperty][消息({}) 没有合法的属性]", message);
+ return;
+ }
+
+ // 2.1 保存设备属性【数据】
+ devicePropertyMapper.insert(device, properties, LocalDateTimeUtil.toEpochMilli(message.getReportTime()));
+
+ // 2.2 保存设备属性【日志】
+ Map properties2 = convertMap(properties.entrySet(), Map.Entry::getKey, entry ->
+ IotDevicePropertyDO.builder().value(entry.getValue()).updateTime(message.getReportTime()).build());
+ deviceDataRedisDAO.putAll(device.getId(), properties2);
+
+ // 2.3 发布属性消息到 Redis Stream（供其他模块如 Ops 订阅）
+ publishPropertyMessage(device, properties, message.getReportTime());
+ }
+
+ /**
+ * 发布设备属性消息
+ *
+ * 同时发布到 Redis Stream（旧方式，保持兼容）和 RocketMQ IntegrationEventBus（新方式）
+ *
+ * @param device 设备信息
+ * @param properties 属性数据
+ * @param reportTime 上报时间
+ */
+ private void publishPropertyMessage(IotDeviceDO device, Map properties, LocalDateTime reportTime) {
+ try {
+ // 获取产品信息
+ IotProductDO product = productService.getProductFromCache(device.getProductId());
+ String productKey = product != null ? product.getProductKey() : "unknown";
+
+ // 发布到 IntegrationEventBus（RocketMQ）
+ publishToIntegrationEventBus(device, productKey, properties, reportTime);
+
+ } catch (Exception e) {
+ log.error("[publishPropertyMessage] 属性消息发布失败: deviceId={}", device.getId(), e);
+ }
+ }
+
+ /**
+ * 发布跨模块属性变更事件到 RocketMQ
+ *
+ * @param device 设备信息
+ * @param productKey 产品标识符（用作 RocketMQ Tag）
+ * @param properties 属性数据
+ * @param reportTime 上报时间
+ */
+ private void publishToIntegrationEventBus(IotDeviceDO device, String productKey,
+ Map properties, LocalDateTime reportTime) {
+ try {
+ DevicePropertyChangedEvent event = DevicePropertyChangedEvent.builder()
+ .deviceId(device.getId())
+ .deviceName(device.getDeviceName())
+ .productId(device.getProductId())
+ .productKey(productKey)
+ .tenantId(device.getTenantId())
+ .properties(properties)
+ .changedIdentifiers(properties.keySet())
+ .eventTime(reportTime)
+ .build();
+
+ integrationEventPublisher.publishPropertyChanged(event);
+ log.debug("[publishToIntegrationEventBus] 跨模块属性变更事件已发布: eventId={}, deviceId={}, productKey={}, properties={}",
+ event.getEventId(), device.getId(), productKey, properties.keySet());
+ } catch (Exception e) {
+ log.error("[publishToIntegrationEventBus] 跨模块属性变更事件发布失败: deviceId={}", device.getId(), e);
+ }
+ }
+
+ @Override
+ public Map getLatestDeviceProperties(Long deviceId) {
+ return deviceDataRedisDAO.get(deviceId);
+ }
+
+ @Override
+ public List getHistoryDevicePropertyList(IotDevicePropertyHistoryListReqVO listReqVO) {
+ try {
+ return devicePropertyMapper.selectListByHistory(listReqVO);
+ } catch (Exception exception) {
+ if (exception.getMessage().contains("Table does not exist")) {
+ return Collections.emptyList();
+ }
+ throw exception;
+ }
+ }
+
+ // ========== 设备时间相关操作 ==========
+
+ @Override
+ public Set getDeviceIdListByReportTime(LocalDateTime maxReportTime) {
+ return deviceReportTimeRedisDAO.range(maxReportTime);
+ }
+
+ @Override
+ @Async
+ public void updateDeviceReportTimeAsync(Long id, LocalDateTime reportTime) {
+ deviceReportTimeRedisDAO.update(id, reportTime);
+ }
+
+ @Override
+ public void updateDeviceServerIdAsync(Long id, String serverId) {
+ if (StrUtil.isEmpty(serverId)) {
+ return;
+ }
+ deviceServerIdRedisDAO.update(id, serverId);
+ }
+
+ @Override
+ public String getDeviceServerId(Long id) {
+ return deviceServerIdRedisDAO.get(id);
+ }
+
}
\ No newline at end of file
diff --git a/viewsh-module-iot/viewsh-module-iot-server/src/main/resources/application.yaml b/viewsh-module-iot/viewsh-module-iot-server/src/main/resources/application.yaml
index f875048..c8ac493 100644
--- a/viewsh-module-iot/viewsh-module-iot-server/src/main/resources/application.yaml
+++ b/viewsh-module-iot/viewsh-module-iot-server/src/main/resources/application.yaml
@@ -1,157 +1,163 @@
-spring:
- application:
- name: iot-server
-
- profiles:
- active: local
-
- main:
- allow-circular-references: true # 允许循环依赖，因为项目是三层架构，无法避免这个情况。
- allow-bean-definition-overriding: true # 允许 Bean 覆盖，例如说 Feign 等会存在重复定义的服务
-
- config:
- import:
- - optional:classpath:application-${spring.profiles.active}.yaml # 加载【本地】配置
- - optional:nacos:${spring.application.name}-${spring.profiles.active}.yaml # 加载【Nacos】的配置
-
- # Servlet 配置
- servlet:
- # 文件上传相关配置项
- multipart:
- max-file-size: 16MB # 单个文件大小
- max-request-size: 32MB # 设置总上传的文件大小
-
- # Jackson 配置项
- jackson:
- serialization:
- write-dates-as-timestamps: true # 设置 LocalDateTime 的格式，使用时间戳
- write-date-timestamps-as-nanoseconds: false # 设置不使用 nanoseconds 的格式。例如说 1611460870.401，而是直接 1611460870401
- write-durations-as-timestamps: true # 设置 Duration 的格式，使用时间戳
- fail-on-empty-beans: false # 允许序列化无属性的 Bean
-
- # Cache 配置项
- cache:
- type: REDIS
- redis:
- time-to-live: 1h # 设置过期时间为 1 小时
-
-server:
- port: 48091
-
-logging:
- file:
- name: ${user.home}/logs/${spring.application.name}.log # 日志文件名，全路径
-
---- #################### 接口文档配置 ####################
-
-springdoc:
- api-docs:
- enabled: true # 1. 是否开启 Swagger 接文档的元数据
- path: /v3/api-docs
- swagger-ui:
- enabled: true # 2.1 是否开启 Swagger 文档的官方 UI 界面
- path: /swagger-ui
- default-flat-param-object: true # 参见 https://doc.xiaominfo.com/docs/faq/v4/knife4j-parameterobject-flat-param 文档
-
-knife4j:
- enable: true
- setting:
- language: zh_cn
-
-# MyBatis Plus 的配置项
-mybatis-plus:
- configuration:
- map-underscore-to-camel-case: true # 虽然默认为 true ，但是还是显示去指定下。
- global-config:
- db-config:
- id-type: NONE # “智能”模式，基于 IdTypeEnvironmentPostProcessor + 数据源的类型，自动适配成 AUTO、INPUT 模式。
- # id-type: AUTO # 自增 ID，适合 MySQL 等直接自增的数据库
- # id-type: INPUT # 用户输入 ID，适合 Oracle、PostgreSQL、Kingbase、DB2、H2 数据库
- # id-type: ASSIGN_ID # 分配 ID，默认使用雪花算法。注意，Oracle、PostgreSQL、Kingbase、DB2、H2 数据库时，需要去除实体类上的 @KeySequence 注解
- logic-delete-value: 1 # 逻辑已删除值(默认为 1)
- logic-not-delete-value: 0 # 逻辑未删除值(默认为 0)
- banner: false # 关闭控制台的 Banner 打印
- type-aliases-package: ${viewsh.info.base-package}.dal.dataobject
- encryptor:
- password: XDV71a+xqStEA3WH # 加解密的秘钥，可使用 https://www.imaegoo.com/2020/aes-key-generator/ 网站生成
-
-mybatis-plus-join:
- banner: false # 关闭控制台的 Banner 打印
-
-# Spring Data Redis 配置
-spring:
- data:
- redis:
- repositories:
- enabled: false # 项目未使用到 Spring Data Redis 的 Repository，所以直接禁用，保证启动速度
-
-# VO 转换（数据翻译）相关
-easy-trans:
- is-enable-global: false # 【默认禁用，对性能确认压力大】启用全局翻译（拦截所有 SpringMVC ResponseBody 进行自动翻译 )。如果对于性能要求很高可关闭此配置，或通过 @IgnoreTrans 忽略某个接口
-
---- #################### RPC 远程调用相关配置 ####################
-
---- #################### 消息队列相关 ####################
-
-# rocketmq 配置项，对应 RocketMQProperties 配置类
-rocketmq:
- # Producer 配置项
- producer:
- group: ${spring.application.name}_PRODUCER # 生产者分组
-
-spring:
- # Kafka 配置项，对应 KafkaProperties 配置类
- kafka:
- # Kafka Producer 配置项
- producer:
- acks: 1 # 0-不应答。1-leader 应答。all-所有 leader 和 follower 应答。
- retries: 3 # 发送失败时，重试发送的次数
- value-serializer: org.springframework.kafka.support.serializer.JsonSerializer # 消息的 value 的序列化
- # Kafka Consumer 配置项
- consumer:
- auto-offset-reset: earliest # 设置消费者分组最初的消费进度为 earliest 。可参考博客 https://blog.csdn.net/lishuangzhe7047/article/details/74530417 理解
- value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
- properties:
- spring.json.trusted.packages: '*'
- # Kafka Consumer Listener 监听器配置
- listener:
- missing-topics-fatal: false # 消费监听接口监听的主题不存在时，默认会报错。所以通过设置为 false ，解决报错
-
---- #################### 定时任务相关配置 ####################
-
-xxl:
- job:
- executor:
- appname: ${spring.application.name} # 执行器 AppName
- logpath: ${user.home}/logs/xxl-job/${spring.application.name} # 执行器运行日志文件存储磁盘路径
- accessToken: default_token # 执行器通讯TOKEN
-
---- #################### 芋道相关配置 ####################
-
-viewsh:
- info:
- version: 1.0.0
- base-package: com.viewsh.module.iot
- web:
- admin-ui:
- url: http://dashboard.viewsh.iocoder.cn # Admin 管理后台 UI 的地址
- xss:
- enable: false
- exclude-urls: # 如下 url，仅仅是为了演示，去掉配置也没关系
- - ${management.endpoints.web.base-path}/** # 不处理 Actuator 的请求
- swagger:
- title: 管理后台
- description: 提供管理员管理的所有功能
- version: ${viewsh.info.version}
- tenant: # 多租户相关配置项
- enable: true
- ignore-urls:
- ignore-tables:
- ignore-caches:
- - iot:device
- - iot:thing_model_list
- iot:
- message-bus:
- type: redis # 消息总线的类型
-
+spring:
+ application:
+ name: iot-server
+
+ profiles:
+ active: local
+
+ main:
+ allow-circular-references: true # 允许循环依赖，因为项目是三层架构，无法避免这个情况。
+ allow-bean-definition-overriding: true # 允许 Bean 覆盖，例如说 Feign 等会存在重复定义的服务
+
+ config:
+ import:
+ - optional:classpath:application-${spring.profiles.active}.yaml # 加载【本地】配置
+ - optional:nacos:${spring.application.name}-${spring.profiles.active}.yaml # 加载【Nacos】的配置
+
+ # Servlet 配置
+ servlet:
+ # 文件上传相关配置项
+ multipart:
+ max-file-size: 16MB # 单个文件大小
+ max-request-size: 32MB # 设置总上传的文件大小
+
+ # Jackson 配置项
+ jackson:
+ serialization:
+ write-dates-as-timestamps: true # 设置 LocalDateTime 的格式，使用时间戳
+ write-date-timestamps-as-nanoseconds: false # 设置不使用 nanoseconds 的格式。例如说 1611460870.401，而是直接 1611460870401
+ write-durations-as-timestamps: true # 设置 Duration 的格式，使用时间戳
+ fail-on-empty-beans: false # 允许序列化无属性的 Bean
+
+ # Cache 配置项
+ cache:
+ type: REDIS
+ redis:
+ time-to-live: 1h # 设置过期时间为 1 小时
+
+server:
+ port: 48091
+
+logging:
+ file:
+ name: ${user.home}/logs/${spring.application.name}.log # 日志文件名，全路径
+
+--- #################### 接口文档配置 ####################
+
+springdoc:
+ api-docs:
+ enabled: true # 1. 是否开启 Swagger 接文档的元数据
+ path: /v3/api-docs
+ swagger-ui:
+ enabled: true # 2.1 是否开启 Swagger 文档的官方 UI 界面
+ path: /swagger-ui
+ default-flat-param-object: true # 参见 https://doc.xiaominfo.com/docs/faq/v4/knife4j-parameterobject-flat-param 文档
+
+knife4j:
+ enable: true
+ setting:
+ language: zh_cn
+
+# MyBatis Plus 的配置项
+mybatis-plus:
+ configuration:
+ map-underscore-to-camel-case: true # 虽然默认为 true ，但是还是显示去指定下。
+ global-config:
+ db-config:
+ id-type: NONE # “智能”模式，基于 IdTypeEnvironmentPostProcessor + 数据源的类型，自动适配成 AUTO、INPUT 模式。
+ # id-type: AUTO # 自增 ID，适合 MySQL 等直接自增的数据库
+ # id-type: INPUT # 用户输入 ID，适合 Oracle、PostgreSQL、Kingbase、DB2、H2 数据库
+ # id-type: ASSIGN_ID # 分配 ID，默认使用雪花算法。注意，Oracle、PostgreSQL、Kingbase、DB2、H2 数据库时，需要去除实体类上的 @KeySequence 注解
+ logic-delete-value: 1 # 逻辑已删除值(默认为 1)
+ logic-not-delete-value: 0 # 逻辑未删除值(默认为 0)
+ banner: false # 关闭控制台的 Banner 打印
+ type-aliases-package: ${viewsh.info.base-package}.dal.dataobject
+ encryptor:
+ password: XDV71a+xqStEA3WH # 加解密的秘钥，可使用 https://www.imaegoo.com/2020/aes-key-generator/ 网站生成
+
+mybatis-plus-join:
+ banner: false # 关闭控制台的 Banner 打印
+
+# Spring Data Redis 配置
+spring:
+ data:
+ redis:
+ repositories:
+ enabled: false # 项目未使用到 Spring Data Redis 的 Repository，所以直接禁用，保证启动速度
+
+# VO 转换（数据翻译）相关
+easy-trans:
+ is-enable-global: false # 【默认禁用，对性能确认压力大】启用全局翻译（拦截所有 SpringMVC ResponseBody 进行自动翻译 )。如果对于性能要求很高可关闭此配置，或通过 @IgnoreTrans 忽略某个接口
+
+--- #################### RPC 远程调用相关配置 ####################
+
+--- #################### 消息队列相关 ####################
+
+# rocketmq 配置项，对应 RocketMQProperties 配置类
+rocketmq:
+ # Producer 配置项
+ producer:
+ group: ${spring.application.name}_PRODUCER # 生产者分组
+
+spring:
+ # Kafka 配置项，对应 KafkaProperties 配置类
+ kafka:
+ # Kafka Producer 配置项
+ producer:
+ acks: 1 # 0-不应答。1-leader 应答。all-所有 leader 和 follower 应答。
+ retries: 3 # 发送失败时，重试发送的次数
+ value-serializer: org.springframework.kafka.support.serializer.JsonSerializer # 消息的 value 的序列化
+ # Kafka Consumer 配置项
+ consumer:
+ auto-offset-reset: earliest # 设置消费者分组最初的消费进度为 earliest 。可参考博客 https://blog.csdn.net/lishuangzhe7047/article/details/74530417 理解
+ value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
+ properties:
+ spring.json.trusted.packages: '*'
+ # Kafka Consumer Listener 监听器配置
+ listener:
+ missing-topics-fatal: false # 消费监听接口监听的主题不存在时，默认会报错。所以通过设置为 false ，解决报错
+
+--- #################### 定时任务相关配置 ####################
+
+xxl:
+ job:
+ executor:
+ appname: ${spring.application.name} # 执行器 AppName
+ logpath: ${user.home}/logs/xxl-job/${spring.application.name} # 执行器运行日志文件存储磁盘路径
+ accessToken: default_token # 执行器通讯TOKEN
+
+--- #################### 芋道相关配置 ####################
+
+viewsh:
+ info:
+ version: 1.0.0
+ base-package: com.viewsh.module.iot
+ web:
+ admin-ui:
+ url: http://dashboard.viewsh.iocoder.cn # Admin 管理后台 UI 的地址
+ xss:
+ enable: false
+ exclude-urls: # 如下 url，仅仅是为了演示，去掉配置也没关系
+ - ${management.endpoints.web.base-path}/** # 不处理 Actuator 的请求
+ swagger:
+ title: 管理后台
+ description: 提供管理员管理的所有功能
+ version: ${viewsh.info.version}
+ tenant: # 多租户相关配置项
+ enable: true
+ ignore-urls:
+ ignore-tables:
+ ignore-caches:
+ - iot:device
+ - iot:thing_model_list
+ iot:
+ message-bus:
+ type: redis # 消息总线的类型
+ # 跨模块事件总线配置（IntegrationEventBus）
+ integration:
+ mq:
+ enabled: true # 是否启用跨模块事件发布
+ producer-group: integration-event-producer # 生产者组名
+ send-timeout-ms: 3000 # 发送超时时间（毫秒）
+
debug: false
\ No newline at end of file