feat(tenant): 租户-项目两级架构 Phase 2 — IoT + Ops 业务迁移
DO 迁移 (15个 TenantBaseDO → ProjectBaseDO): - IoT: IotDeviceDO - Ops 核心: OpsOrderDO, OpsOrderEventDO, OpsOrderDispatchDO, OpsOrderQueueDO, OpsBusAreaDO, OpsAreaDeviceRelationDO, OpsDeviceTrajectoryDO - Ops 保洁: OpsOrderCleanExtDO, OpsCleanerStatusDO, OpsCleanerPerformanceMonthlyDO, OpsInspectionRecordDO, OpsInspectionRecordItemDO - Ops 安保: OpsOrderSecurityExtDO, OpsAreaSecurityUserDO IoT 适配: - IotDeviceRespDTO 新增 projectId 字段 - IotDeviceMessage 新增 projectId 字段 - IotDeviceMessageServiceImpl.appendDeviceMessage() 设置 projectId - IotCleanRuleMessageHandler 嵌套 ProjectUtils.execute() 设置项目上下文 缓存改造: - ProjectRedisCacheManager extends TenantRedisCacheManager，追加 :projectId 后缀 - ViewshTenantAutoConfiguration 替换为 ProjectRedisCacheManager SQL 迁移脚本 (sql/mysql/project/): - 01-create-tables.sql: system_project + system_user_project 建表 - 02-default-data.sql: 默认项目 + 用户关联回填 - 03-alter-business-tables.sql: 15 张表添加 project_id (NULL → 回填 → NOT NULL → 索引) - 04-index-audit.sql: 现有索引审计 + project_id 补充建议 - 99-rollback.sql: 完整回滚方案 附带修复: - fix(ops): UserDispatchStatusServiceImpl 添加缺失的 KEY_PREFIX 常量 Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -1,138 +1,139 @@
|
||||
package com.viewsh.module.iot.gateway.service.device.message;
|
||||
|
||||
import cn.hutool.core.util.StrUtil;
|
||||
import com.viewsh.framework.common.util.collection.CollectionUtils;
|
||||
import com.viewsh.module.iot.core.biz.dto.IotDeviceRespDTO;
|
||||
import com.viewsh.module.iot.core.mq.message.IotDeviceMessage;
|
||||
import com.viewsh.module.iot.core.mq.producer.IotDeviceMessageProducer;
|
||||
import com.viewsh.module.iot.core.util.IotDeviceMessageUtils;
|
||||
import com.viewsh.module.iot.gateway.codec.IotDeviceMessageCodec;
|
||||
import com.viewsh.module.iot.gateway.service.device.IotDeviceService;
|
||||
import jakarta.annotation.Resource;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import static com.viewsh.framework.common.exception.util.ServiceExceptionUtil.exception;
|
||||
import static com.viewsh.module.iot.gateway.enums.ErrorCodeConstants.DEVICE_NOT_EXISTS;
|
||||
|
||||
/**
|
||||
* IoT 设备消息 Service 实现类
|
||||
*
|
||||
* @author 芋道源码
|
||||
*/
|
||||
@Service
|
||||
@Slf4j
|
||||
public class IotDeviceMessageServiceImpl implements IotDeviceMessageService {
|
||||
|
||||
/**
|
||||
* 编解码器
|
||||
*/
|
||||
private final Map<String, IotDeviceMessageCodec> codes;
|
||||
|
||||
@Resource
|
||||
private IotDeviceService deviceService;
|
||||
|
||||
@Resource
|
||||
private IotDeviceMessageProducer deviceMessageProducer;
|
||||
|
||||
public IotDeviceMessageServiceImpl(List<IotDeviceMessageCodec> codes) {
|
||||
this.codes = CollectionUtils.convertMap(codes, IotDeviceMessageCodec::type);
|
||||
}
|
||||
|
||||
@Override
|
||||
public byte[] encodeDeviceMessage(IotDeviceMessage message,
|
||||
String productKey, String deviceName) {
|
||||
// 1.1 获取设备信息
|
||||
IotDeviceRespDTO device = deviceService.getDeviceFromCache(productKey, deviceName);
|
||||
if (device == null) {
|
||||
throw exception(DEVICE_NOT_EXISTS, productKey, deviceName);
|
||||
}
|
||||
// 1.2 获取编解码器
|
||||
IotDeviceMessageCodec codec = codes.get(device.getCodecType());
|
||||
if (codec == null) {
|
||||
throw new IllegalArgumentException(StrUtil.format("编解码器({}) 不存在", device.getCodecType()));
|
||||
}
|
||||
|
||||
// 2. 编码消息
|
||||
return codec.encode(message);
|
||||
}
|
||||
|
||||
@Override
|
||||
public byte[] encodeDeviceMessage(IotDeviceMessage message,
|
||||
String codecType) {
|
||||
// 1. 获取编解码器
|
||||
IotDeviceMessageCodec codec = codes.get(codecType);
|
||||
if (codec == null) {
|
||||
throw new IllegalArgumentException(StrUtil.format("编解码器({}) 不存在", codecType));
|
||||
}
|
||||
|
||||
// 2. 编码消息
|
||||
return codec.encode(message);
|
||||
}
|
||||
|
||||
@Override
|
||||
public IotDeviceMessage decodeDeviceMessage(byte[] bytes,
|
||||
String productKey, String deviceName) {
|
||||
// 1.1 获取设备信息
|
||||
IotDeviceRespDTO device = deviceService.getDeviceFromCache(productKey, deviceName);
|
||||
if (device == null) {
|
||||
throw exception(DEVICE_NOT_EXISTS, productKey, deviceName);
|
||||
}
|
||||
// 1.2 获取编解码器
|
||||
IotDeviceMessageCodec codec = codes.get(device.getCodecType());
|
||||
if (codec == null) {
|
||||
throw new IllegalArgumentException(StrUtil.format("编解码器({}) 不存在", device.getCodecType()));
|
||||
}
|
||||
|
||||
// 2. 解码消息
|
||||
return codec.decode(bytes);
|
||||
}
|
||||
|
||||
@Override
|
||||
public IotDeviceMessage decodeDeviceMessage(byte[] bytes, String codecType) {
|
||||
// 1. 获取编解码器
|
||||
IotDeviceMessageCodec codec = codes.get(codecType);
|
||||
if (codec == null) {
|
||||
throw new IllegalArgumentException(StrUtil.format("编解码器({}) 不存在", codecType));
|
||||
}
|
||||
|
||||
// 2. 解码消息
|
||||
return codec.decode(bytes);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void sendDeviceMessage(IotDeviceMessage message,
|
||||
String productKey, String deviceName, String serverId) {
|
||||
// 1. 获取设备信息
|
||||
IotDeviceRespDTO device = deviceService.getDeviceFromCache(productKey, deviceName);
|
||||
if (device == null) {
|
||||
throw exception(DEVICE_NOT_EXISTS, productKey, deviceName);
|
||||
}
|
||||
|
||||
// 2. 发送消息
|
||||
appendDeviceMessage(message, device, serverId);
|
||||
deviceMessageProducer.sendDeviceMessage(message);
|
||||
}
|
||||
|
||||
/**
|
||||
* 补充消息的后端字段
|
||||
*
|
||||
* @param message 消息
|
||||
* @param device 设备信息
|
||||
* @param serverId 设备连接的 serverId
|
||||
*/
|
||||
private void appendDeviceMessage(IotDeviceMessage message,
|
||||
IotDeviceRespDTO device, String serverId) {
|
||||
message.setId(IotDeviceMessageUtils.generateMessageId()).setReportTime(LocalDateTime.now())
|
||||
.setDeviceId(device.getId()).setTenantId(device.getTenantId()).setServerId(serverId);
|
||||
// 特殊：如果设备没有指定 requestId，则使用 messageId
|
||||
if (StrUtil.isEmpty(message.getRequestId())) {
|
||||
message.setRequestId(message.getId());
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
package com.viewsh.module.iot.gateway.service.device.message;
|
||||
|
||||
import cn.hutool.core.util.StrUtil;
|
||||
import com.viewsh.framework.common.util.collection.CollectionUtils;
|
||||
import com.viewsh.module.iot.core.biz.dto.IotDeviceRespDTO;
|
||||
import com.viewsh.module.iot.core.mq.message.IotDeviceMessage;
|
||||
import com.viewsh.module.iot.core.mq.producer.IotDeviceMessageProducer;
|
||||
import com.viewsh.module.iot.core.util.IotDeviceMessageUtils;
|
||||
import com.viewsh.module.iot.gateway.codec.IotDeviceMessageCodec;
|
||||
import com.viewsh.module.iot.gateway.service.device.IotDeviceService;
|
||||
import jakarta.annotation.Resource;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import static com.viewsh.framework.common.exception.util.ServiceExceptionUtil.exception;
|
||||
import static com.viewsh.module.iot.gateway.enums.ErrorCodeConstants.DEVICE_NOT_EXISTS;
|
||||
|
||||
/**
|
||||
* IoT 设备消息 Service 实现类
|
||||
*
|
||||
* @author 芋道源码
|
||||
*/
|
||||
@Service
|
||||
@Slf4j
|
||||
public class IotDeviceMessageServiceImpl implements IotDeviceMessageService {
|
||||
|
||||
/**
|
||||
* 编解码器
|
||||
*/
|
||||
private final Map<String, IotDeviceMessageCodec> codes;
|
||||
|
||||
@Resource
|
||||
private IotDeviceService deviceService;
|
||||
|
||||
@Resource
|
||||
private IotDeviceMessageProducer deviceMessageProducer;
|
||||
|
||||
public IotDeviceMessageServiceImpl(List<IotDeviceMessageCodec> codes) {
|
||||
this.codes = CollectionUtils.convertMap(codes, IotDeviceMessageCodec::type);
|
||||
}
|
||||
|
||||
@Override
|
||||
public byte[] encodeDeviceMessage(IotDeviceMessage message,
|
||||
String productKey, String deviceName) {
|
||||
// 1.1 获取设备信息
|
||||
IotDeviceRespDTO device = deviceService.getDeviceFromCache(productKey, deviceName);
|
||||
if (device == null) {
|
||||
throw exception(DEVICE_NOT_EXISTS, productKey, deviceName);
|
||||
}
|
||||
// 1.2 获取编解码器
|
||||
IotDeviceMessageCodec codec = codes.get(device.getCodecType());
|
||||
if (codec == null) {
|
||||
throw new IllegalArgumentException(StrUtil.format("编解码器({}) 不存在", device.getCodecType()));
|
||||
}
|
||||
|
||||
// 2. 编码消息
|
||||
return codec.encode(message);
|
||||
}
|
||||
|
||||
@Override
|
||||
public byte[] encodeDeviceMessage(IotDeviceMessage message,
|
||||
String codecType) {
|
||||
// 1. 获取编解码器
|
||||
IotDeviceMessageCodec codec = codes.get(codecType);
|
||||
if (codec == null) {
|
||||
throw new IllegalArgumentException(StrUtil.format("编解码器({}) 不存在", codecType));
|
||||
}
|
||||
|
||||
// 2. 编码消息
|
||||
return codec.encode(message);
|
||||
}
|
||||
|
||||
@Override
|
||||
public IotDeviceMessage decodeDeviceMessage(byte[] bytes,
|
||||
String productKey, String deviceName) {
|
||||
// 1.1 获取设备信息
|
||||
IotDeviceRespDTO device = deviceService.getDeviceFromCache(productKey, deviceName);
|
||||
if (device == null) {
|
||||
throw exception(DEVICE_NOT_EXISTS, productKey, deviceName);
|
||||
}
|
||||
// 1.2 获取编解码器
|
||||
IotDeviceMessageCodec codec = codes.get(device.getCodecType());
|
||||
if (codec == null) {
|
||||
throw new IllegalArgumentException(StrUtil.format("编解码器({}) 不存在", device.getCodecType()));
|
||||
}
|
||||
|
||||
// 2. 解码消息
|
||||
return codec.decode(bytes);
|
||||
}
|
||||
|
||||
@Override
|
||||
public IotDeviceMessage decodeDeviceMessage(byte[] bytes, String codecType) {
|
||||
// 1. 获取编解码器
|
||||
IotDeviceMessageCodec codec = codes.get(codecType);
|
||||
if (codec == null) {
|
||||
throw new IllegalArgumentException(StrUtil.format("编解码器({}) 不存在", codecType));
|
||||
}
|
||||
|
||||
// 2. 解码消息
|
||||
return codec.decode(bytes);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void sendDeviceMessage(IotDeviceMessage message,
|
||||
String productKey, String deviceName, String serverId) {
|
||||
// 1. 获取设备信息
|
||||
IotDeviceRespDTO device = deviceService.getDeviceFromCache(productKey, deviceName);
|
||||
if (device == null) {
|
||||
throw exception(DEVICE_NOT_EXISTS, productKey, deviceName);
|
||||
}
|
||||
|
||||
// 2. 发送消息
|
||||
appendDeviceMessage(message, device, serverId);
|
||||
deviceMessageProducer.sendDeviceMessage(message);
|
||||
}
|
||||
|
||||
/**
|
||||
* 补充消息的后端字段
|
||||
*
|
||||
* @param message 消息
|
||||
* @param device 设备信息
|
||||
* @param serverId 设备连接的 serverId
|
||||
*/
|
||||
private void appendDeviceMessage(IotDeviceMessage message,
|
||||
IotDeviceRespDTO device, String serverId) {
|
||||
message.setId(IotDeviceMessageUtils.generateMessageId()).setReportTime(LocalDateTime.now())
|
||||
.setDeviceId(device.getId()).setTenantId(device.getTenantId())
|
||||
.setProjectId(device.getProjectId()).setServerId(serverId);
|
||||
// 特殊：如果设备没有指定 requestId，则使用 messageId
|
||||
if (StrUtil.isEmpty(message.getRequestId())) {
|
||||
message.setRequestId(message.getId());
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user