BJ_ZTEITS-Developers / parking-cloud-yudao

Commit bfbc352a1ca80c7ddedd5dfa535b4083218ebe86

Authored by YunaiV
1 parent 18ed7b50

feat(iot):【网关设备：65%】整体初步实现（优化部分代码）

Inline Side-by-side
Showing 14 changed files with 579 additions and 556 deletions
yudao-framework/yudao-common/src/main/java/cn/iocoder/yudao/framework/common/util/json/JsonUtils.java
  View file @bfbc352
... ... @@ -229,4 +229,37 @@ public class JsonUtils {
229 229 return JSONUtil.isTypeJSONObject(str);
230 230 }
231 231  
  232 + /**
  233 + * 将 Object 转换为目标类型
  234 + * <p>
  235 + * 避免先转 jsonString 再 parseObject 的性能损耗
  236 + *
  237 + * @param obj 源对象（可以是 Map、POJO 等）
  238 + * @param clazz 目标类型
  239 + * @return 转换后的对象
  240 + */
  241 + public static <T> T convertObject(Object obj, Class<T> clazz) {
  242 + if (obj == null) {
  243 + return null;
  244 + }
  245 + if (clazz.isInstance(obj)) {
  246 + return clazz.cast(obj);
  247 + }
  248 + return objectMapper.convertValue(obj, clazz);
  249 + }
  250 +
  251 + /**
  252 + * 将 Object 转换为目标类型（支持泛型）
  253 + *
  254 + * @param obj 源对象
  255 + * @param typeReference 目标类型引用
  256 + * @return 转换后的对象
  257 + */
  258 + public static <T> T convertObject(Object obj, TypeReference<T> typeReference) {
  259 + if (obj == null) {
  260 + return null;
  261 + }
  262 + return objectMapper.convertValue(obj, typeReference);
  263 + }
  264 +
232 265 }
... ...
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/IotDeviceService.java
  View file @bfbc352
... ... @@ -4,6 +4,10 @@ import cn.iocoder.yudao.framework.common.pojo.PageResult;
4 4 import cn.iocoder.yudao.module.iot.controller.admin.device.vo.device.*;
5 5 import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceAuthReqDTO;
6 6 import cn.iocoder.yudao.module.iot.core.enums.IotDeviceStateEnum;
  7 +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
  8 +import cn.iocoder.yudao.module.iot.core.topic.IotDeviceIdentity;
  9 +import cn.iocoder.yudao.module.iot.core.topic.auth.IotSubDeviceRegisterRespDTO;
  10 +import cn.iocoder.yudao.module.iot.core.topic.topo.IotDeviceTopoGetRespDTO;
7 11 import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
8 12 import jakarta.validation.Valid;
9 13  
... ... @@ -321,4 +325,41 @@ public interface IotDeviceService {
321 325 */
322 326 List<IotDeviceDO> getDeviceListByGatewayId(Long gatewayId);
323 327  
  328 + // ========== 网关-拓扑管理（设备上报） ==========
  329 +
  330 + /**
  331 + * 处理添加拓扑关系消息（网关设备上报）
  332 + *
  333 + * @param message 消息
  334 + * @param gatewayDevice 网关设备
  335 + * @return 成功添加的子设备列表
  336 + */
  337 + List<IotDeviceIdentity> handleTopoAddMessage(IotDeviceMessage message, IotDeviceDO gatewayDevice);
  338 +
  339 + /**
  340 + * 处理删除拓扑关系消息（网关设备上报）
  341 + *
  342 + * @param message 消息
  343 + * @param gatewayDevice 网关设备
  344 + * @return 成功删除的子设备列表
  345 + */
  346 + List<IotDeviceIdentity> handleTopoDeleteMessage(IotDeviceMessage message, IotDeviceDO gatewayDevice);
  347 +
  348 + /**
  349 + * 处理获取拓扑关系消息（网关设备上报）
  350 + *
  351 + * @param gatewayDevice 网关设备
  352 + * @return 拓扑关系响应
  353 + */
  354 + IotDeviceTopoGetRespDTO handleTopoGetMessage(IotDeviceDO gatewayDevice);
  355 +
  356 + /**
  357 + * 处理子设备动态注册消息（网关设备上报）
  358 + *
  359 + * @param message 消息
  360 + * @param gatewayDevice 网关设备
  361 + * @return 注册结果列表
  362 + */
  363 + List<IotSubDeviceRegisterRespDTO> handleSubDeviceRegisterMessage(IotDeviceMessage message, IotDeviceDO gatewayDevice);
  364 +
324 365 }
... ...
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/IotDeviceServiceImpl.java
  View file @bfbc352
... ... @@ -7,6 +7,7 @@ import cn.hutool.core.util.StrUtil;
7 7 import cn.hutool.extra.spring.SpringUtil;
8 8 import cn.iocoder.yudao.framework.common.exception.ServiceException;
9 9 import cn.iocoder.yudao.framework.common.pojo.PageResult;
  10 +import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
10 11 import cn.iocoder.yudao.framework.common.util.object.BeanUtils;
11 12 import cn.iocoder.yudao.framework.common.util.validation.ValidationUtils;
12 13 import cn.iocoder.yudao.framework.tenant.core.aop.TenantIgnore;
... ... @@ -14,6 +15,13 @@ import cn.iocoder.yudao.framework.tenant.core.util.TenantUtils;
14 15 import cn.iocoder.yudao.module.iot.controller.admin.device.vo.device.*;
15 16 import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceAuthReqDTO;
16 17 import cn.iocoder.yudao.module.iot.core.enums.IotDeviceStateEnum;
  18 +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
  19 +import cn.iocoder.yudao.module.iot.core.topic.IotDeviceIdentity;
  20 +import cn.iocoder.yudao.module.iot.core.topic.auth.IotSubDeviceRegisterReqDTO;
  21 +import cn.iocoder.yudao.module.iot.core.topic.auth.IotSubDeviceRegisterRespDTO;
  22 +import cn.iocoder.yudao.module.iot.core.topic.topo.IotDeviceTopoAddReqDTO;
  23 +import cn.iocoder.yudao.module.iot.core.topic.topo.IotDeviceTopoDeleteReqDTO;
  24 +import cn.iocoder.yudao.module.iot.core.topic.topo.IotDeviceTopoGetRespDTO;
17 25 import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils;
18 26 import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
19 27 import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceGroupDO;
... ... @@ -64,6 +72,10 @@ public class IotDeviceServiceImpl implements IotDeviceService {
64 72  
65 73 @Override
66 74 public Long createDevice(IotDeviceSaveReqVO createReqVO) {
  75 + return createDevice0(createReqVO).getId();
  76 + }
  77 +
  78 + private IotDeviceDO createDevice0(IotDeviceSaveReqVO createReqVO) {
67 79 // 1.1 校验产品是否存在
68 80 IotProductDO product = productService.getProduct(createReqVO.getProductId());
69 81 if (product == null) {
... ... @@ -81,7 +93,7 @@ public class IotDeviceServiceImpl implements IotDeviceService {
81 93 IotDeviceDO device = BeanUtils.toBean(createReqVO, IotDeviceDO.class);
82 94 initDevice(device, product);
83 95 deviceMapper.insert(device);
84   - return device.getId();
  96 + return device;
85 97 }
86 98  
87 99 private void validateCreateDeviceParam(String productKey, String deviceName,
... ... @@ -298,6 +310,37 @@ public class IotDeviceServiceImpl implements IotDeviceService {
298 310  
299 311 // 2. 清空对应缓存
300 312 deleteDeviceCache(device);
  313 +
  314 + // 3. 网关设备下线时，联动所有子设备下线
  315 + if (Objects.equals(state, IotDeviceStateEnum.OFFLINE.getState())
  316 + && IotProductDeviceTypeEnum.isGateway(device.getDeviceType())) {
  317 + handleGatewayOffline(device);
  318 + }
  319 + }
  320 +
  321 + /**
  322 + * 处理网关下线，联动所有子设备下线
  323 + *
  324 + * @param gatewayDevice 网关设备
  325 + */
  326 + private void handleGatewayOffline(IotDeviceDO gatewayDevice) {
  327 + List<IotDeviceDO> subDevices = deviceMapper.selectListByGatewayId(gatewayDevice.getId());
  328 + if (CollUtil.isEmpty(subDevices)) {
  329 + return;
  330 + }
  331 + for (IotDeviceDO subDevice : subDevices) {
  332 + if (Objects.equals(subDevice.getState(), IotDeviceStateEnum.ONLINE.getState())) {
  333 + try {
  334 + updateDeviceState(subDevice, IotDeviceStateEnum.OFFLINE.getState());
  335 + log.info("[handleGatewayOffline][网关({}/{}) 下线，子设备({}/{}) 联动下线]",
  336 + gatewayDevice.getProductKey(), gatewayDevice.getDeviceName(),
  337 + subDevice.getProductKey(), subDevice.getDeviceName());
  338 + } catch (Exception ex) {
  339 + log.error("[handleGatewayOffline][子设备({}/{}) 下线失败]",
  340 + subDevice.getProductKey(), subDevice.getDeviceName(), ex);
  341 + }
  342 + }
  343 + }
301 344 }
302 345  
303 346 @Override
... ... @@ -533,19 +576,10 @@ public class IotDeviceServiceImpl implements IotDeviceService {
533 576 }
534 577 // 1.1 校验网关设备存在且类型正确
535 578 validateGatewayDeviceExists(gatewayId);
536   - // 1.2 校验子设备存在
  579 + // 1.2 校验每个设备是否可绑定
537 580 List<IotDeviceDO> devices = deviceMapper.selectByIds(ids);
538   - if (devices.size() != ids.size()) {
539   - throw exception(DEVICE_NOT_EXISTS);
540   - }
541   - // 1.3 校验每个设备是否可绑定
542 581 for (IotDeviceDO device : devices) {
543   - if (!IotProductDeviceTypeEnum.isGatewaySub(device.getDeviceType())) {
544   - throw exception(DEVICE_NOT_GATEWAY_SUB, device.getProductKey(), device.getDeviceName());
545   - }
546   - if (device.getGatewayId() != null && !device.getGatewayId().equals(gatewayId)) {
547   - throw exception(DEVICE_GATEWAY_BINDTO_EXISTS, device.getProductKey(), device.getDeviceName());
548   - }
  582 + checkSubDeviceCanBind(device, gatewayId);
549 583 }
550 584  
551 585 // 2. 批量更新数据库
... ... @@ -555,21 +589,33 @@ public class IotDeviceServiceImpl implements IotDeviceService {
555 589  
556 590 // 3. 清空对应缓存
557 591 deleteDeviceCache(devices);
  592 +
  593 + // TODO @AI：需要下发网关设备，让其建立拓扑关系吗？（增加）
  594 + }
  595 +
  596 + private void checkSubDeviceCanBind(IotDeviceDO device, Long gatewayId) {
  597 + if (!IotProductDeviceTypeEnum.isGatewaySub(device.getDeviceType())) {
  598 + throw exception(DEVICE_NOT_GATEWAY_SUB, device.getProductKey(), device.getDeviceName());
  599 + }
  600 + if (ObjUtil.equals(device.getGatewayId(), gatewayId)) {
  601 + throw exception(DEVICE_GATEWAY_BINDTO_EXISTS, device.getProductKey(), device.getDeviceName());
  602 + }
558 603 }
559 604  
560 605 @Override
561 606 @Transactional(rollbackFor = Exception.class)
562 607 public void unbindDeviceGateway(Collection<Long> ids) {
  608 + // 1. 校验设备存在
563 609 if (CollUtil.isEmpty(ids)) {
564 610 return;
565 611 }
566   - // 1. 校验设备存在
567 612 List<IotDeviceDO> devices = deviceMapper.selectByIds(ids);
568   - if (devices.size() != ids.size()) {
569   - throw exception(DEVICE_NOT_EXISTS);
  613 + if (CollUtil.isNotEmpty(devices)) {
  614 + return;
570 615 }
571 616  
572 617 // 2. 批量更新数据库（将 gatewayId 设置为 null）
  618 + // TODO @AI：需要搞个方法，专门批量更新某个字段为 null。
573 619 List<IotDeviceDO> updateList = devices.stream()
574 620 .filter(device -> device.getGatewayId() != null)
575 621 .map(device -> new IotDeviceDO().setId(device.getId()).setGatewayId(null))
... ... @@ -580,6 +626,8 @@ public class IotDeviceServiceImpl implements IotDeviceService {
580 626  
581 627 // 3. 清空对应缓存
582 628 deleteDeviceCache(devices);
  629 +
  630 + // TODO @AI：需要下发网关设备，让其建立拓扑关系吗？（减少）
583 631 }
584 632  
585 633 @Override
... ... @@ -592,6 +640,202 @@ public class IotDeviceServiceImpl implements IotDeviceService {
592 640 return deviceMapper.selectListByGatewayId(gatewayId);
593 641 }
594 642  
  643 + // ========== 网关-拓扑管理（设备上报） ==========
  644 +
  645 + @Override
  646 + public List<IotDeviceIdentity> handleTopoAddMessage(IotDeviceMessage message, IotDeviceDO gatewayDevice) {
  647 + // 1.1 校验网关设备类型
  648 + if (!IotProductDeviceTypeEnum.isGateway(gatewayDevice.getDeviceType())) {
  649 + throw exception(DEVICE_NOT_GATEWAY);
  650 + }
  651 + // 1.2 解析参数
  652 + IotDeviceTopoAddReqDTO params = JsonUtils.parseObject(JsonUtils.toJsonString(message.getParams()),
  653 + IotDeviceTopoAddReqDTO.class);
  654 + if (params == null || CollUtil.isEmpty(params.getSubDevices())) {
  655 + throw exception(DEVICE_TOPO_PARAMS_INVALID);
  656 + }
  657 +
  658 + // 2. 遍历处理每个子设备
  659 + List<IotDeviceIdentity> addedSubDevices = new ArrayList<>();
  660 + for (IotDeviceAuthReqDTO subDeviceAuth : params.getSubDevices()) {
  661 + try {
  662 + IotDeviceDO subDevice = addDeviceTopo(gatewayDevice, subDeviceAuth);
  663 + addedSubDevices.add(new IotDeviceIdentity(subDevice.getProductKey(), subDevice.getDeviceName()));
  664 + } catch (Exception ex) {
  665 + log.warn("[handleTopoAddMessage][网关({}/{}) 添加子设备失败，message={}]",
  666 + gatewayDevice.getProductKey(), gatewayDevice.getDeviceName(), message, ex);
  667 + }
  668 + }
  669 +
  670 + // 3. 返回响应数据（包含成功添加的子设备列表）
  671 + return addedSubDevices;
  672 + }
  673 +
  674 + private IotDeviceDO addDeviceTopo(IotDeviceDO gatewayDevice, IotDeviceAuthReqDTO subDeviceAuth) {
  675 + // 1.1 解析子设备信息
  676 + IotDeviceAuthUtils.DeviceInfo subDeviceInfo = IotDeviceAuthUtils.parseUsername(subDeviceAuth.getUsername());
  677 + if (subDeviceInfo == null) {
  678 + throw exception(DEVICE_TOPO_SUB_DEVICE_USERNAME_INVALID);
  679 + }
  680 + // 1.2 校验子设备认证信息
  681 + if (!authDevice(subDeviceAuth)) {
  682 + throw exception(DEVICE_TOPO_SUB_DEVICE_AUTH_FAILED);
  683 + }
  684 + // 1.3 获取子设备
  685 + IotDeviceDO subDevice = getSelf().getDeviceFromCache(subDeviceInfo.getProductKey(), subDeviceInfo.getDeviceName());
  686 + if (subDevice == null) {
  687 + throw exception(DEVICE_NOT_EXISTS);
  688 + }
  689 + // 1.4 校验子设备类型
  690 + checkSubDeviceCanBind(subDevice, gatewayDevice.getId());
  691 +
  692 + // 2. 更新数据库
  693 + deviceMapper.updateById(new IotDeviceDO().setId(subDevice.getId()).setGatewayId(subDevice.getGatewayId()));
  694 + log.info("[addDeviceTopo][网关({}/{}) 绑定子设备({}/{})]",
  695 + gatewayDevice.getProductKey(), gatewayDevice.getDeviceName(),
  696 + subDevice.getProductKey(), subDevice.getDeviceName());
  697 +
  698 + // 3. 清空对应缓存
  699 + deleteDeviceCache(subDevice);
  700 + return subDevice;
  701 + }
  702 +
  703 + @Override
  704 + public List<IotDeviceIdentity> handleTopoDeleteMessage(IotDeviceMessage message, IotDeviceDO gatewayDevice) {
  705 + // 1.1 校验网关设备类型
  706 + if (!IotProductDeviceTypeEnum.isGateway(gatewayDevice.getDeviceType())) {
  707 + throw exception(DEVICE_NOT_GATEWAY);
  708 + }
  709 + // 1.2 解析参数
  710 + IotDeviceTopoDeleteReqDTO params = JsonUtils.parseObject(JsonUtils.toJsonString(message.getParams()),
  711 + IotDeviceTopoDeleteReqDTO.class);
  712 + if (params == null || CollUtil.isEmpty(params.getSubDevices())) {
  713 + throw exception(DEVICE_TOPO_PARAMS_INVALID);
  714 + }
  715 +
  716 + // 2. 遍历处理每个子设备
  717 + List<IotDeviceIdentity> deletedSubDevices = new ArrayList<>();
  718 + for (IotDeviceIdentity subDeviceIdentity : params.getSubDevices()) {
  719 + try {
  720 + deleteDeviceTopo(gatewayDevice, subDeviceIdentity);
  721 + deletedSubDevices.add(subDeviceIdentity);
  722 + } catch (Exception ex) {
  723 + log.warn("[handleTopoDeleteMessage][网关({}/{}) 删除子设备失败，productKey={}, deviceName={}]",
  724 + gatewayDevice.getProductKey(), gatewayDevice.getDeviceName(),
  725 + subDeviceIdentity.getProductKey(), subDeviceIdentity.getDeviceName(), ex);
  726 + }
  727 + }
  728 +
  729 + // 3. 返回响应数据（包含成功删除的子设备列表）
  730 + return deletedSubDevices;
  731 + }
  732 +
  733 + private void deleteDeviceTopo(IotDeviceDO gatewayDevice, IotDeviceIdentity subDeviceIdentity) {
  734 + // 1.1 获取子设备
  735 + IotDeviceDO subDevice = getSelf().getDeviceFromCache(subDeviceIdentity.getProductKey(), subDeviceIdentity.getDeviceName());
  736 + if (subDevice == null) {
  737 + throw exception(DEVICE_NOT_EXISTS);
  738 + }
  739 + // 1.2 校验子设备是否绑定到该网关
  740 + if (ObjUtil.notEqual(subDevice.getGatewayId(), gatewayDevice.getId())) {
  741 + throw exception(DEVICE_TOPO_SUB_NOT_BINDTO_GATEWAY,
  742 + subDeviceIdentity.getProductKey(), subDeviceIdentity.getDeviceName());
  743 + }
  744 +
  745 + // 2. 更新数据库
  746 + // TODO @AI：直接调用更新方法；
  747 +// unbindDeviceGateway(Collections.singletonList(subDevice.getId()));
  748 + log.info("[deleteDeviceTopo][网关({}/{}) 解绑子设备({}/{})]",
  749 + gatewayDevice.getProductKey(), gatewayDevice.getDeviceName(),
  750 + subDevice.getProductKey(), subDevice.getDeviceName());
  751 +
  752 + // 3. 清空对应缓存
  753 + deleteDeviceCache(subDevice);
  754 +
  755 + // 4. 子设备下线
  756 + if (Objects.equals(subDevice.getState(), IotDeviceStateEnum.ONLINE.getState())) {
  757 + updateDeviceState(subDevice, IotDeviceStateEnum.OFFLINE.getState());
  758 + }
  759 + }
  760 +
  761 + @Override
  762 + public IotDeviceTopoGetRespDTO handleTopoGetMessage(IotDeviceDO gatewayDevice) {
  763 + // 1. 校验网关设备类型
  764 + if (!IotProductDeviceTypeEnum.isGateway(gatewayDevice.getDeviceType())) {
  765 + throw exception(DEVICE_NOT_GATEWAY);
  766 + }
  767 +
  768 + // 2. 获取子设备列表并转换
  769 + List<IotDeviceDO> subDevices = deviceMapper.selectListByGatewayId(gatewayDevice.getId());
  770 + List<IotDeviceIdentity> subDeviceIdentities = convertList(subDevices, subDevice ->
  771 + new IotDeviceIdentity(subDevice.getProductKey(), subDevice.getDeviceName()));
  772 + return new IotDeviceTopoGetRespDTO().setSubDevices(subDeviceIdentities);
  773 + }
  774 +
  775 + @Override
  776 + public List<IotSubDeviceRegisterRespDTO> handleSubDeviceRegisterMessage(IotDeviceMessage message, IotDeviceDO gatewayDevice) {
  777 + // 1.1 校验网关设备类型
  778 + if (!IotProductDeviceTypeEnum.isGateway(gatewayDevice.getDeviceType())) {
  779 + throw exception(DEVICE_NOT_GATEWAY);
  780 + }
  781 + // 1.2 解析参数
  782 + if (!(message.getParams() instanceof List)) {
  783 + throw exception(DEVICE_SUB_REGISTER_PARAMS_INVALID);
  784 + }
  785 + // TODO @AI：这个要不也弄到 JsonUtils 里面去？感觉类似 convertObject 呀。
  786 + List<IotSubDeviceRegisterReqDTO> paramsList = JsonUtils.parseArray(JsonUtils.toJsonString(message.getParams()),
  787 + IotSubDeviceRegisterReqDTO.class);
  788 + if (CollUtil.isEmpty(paramsList)) {
  789 + throw exception(DEVICE_SUB_REGISTER_PARAMS_INVALID);
  790 + }
  791 +
  792 + // 2. 遍历注册每个子设备
  793 + List<IotSubDeviceRegisterRespDTO> results = new ArrayList<>(paramsList.size());
  794 + for (IotSubDeviceRegisterReqDTO params : paramsList) {
  795 + try {
  796 + IotDeviceDO device = registerSubDevice(gatewayDevice, params);
  797 + results.add(new IotSubDeviceRegisterRespDTO(
  798 + params.getProductKey(), params.getDeviceName(), device.getDeviceSecret()));
  799 + } catch (Exception ex) {
  800 + log.error("[handleSubDeviceRegisterMessage][子设备({}/{}) 注册失败]",
  801 + params.getProductKey(), params.getDeviceName(), ex);
  802 + }
  803 + }
  804 +
  805 + // 3. 返回响应数据（包含成功注册的子设备列表）
  806 + return results;
  807 + }
  808 +
  809 + private IotDeviceDO registerSubDevice(IotDeviceDO gatewayDevice, IotSubDeviceRegisterReqDTO params) {
  810 + // 1.1 校验产品
  811 + IotProductDO product = productService.getProductByProductKey(params.getProductKey());
  812 + if (product == null) {
  813 + throw exception(PRODUCT_NOT_EXISTS);
  814 + }
  815 + // 1.2 校验产品是否为网关子设备类型
  816 + if (!IotProductDeviceTypeEnum.isGatewaySub(product.getDeviceType())) {
  817 + throw exception(DEVICE_SUB_REGISTER_PRODUCT_NOT_GATEWAY_SUB, params.getProductKey());
  818 + }
  819 + // 1.3 查找设备是否已存在
  820 + // TODO @AI：存在的时候，必须父设备是自己，才返回，否则抛出业务异常；
  821 + IotDeviceDO existDevice = getSelf().getDeviceFromCache(params.getProductKey(), params.getDeviceName());
  822 + if (existDevice != null) {
  823 + // 已存在则返回设备信息
  824 + return existDevice;
  825 + }
  826 +
  827 + // 2. 创建新设备
  828 + IotDeviceSaveReqVO createReqVO = new IotDeviceSaveReqVO()
  829 + .setDeviceName(params.getDeviceName())
  830 + .setProductId(product.getId())
  831 + .setGatewayId(gatewayDevice.getId());
  832 + IotDeviceDO newDevice = createDevice0(createReqVO);
  833 + log.info("[registerSubDevice][网关({}/{}) 注册子设备({}/{})]",
  834 + gatewayDevice.getProductKey(), gatewayDevice.getDeviceName(),
  835 + newDevice.getProductKey(), newDevice.getDeviceName());
  836 + return newDevice;
  837 + }
  838 +
595 839 private IotDeviceServiceImpl getSelf() {
596 840 return SpringUtil.getBean(getClass());
597 841 }
... ...
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/message/IotDeviceMessageServiceImpl.java
  View file @bfbc352
... ... @@ -4,7 +4,6 @@ import cn.hutool.core.collection.CollUtil;
4 4 import cn.hutool.core.date.LocalDateTimeUtil;
5 5 import cn.hutool.core.lang.Assert;
6 6 import cn.hutool.core.map.MapUtil;
7   -import cn.hutool.core.util.ObjectUtil;
8 7 import cn.hutool.core.util.StrUtil;
9 8 import cn.hutool.extra.spring.SpringUtil;
10 9 import cn.iocoder.yudao.framework.common.exception.ServiceException;
... ... @@ -12,30 +11,21 @@ import cn.iocoder.yudao.framework.common.pojo.PageResult;
12 11 import cn.iocoder.yudao.framework.common.util.date.LocalDateTimeUtils;
13 12 import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
14 13 import cn.iocoder.yudao.framework.common.util.object.BeanUtils;
15   -import cn.iocoder.yudao.module.iot.controller.admin.device.vo.device.IotDeviceSaveReqVO;
16 14 import cn.iocoder.yudao.module.iot.controller.admin.device.vo.message.IotDeviceMessagePageReqVO;
17 15 import cn.iocoder.yudao.module.iot.controller.admin.statistics.vo.IotStatisticsDeviceMessageReqVO;
18 16 import cn.iocoder.yudao.module.iot.controller.admin.statistics.vo.IotStatisticsDeviceMessageSummaryByDateRespVO;
19   -import cn.iocoder.yudao.module.iot.core.biz.dto.*;
20 17 import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
21   -import cn.iocoder.yudao.module.iot.core.enums.IotDeviceStateEnum;
22 18 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
23   -import cn.iocoder.yudao.module.iot.core.topic.property.IotDevicePropertyPackPostReqDTO;
24   -import cn.iocoder.yudao.module.iot.core.topic.auth.IotSubDeviceRegisterReqDTO;
25   -import cn.iocoder.yudao.module.iot.core.topic.auth.IotSubDeviceRegisterRespDTO;
26   -import cn.iocoder.yudao.module.iot.core.topic.topo.IotDeviceTopoAddReqDTO;
27   -import cn.iocoder.yudao.module.iot.core.topic.topo.IotDeviceTopoDeleteReqDTO;
28   -import cn.iocoder.yudao.module.iot.core.topic.topo.IotDeviceTopoRespDTO;
29   -import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils;
30 19 import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer;
  20 +import cn.iocoder.yudao.module.iot.core.topic.IotDeviceIdentity;
  21 +import cn.iocoder.yudao.module.iot.core.topic.event.IotDeviceEventPostReqDTO;
  22 +import cn.iocoder.yudao.module.iot.core.topic.property.IotDevicePropertyPackPostReqDTO;
  23 +import cn.iocoder.yudao.module.iot.core.topic.property.IotDevicePropertyPostReqDTO;
31 24 import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
32 25 import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
33 26 import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceMessageDO;
34   -import cn.iocoder.yudao.module.iot.dal.dataobject.product.IotProductDO;
35 27 import cn.iocoder.yudao.module.iot.dal.tdengine.IotDeviceMessageMapper;
36   -import cn.iocoder.yudao.module.iot.enums.product.IotProductDeviceTypeEnum;
37 28 import cn.iocoder.yudao.module.iot.service.device.IotDeviceService;
38   -import cn.iocoder.yudao.module.iot.service.product.IotProductService;
39 29 import cn.iocoder.yudao.module.iot.service.device.property.IotDevicePropertyService;
40 30 import cn.iocoder.yudao.module.iot.service.ota.IotOtaTaskRecordService;
41 31 import com.baomidou.mybatisplus.core.metadata.IPage;
... ... @@ -50,11 +40,12 @@ import org.springframework.validation.annotation.Validated;
50 40  
51 41 import java.sql.Timestamp;
52 42 import java.time.LocalDateTime;
53   -import java.util.*;
  43 +import java.util.List;
  44 +import java.util.Map;
54 45  
55 46 import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
56 47 import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertList;
57   -import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.*;
  48 +import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.DEVICE_DOWNSTREAM_FAILED_SERVER_ID_NULL;
58 49  
59 50 /**
60 51 * IoT 设备消息 Service 实现类
... ... @@ -73,9 +64,6 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService {
73 64 @Resource
74 65 @Lazy // 延迟加载，避免循环依赖
75 66 private IotOtaTaskRecordService otaTaskRecordService;
76   - @Resource
77   - @Lazy // 延迟加载，避免循环依赖
78   - private IotProductService productService;
79 67  
80 68 @Resource
81 69 private IotDeviceMessageMapper deviceMessageMapper;
... ... @@ -210,11 +198,6 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService {
210 198 Assert.notEmpty(stateStr, "设备状态不能为空");
211 199 Integer state = Integer.valueOf(stateStr);
212 200 deviceService.updateDeviceState(device, state);
213   - // 特殊：网关设备下线时，网关子设备联动下线
214   - if (Objects.equal(state, IotDeviceStateEnum.OFFLINE.getState())
215   - && IotProductDeviceTypeEnum.isGateway(device.getDeviceType())) {
216   - handleGatewayOffline(device, message.getServerId());
217   - }
218 201 return null;
219 202 }
220 203  
... ... @@ -237,349 +220,31 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService {
237 220  
238 221 // 添加拓扑关系
239 222 if (Objects.equal(message.getMethod(), IotDeviceMessageMethodEnum.TOPO_ADD.getMethod())) {
240   - return handleTopoAdd(message, device);
  223 + return deviceService.handleTopoAddMessage(message, device);
241 224 }
242 225 // 删除拓扑关系
243 226 if (Objects.equal(message.getMethod(), IotDeviceMessageMethodEnum.TOPO_DELETE.getMethod())) {
244   - return handleTopoDelete(message, device);
  227 + return deviceService.handleTopoDeleteMessage(message, device);
245 228 }
246 229 // 获取拓扑关系
247 230 if (Objects.equal(message.getMethod(), IotDeviceMessageMethodEnum.TOPO_GET.getMethod())) {
248   - return handleTopoGet(device);
  231 + return deviceService.handleTopoGetMessage(device);
249 232 }
250 233  
251 234 // 子设备动态注册
252 235 if (Objects.equal(message.getMethod(), IotDeviceMessageMethodEnum.SUB_DEVICE_REGISTER.getMethod())) {
253   - return handleSubDeviceRegister(message, device);
254   - }
255   -
256   - return null;
257   - }
258   -
259   - // ========== 拓扑管理处理方法 ==========
260   -
261   - // TODO @AI：是不是更适合在 deviceService 里面处理？
262   - /**
263   - * 处理添加拓扑关系请求
264   - *
265   - * @param message 消息
266   - * @param gatewayDevice 网关设备
267   - * @return 响应数据
268   - */
269   - private Object handleTopoAdd(IotDeviceMessage message, IotDeviceDO gatewayDevice) {
270   - // 1.1 校验网关设备类型
271   - if (!IotProductDeviceTypeEnum.isGateway(gatewayDevice.getDeviceType())) {
272   - throw exception(DEVICE_NOT_GATEWAY);
273   - }
274   - // 1.2 解析参数
275   - // TODO @AI：是不是 parseObject 增加一个方法，允许传入 object 类型，避免先转 jsonString 再 parseObject ；
276   - IotDeviceTopoAddReqDTO params = JsonUtils.parseObject(JsonUtils.toJsonString(message.getParams()),
277   - IotDeviceTopoAddReqDTO.class);
278   - if (params == null || CollUtil.isEmpty(params.getSubDevices())) {
279   - throw exception(DEVICE_TOPO_PARAMS_INVALID);
280   - }
281   -
282   - // 2. 遍历处理每个子设备
283   - // TODO @AI：processTopoAddSubDevice 不要抽成小方法；
284   - List<IotDeviceDO> addedSubDevices = new ArrayList<>(params.getSubDevices().size());
285   - for (IotDeviceAuthReqDTO subDeviceAuth : params.getSubDevices()) {
286   - try {
287   - IotDeviceDO subDevice = processTopoAddSubDevice(subDeviceAuth, gatewayDevice);
288   - addedSubDevices.add(subDevice);
289   - } catch (Exception ex) {
290   - log.warn("[handleTopoAdd][网关({}/{}) 添加子设备失败，username={}]",
291   - gatewayDevice.getProductKey(), gatewayDevice.getDeviceName(),
292   - subDeviceAuth.getUsername(), ex);
293   - }
294   - }
295   - // TODO @AI：http://help.aliyun.com/zh/marketplace/add-topological-relationship 要回复的！
296   -
297   - // 3. 发送拓扑变更通知
298   - // TODO @AI：这里不应该发，它更多发生在，管理后台改动后，主动下发；http://help.aliyun.com/zh/marketplace/notify-gateway-topology-changes
299   - for (IotDeviceDO subDevice : addedSubDevices) {
300   - sendTopoChangeNotify(gatewayDevice, "add", subDevice);
301   - }
302   - return null;
303   - }
304   -
305   - /**
306   - * 处理单个子设备的拓扑添加
307   - *
308   - * @param subDeviceAuth 子设备认证信息
309   - * @param gatewayDevice 网关设备
310   - * @return 添加成功的子设备，失败返回 null
311   - */
312   - private IotDeviceDO processTopoAddSubDevice(IotDeviceAuthReqDTO subDeviceAuth, IotDeviceDO gatewayDevice) {
313   - // 1.1 解析子设备信息
314   - IotDeviceAuthUtils.DeviceInfo subDeviceInfo = IotDeviceAuthUtils.parseUsername(subDeviceAuth.getUsername());
315   - if (subDeviceInfo == null) {
316   - throw exception(DEVICE_TOPO_SUB_DEVICE_USERNAME_INVALID);
317   - }
318   - // 1.2 校验子设备认证信息
319   - if (!deviceService.authDevice(subDeviceAuth)) {
320   - throw exception(DEVICE_TOPO_SUB_DEVICE_AUTH_FAILED);
321   - }
322   -
323   - // 1.3 获取子设备
324   - IotDeviceDO subDevice = deviceService.getDeviceFromCache(subDeviceInfo.getProductKey(), subDeviceInfo.getDeviceName());
325   - if (subDevice == null) {
326   - throw exception(DEVICE_NOT_EXISTS);
327   - }
328   - // 1.4 校验子设备类型
329   - if (!IotProductDeviceTypeEnum.isGatewaySub(subDevice.getDeviceType())) {
330   - throw exception(DEVICE_NOT_GATEWAY_SUB, subDevice.getProductKey(), subDevice.getDeviceName());
331   - }
332   - // 1.5 校验子设备是否已绑定到其他网关
333   - if (subDevice.getGatewayId() != null
334   - && ObjectUtil.notEqual(subDevice.getGatewayId(), gatewayDevice.getId())) {
335   - throw exception(DEVICE_GATEWAY_BINDTO_EXISTS, subDevice.getProductKey(), subDevice.getDeviceName());
336   - }
337   -
338   - // 2. 绑定拓扑关系
339   - // TODO @AI：上面的校验，貌似和 bindDeviceGateway 里的，有点重复；
340   - deviceService.bindDeviceGateway(Collections.singletonList(subDevice.getId()), gatewayDevice.getId());
341   - log.info("[processTopoAddSubDevice][网关({}/{}) 绑定子设备({}/{})]",
342   - gatewayDevice.getProductKey(), gatewayDevice.getDeviceName(),
343   - subDevice.getProductKey(), subDevice.getDeviceName());
344   - return subDevice;
345   - }
346   -
347   - // TODO @AI：是不是更适合在 deviceService 里面处理？
348   -
349   - /**
350   - * 处理删除拓扑关系请求
351   - *
352   - * @param message 消息
353   - * @param gatewayDevice 网关设备
354   - * @return 响应数据
355   - */
356   - private Object handleTopoDelete(IotDeviceMessage message, IotDeviceDO gatewayDevice) {
357   - // 1. 校验网关设备类型
358   - if (!IotProductDeviceTypeEnum.isGateway(gatewayDevice.getDeviceType())) {
359   - throw exception(DEVICE_NOT_GATEWAY);
360   - }
361   -
362   - // 2. 解析参数（数组格式）
363   - IotDeviceTopoDeleteReqDTO params = JsonUtils.parseObject(JsonUtils.toJsonString(message.getParams()),
364   - IotDeviceTopoDeleteReqDTO.class);
365   - if (params == null || params.getSubDevices() == null || params.getSubDevices().isEmpty()) {
366   - throw exception(DEVICE_TOPO_PARAMS_INVALID);
  236 + return deviceService.handleSubDeviceRegisterMessage(message, device);
367 237 }
368 238  
369   - // 3. 遍历处理每个子设备
370   - List<IotDeviceDO> deletedSubDevices = new ArrayList<>();
371   - for (IotDeviceTopoDeleteReqDTO.SubDevice subDeviceInfo : params.getSubDevices()) {
372   - try {
373   - IotDeviceDO subDevice = processTopoDeleteSubDevice(subDeviceInfo, gatewayDevice);
374   - deletedSubDevices.add(subDevice);
375   - } catch (Exception ex) {
376   - log.warn("[handleTopoDelete][网关({}/{}) 删除子设备失败，productKey={}, deviceName={}]",
377   - gatewayDevice.getProductKey(), gatewayDevice.getDeviceName(),
378   - subDeviceInfo.getProductKey(), subDeviceInfo.getDeviceName(), ex);
379   - }
380   - }
381   -
382   - // 4. 发送拓扑变更通知
383   - for (IotDeviceDO subDevice : deletedSubDevices) {
384   - sendTopoChangeNotify(gatewayDevice, "delete", subDevice);
385   - }
386 239 return null;
387 240 }
388 241  
389   - // TODO @AI：是不是更适合在 deviceService 里面处理？
390   -
391   - /**
392   - * 处理单个子设备的拓扑删除
393   - *
394   - * @param subDeviceInfo 子设备标识
395   - * @param gatewayDevice 网关设备
396   - * @return 删除成功的子设备，失败返回 null
397   - */
398   - private IotDeviceDO processTopoDeleteSubDevice(IotDeviceTopoDeleteReqDTO.SubDevice subDeviceInfo,
399   - IotDeviceDO gatewayDevice) {
400   - // 1. 获取子设备
401   - IotDeviceDO subDevice = deviceService.getDeviceFromCache(subDeviceInfo.getProductKey(), subDeviceInfo.getDeviceName());
402   - if (subDevice == null) {
403   - throw exception(DEVICE_NOT_EXISTS);
404   - }
405   -
406   - // 2. 校验子设备是否绑定到该网关
407   - if (!Objects.equal(subDevice.getGatewayId(), gatewayDevice.getId())) {
408   - throw exception(DEVICE_TOPO_SUB_NOT_BINDTO_GATEWAY, subDeviceInfo.getProductKey(), subDeviceInfo.getDeviceName());
409   - }
410   -
411   - // 3. 解绑拓扑关系
412   - deviceService.unbindDeviceGateway(Collections.singletonList(subDevice.getId()));
413   - log.info("[processTopoDeleteSubDevice][网关({}/{}) 解绑子设备({}/{})]",
414   - gatewayDevice.getProductKey(), gatewayDevice.getDeviceName(),
415   - subDevice.getProductKey(), subDevice.getDeviceName());
416   -
417   - // 4. 子设备下线
418   - if (Objects.equal(subDevice.getState(), IotDeviceStateEnum.ONLINE.getState())) {
419   - deviceService.updateDeviceState(subDevice, IotDeviceStateEnum.OFFLINE.getState());
420   - }
421   -
422   - return subDevice;
423   - }
424   -
425   - // TODO @AI：是不是更适合在 deviceService 里面处理？
426   - /**
427   - * 处理获取拓扑关系请求
428   - *
429   - * @param gatewayDevice 网关设备
430   - * @return 子设备列表
431   - */
432   - private Object handleTopoGet(IotDeviceDO gatewayDevice) {
433   - // 1. 校验网关设备类型
434   - if (!IotProductDeviceTypeEnum.isGateway(gatewayDevice.getDeviceType())) {
435   - throw exception(DEVICE_NOT_GATEWAY);
436   - }
437   -
438   - // 2. 获取子设备列表
439   - List<IotDeviceDO> subDevices = deviceService.getDeviceListByGatewayId(gatewayDevice.getId());
440   -
441   - // 3. 转换为响应格式
442   - return convertList(subDevices, subDevice -> new IotDeviceTopoRespDTO()
443   - .setProductKey(subDevice.getProductKey())
444   - .setDeviceName(subDevice.getDeviceName()));
445   - }
446   -
447   - // TODO @AI：是不是更适合在 deviceService 里面处理？
448   - /**
449   - * 发送拓扑变更通知
450   - *
451   - * @param gatewayDevice 网关设备
452   - * @param changeType 变更类型：add/delete
453   - * @param subDevice 子设备
454   - */
455   - private void sendTopoChangeNotify(IotDeviceDO gatewayDevice, String changeType, IotDeviceDO subDevice) {
456   - try {
457   - String serverId = devicePropertyService.getDeviceServerId(gatewayDevice.getId());
458   - if (StrUtil.isEmpty(serverId)) {
459   - log.warn("[sendTopoChangeNotify][网关({}/{}) serverId 为空，无法发送拓扑变更通知]",
460   - gatewayDevice.getProductKey(), gatewayDevice.getDeviceName());
461   - return;
462   - }
463   -
464   - Map<String, Object> params = MapUtil.builder(new HashMap<String, Object>())
465   - .put("changeType", changeType)
466   - .put("subDevice", MapUtil.builder(new HashMap<String, Object>())
467   - .put("productKey", subDevice.getProductKey())
468   - .put("deviceName", subDevice.getDeviceName())
469   - .build())
470   - .build();
471   -
472   - IotDeviceMessage notifyMessage = IotDeviceMessage.requestOf(
473   - IotDeviceMessageMethodEnum.TOPO_CHANGE.getMethod(), params);
474   - sendDeviceMessage(notifyMessage, gatewayDevice, serverId);
475   - } catch (Exception ex) {
476   - log.error("[sendTopoChangeNotify][发送拓扑变更通知失败，网关({}/{}), 子设备({}/{})]",
477   - gatewayDevice.getProductKey(), gatewayDevice.getDeviceName(),
478   - subDevice.getProductKey(), subDevice.getDeviceName(), ex);
479   - }
480   - }
481   -
482   - // ========== 子设备注册处理方法 ==========
483   -
484   - // TODO @AI：是不是更适合在 deviceService 里面处理？
485   - /**
486   - * 处理子设备动态注册请求
487   - *
488   - * @param message 消息
489   - * @param gatewayDevice 网关设备
490   - * @return 注册结果列表
491   - */
492   - private Object handleSubDeviceRegister(IotDeviceMessage message, IotDeviceDO gatewayDevice) {
493   - // 1. 校验网关设备类型
494   - if (!IotProductDeviceTypeEnum.isGateway(gatewayDevice.getDeviceType())) {
495   - throw exception(DEVICE_NOT_GATEWAY);
496   - }
497   -
498   - // 2. 解析参数（数组）
499   - List<IotSubDeviceRegisterReqDTO> paramsList;
500   - if (message.getParams() instanceof List) {
501   - paramsList = JsonUtils.parseArray(JsonUtils.toJsonString(message.getParams()),
502   - IotSubDeviceRegisterReqDTO.class);
503   - } else {
504   - throw exception(DEVICE_SUB_REGISTER_PARAMS_INVALID);
505   - }
506   -
507   - if (paramsList == null || paramsList.isEmpty()) {
508   - throw exception(DEVICE_SUB_REGISTER_PARAMS_INVALID);
509   - }
510   -
511   - // 3. 遍历注册每个子设备
512   - List<IotSubDeviceRegisterRespDTO> results = new ArrayList<>();
513   - for (IotSubDeviceRegisterReqDTO params : paramsList) {
514   - try {
515   - IotSubDeviceRegisterRespDTO result = registerSubDevice(params, gatewayDevice);
516   - results.add(result);
517   - } catch (Exception ex) {
518   - log.error("[handleSubDeviceRegister][子设备({}/{}) 注册失败]",
519   - params.getProductKey(), params.getDeviceName(), ex);
520   - // 继续处理其他子设备，不影响整体流程
521   - }
522   - }
523   -
524   - return results;
525   - }
526   -
527   - // TODO @AI：是不是更适合在 deviceService 里面处理？
528   - /**
529   - * 注册单个子设备
530   - *
531   - * @param params 注册参数
532   - * @param gatewayDevice 网关设备
533   - * @return 注册结果
534   - */
535   - private IotSubDeviceRegisterRespDTO registerSubDevice(IotSubDeviceRegisterReqDTO params,
536   - IotDeviceDO gatewayDevice) {
537   - // 1. 查找产品
538   - IotProductDO product = productService.getProductByProductKey(params.getProductKey());
539   - if (product == null) {
540   - throw exception(PRODUCT_NOT_EXISTS);
541   - }
542   -
543   - // 2. 校验产品是否为网关子设备类型
544   - if (!IotProductDeviceTypeEnum.isGatewaySub(product.getDeviceType())) {
545   - throw exception(DEVICE_SUB_REGISTER_PRODUCT_NOT_GATEWAY_SUB, params.getProductKey());
546   - }
547   -
548   - // 3. 查找设备是否已存在
549   - IotDeviceDO existDevice = deviceService.getDeviceFromCache(params.getProductKey(), params.getDeviceName());
550   - if (existDevice != null) {
551   - // 已存在则返回设备信息
552   - return new IotSubDeviceRegisterRespDTO()
553   - .setProductKey(existDevice.getProductKey())
554   - .setDeviceName(existDevice.getDeviceName())
555   - .setDeviceSecret(existDevice.getDeviceSecret());
556   - }
557   -
558   - // 4. 创建新设备
559   - IotDeviceSaveReqVO createReqVO = new IotDeviceSaveReqVO()
560   - .setDeviceName(params.getDeviceName())
561   - .setProductId(product.getId())
562   - .setGatewayId(gatewayDevice.getId());
563   - Long deviceId = deviceService.createDevice(createReqVO);
564   -
565   - // 5. 获取新创建的设备信息
566   - IotDeviceDO newDevice = deviceService.getDevice(deviceId);
567   - log.info("[registerSubDevice][网关({}/{}) 注册子设备({}/{})]",
568   - gatewayDevice.getProductKey(), gatewayDevice.getDeviceName(),
569   - newDevice.getProductKey(), newDevice.getDeviceName());
570   -
571   - return new IotSubDeviceRegisterRespDTO()
572   - .setProductKey(newDevice.getProductKey())
573   - .setDeviceName(newDevice.getDeviceName())
574   - .setDeviceSecret(newDevice.getDeviceSecret());
575   - }
576   -
577 242 // ========== 批量上报处理方法 ==========
578 243  
579 244 /**
580 245 * 处理批量上报消息
581 246 * <p>
582   - * 将 pack 消息拆分成多条标准消息，递归处理
  247 + * 将 pack 消息拆分成多条标准消息，发送到 MQ 让规则引擎处理
583 248 *
584 249 * @param packMessage 批量消息
585 250 * @param gatewayDevice 网关设备
... ... @@ -590,135 +255,72 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService {
590 255 JsonUtils.toJsonString(packMessage.getParams()),
591 256 IotDevicePropertyPackPostReqDTO.class);
592 257 if (params == null) {
593   - log.warn("[handlePackMessage][消息({}) 参数解析失败]", packMessage.getId());
  258 + log.warn("[handlePackMessage][消息({}) 参数解析失败]", packMessage);
594 259 return;
595 260 }
596 261  
597   - // 2. 处理网关自身属性
598   - // TODO @AI：是不是经过总线会更好：
599   - // TODO @AI：是不是少处理了 event 事件？
600   - if (params.getProperties() != null && !params.getProperties().isEmpty()) {
601   - Map<String, Object> gatewayProperties = convertPackProperties(params.getProperties());
602   - IotDeviceMessage gatewayMsg = IotDeviceMessage.builder()
603   - .id(IotDeviceMessageUtils.generateMessageId())
604   - .deviceId(gatewayDevice.getId())
605   - .tenantId(gatewayDevice.getTenantId())
606   - .serverId(packMessage.getServerId())
607   - .method(IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod())
608   - .params(gatewayProperties)
609   - .reportTime(LocalDateTime.now())
610   - .build();
611   - // 直接调用处理，不通过消息总线
612   - try {
613   - devicePropertyService.saveDeviceProperty(gatewayDevice, gatewayMsg);
614   - getSelf().createDeviceLogAsync(gatewayMsg);
615   - } catch (Exception ex) {
616   - log.error("[handlePackMessage][网关({}) 属性处理失败]", gatewayDevice.getId(), ex);
617   - }
618   - }
  262 + // 2. 处理网关设备（自身）的数据
  263 + sendDevicePackData(gatewayDevice, packMessage.getServerId(), params.getProperties(), params.getEvents());
619 264  
620   - // 3. 处理子设备数据
621   - if (params.getSubDevices() != null) {
622   - for (IotDevicePropertyPackPostReqDTO.SubDeviceData subDeviceData : params.getSubDevices()) {
623   - try {
624   - handleSubDevicePackData(packMessage, subDeviceData);
625   - } catch (Exception ex) {
626   - log.error("[handlePackMessage][子设备({}/{}) 数据处理失败]",
627   - subDeviceData.getIdentity().getProductKey(),
628   - subDeviceData.getIdentity().getDeviceName(), ex);
629   - }
630   - }
631   - }
632   - }
633   -
634   - /**
635   - * 处理子设备的 pack 数据
636   - *
637   - * @param packMessage 原始 pack 消息
638   - * @param subDeviceData 子设备数据
639   - */
640   - private void handleSubDevicePackData(IotDeviceMessage packMessage,
641   - IotDevicePropertyPackPostReqDTO.SubDeviceData subDeviceData) {
642   - // 1. 获取子设备
643   - IotDevicePropertyPackPostReqDTO.DeviceIdentity identity = subDeviceData.getIdentity();
644   - IotDeviceDO subDevice = deviceService.getDeviceFromCache(identity.getProductKey(), identity.getDeviceName());
645   - if (subDevice == null) {
646   - log.warn("[handleSubDevicePackData][子设备({}/{}) 不存在]",
647   - identity.getProductKey(), identity.getDeviceName());
  265 + // 3. 处理子设备的数据
  266 + if (CollUtil.isEmpty(params.getSubDevices())) {
648 267 return;
649 268 }
650   -
651   - // 2. 处理子设备属性
652   - if (subDeviceData.getProperties() != null && !subDeviceData.getProperties().isEmpty()) {
653   - Map<String, Object> properties = convertPackProperties(subDeviceData.getProperties());
654   - IotDeviceMessage subMsg = IotDeviceMessage.builder()
655   - .id(IotDeviceMessageUtils.generateMessageId())
656   - .deviceId(subDevice.getId())
657   - .tenantId(subDevice.getTenantId())
658   - .serverId(packMessage.getServerId())
659   - .method(IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod())
660   - .params(properties)
661   - .reportTime(LocalDateTime.now())
662   - .build();
663   - devicePropertyService.saveDeviceProperty(subDevice, subMsg);
664   - getSelf().createDeviceLogAsync(subMsg);
665   - }
666   -
667   - // 3. 处理子设备事件
668   - // TODO @AI：事件处理可以后续扩展
669   - }
670   -
671   - /**
672   - * 转换 pack 属性格式为标准属性格式
673   - * <p>
674   - * pack 格式：{"temperature": {"value": 25.5, "time": 1524448722000}}
675   - * 标准格式：{"temperature": 25.5}
676   - *
677   - * @param packProperties pack 属性
678   - * @return 标准属性
679   - */
680   - private Map<String, Object> convertPackProperties(Map<String, IotDevicePropertyPackPostReqDTO.PropertyValue> packProperties) {
681   - Map<String, Object> result = new HashMap<>();
682   - for (Map.Entry<String, IotDevicePropertyPackPostReqDTO.PropertyValue> entry : packProperties.entrySet()) {
683   - if (entry.getValue() != null) {
684   - result.put(entry.getKey(), entry.getValue().getValue());
  269 + for (IotDevicePropertyPackPostReqDTO.SubDeviceData subDeviceData : params.getSubDevices()) {
  270 + try {
  271 + IotDeviceIdentity identity = subDeviceData.getIdentity();
  272 + IotDeviceDO subDevice = deviceService.getDeviceFromCache(identity.getProductKey(), identity.getDeviceName());
  273 + if (subDevice == null) {
  274 + log.warn("[handlePackMessage][子设备({}/{}) 不存在]", identity.getProductKey(), identity.getDeviceName());
  275 + return;
  276 + }
  277 + sendDevicePackData(subDevice, packMessage.getServerId(), subDeviceData.getProperties(), subDeviceData.getEvents());
  278 + } catch (Exception ex) {
  279 + log.error("[handlePackMessage][子设备({}/{}) 数据处理失败]", subDeviceData.getIdentity().getProductKey(),
  280 + subDeviceData.getIdentity().getDeviceName(), ex);
685 281 }
686 282 }
687   - return result;
688 283 }
689 284  
690   - // ========== 网关下线联动处理 ==========
691   -
692   - // TODO 芋艿：是不是写到 deviceService 里更合适？更解耦。
693 285 /**
694   - * 处理网关下线，联动所有子设备下线
  286 + * 发送设备 pack 数据到 MQ（属性 + 事件）
695 287 *
696   - * @param gatewayDevice 网关设备
697   - * @param serverId 服务标识
  288 + * @param device 设备
  289 + * @param serverId 服务标识
  290 + * @param properties 属性数据
  291 + * @param events 事件数据
698 292 */
699   - private void handleGatewayOffline(IotDeviceDO gatewayDevice, String serverId) {
700   - // 1. 获取网关下所有子设备
701   - List<IotDeviceDO> subDevices = deviceService.getDeviceListByGatewayId(gatewayDevice.getId());
702   - if (subDevices == null || subDevices.isEmpty()) {
703   - return;
704   - }
705   -
706   - // 2. 将在线的子设备设置为下线
707   - for (IotDeviceDO subDevice : subDevices) {
708   - if (Objects.equal(subDevice.getState(), IotDeviceStateEnum.ONLINE.getState())) {
709   - try {
710   - deviceService.updateDeviceState(subDevice, IotDeviceStateEnum.OFFLINE.getState());
711   - log.info("[handleGatewayOffline][网关({}/{}) 下线，子设备({}/{}) 联动下线]",
712   - gatewayDevice.getProductKey(), gatewayDevice.getDeviceName(),
713   - subDevice.getProductKey(), subDevice.getDeviceName());
714   - } catch (Exception ex) {
715   - log.error("[handleGatewayOffline][子设备({}/{}) 下线失败]",
716   - subDevice.getProductKey(), subDevice.getDeviceName(), ex);
  293 + private void sendDevicePackData(IotDeviceDO device, String serverId,
  294 + Map<String, Object> properties,
  295 + Map<String, IotDevicePropertyPackPostReqDTO.EventValue> events) {
  296 + // 1. 发送属性消息
  297 + if (MapUtil.isNotEmpty(properties)) {
  298 + IotDeviceMessage propertyMsg = IotDeviceMessage.requestOf(
  299 + device.getId(), device.getTenantId(), serverId,
  300 + IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod(),
  301 + IotDevicePropertyPostReqDTO.of(properties));
  302 + deviceMessageProducer.sendDeviceMessage(propertyMsg);
  303 + }
  304 +
  305 + // 2. 发送事件消息
  306 + if (MapUtil.isNotEmpty(events)) {
  307 + for (Map.Entry<String, IotDevicePropertyPackPostReqDTO.EventValue> eventEntry : events.entrySet()) {
  308 + String eventId = eventEntry.getKey();
  309 + IotDevicePropertyPackPostReqDTO.EventValue eventValue = eventEntry.getValue();
  310 + if (eventValue == null) {
  311 + continue;
717 312 }
  313 + IotDeviceMessage eventMsg = IotDeviceMessage.requestOf(
  314 + device.getId(), device.getTenantId(), serverId,
  315 + IotDeviceMessageMethodEnum.EVENT_POST.getMethod(),
  316 + IotDeviceEventPostReqDTO.of(eventId, eventValue.getValue(), eventValue.getTime()));
  317 + deviceMessageProducer.sendDeviceMessage(eventMsg);
718 318 }
719 319 }
720 320 }
721 321  
  322 + // ========= 设备消息查询 ==========
  323 +
722 324 @Override
723 325 public PageResult<IotDeviceMessageDO> getDeviceMessagePage(IotDeviceMessagePageReqVO pageReqVO) {
724 326 try {
... ...
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/mq/message/IotDeviceMessage.java
  View file @bfbc352
... ... @@ -108,6 +108,23 @@ public class IotDeviceMessage {
108 108 return of(requestId, method, params, null, null, null);
109 109 }
110 110  
  111 + /**
  112 + * 创建设备请求消息（包含设备信息）
  113 + *
  114 + * @param deviceId 设备编号
  115 + * @param tenantId 租户编号
  116 + * @param serverId 服务标识
  117 + * @param method 消息方法
  118 + * @param params 消息参数
  119 + * @return 消息对象
  120 + */
  121 + public static IotDeviceMessage requestOf(Long deviceId, Long tenantId, String serverId,
  122 + String method, Object params) {
  123 + IotDeviceMessage message = of(null, method, params, null, null, null);
  124 + return message.setId(IotDeviceMessageUtils.generateMessageId())
  125 + .setDeviceId(deviceId).setTenantId(tenantId).setServerId(serverId);
  126 + }
  127 +
111 128 public static IotDeviceMessage replyOf(String requestId, String method,
112 129 Object data, Integer code, String msg) {
113 130 if (code == null) {
... ...
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/IotDeviceIdentify.java deleted
View file @18ed7b5
1   -package cn.iocoder.yudao.module.iot.core.topic;
2   -
3   -// TODO @AI：增加 productKey、DeviceName
4   -// TODO @AI：有更合适的类名么？？？
5   -public class IotDeviceIdentify {
6   -}
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/IotDeviceIdentity.java 0 → 100644
  View file @bfbc352
  1 +package cn.iocoder.yudao.module.iot.core.topic;
  2 +
  3 +import jakarta.validation.constraints.NotEmpty;
  4 +import lombok.AllArgsConstructor;
  5 +import lombok.Data;
  6 +import lombok.NoArgsConstructor;
  7 +
  8 +/**
  9 + * IoT 设备标识
  10 + *
  11 + * 用于标识一个设备的基本信息（productKey + deviceName）
  12 + *
  13 + * @author 芋道源码
  14 + */
  15 +@Data
  16 +@NoArgsConstructor
  17 +@AllArgsConstructor
  18 +public class IotDeviceIdentity {
  19 +
  20 + /**
  21 + * 产品标识
  22 + */
  23 + @NotEmpty(message = "产品标识不能为空")
  24 + private String productKey;
  25 +
  26 + /**
  27 + * 设备名称
  28 + */
  29 + @NotEmpty(message = "设备名称不能为空")
  30 + private String deviceName;
  31 +
  32 +}
... ...
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/auth/IotSubDeviceRegisterRespDTO.java
  View file @bfbc352
1 1 package cn.iocoder.yudao.module.iot.core.topic.auth;
2 2  
  3 +import lombok.AllArgsConstructor;
3 4 import lombok.Data;
  5 +import lombok.NoArgsConstructor;
4 6  
5 7 /**
6 8 * IoT 子设备动态注册 Response DTO
... ... @@ -11,6 +13,8 @@ import lombok.Data;
11 13 * @see <a href="http://help.aliyun.com/zh/marketplace/dynamic-registration-of-sub-devices">阿里云 - 动态注册子设备</a>
12 14 */
13 15 @Data
  16 +@NoArgsConstructor
  17 +@AllArgsConstructor
14 18 public class IotSubDeviceRegisterRespDTO {
15 19  
16 20 /**
... ...
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/event/IotDeviceEventPostReqDTO.java 0 → 100644
  View file @bfbc352
  1 +package cn.iocoder.yudao.module.iot.core.topic.event;
  2 +
  3 +import lombok.Data;
  4 +
  5 +/**
  6 + * IoT 设备事件上报 Request DTO
  7 + * <p>
  8 + * 用于 thing.event.{eventId}.post 消息的 params 参数
  9 + *
  10 + * @author 芋道源码
  11 + * @see <a href="https://help.aliyun.com/zh/iot/user-guide/device-properties-events-and-services">阿里云 - 设备上报事件</a>
  12 + */
  13 +@Data
  14 +public class IotDeviceEventPostReqDTO {
  15 +
  16 + /**
  17 + * 事件标识符
  18 + */
  19 + private String eventId;
  20 +
  21 + /**
  22 + * 事件输出参数
  23 + */
  24 + private Object value;
  25 +
  26 + /**
  27 + * 上报时间（毫秒时间戳，可选）
  28 + */
  29 + private Long time;
  30 +
  31 + /**
  32 + * 创建事件上报 DTO
  33 + *
  34 + * @param eventId 事件标识符
  35 + * @param value 事件值
  36 + * @return DTO 对象
  37 + */
  38 + public static IotDeviceEventPostReqDTO of(String eventId, Object value) {
  39 + return of(eventId, value, null);
  40 + }
  41 +
  42 + /**
  43 + * 创建事件上报 DTO（带时间）
  44 + *
  45 + * @param eventId 事件标识符
  46 + * @param value 事件值
  47 + * @param time 上报时间
  48 + * @return DTO 对象
  49 + */
  50 + public static IotDeviceEventPostReqDTO of(String eventId, Object value, Long time) {
  51 + return new IotDeviceEventPostReqDTO()
  52 + .setEventId(eventId)
  53 + .setValue(value)
  54 + .setTime(time);
  55 + }
  56 +
  57 +}
... ...
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/property/IotDevicePropertyPackPostReqDTO.java
  View file @bfbc352
1 1 package cn.iocoder.yudao.module.iot.core.topic.property;
2 2  
  3 +import cn.iocoder.yudao.module.iot.core.topic.IotDeviceIdentity;
3 4 import lombok.Data;
4   -import lombok.experimental.Accessors;
5 5  
6 6 import java.util.List;
7 7 import java.util.Map;
... ... @@ -17,14 +17,13 @@ import java.util.Map;
17 17 @Data
18 18 public class IotDevicePropertyPackPostReqDTO {
19 19  
20   - // TODO @AI：不用 PropertyValue，直接使用 Object 接收就行！
21 20 /**
22 21 * 网关自身属性
23 22 * <p>
24 23 * key: 属性标识符
25   - * value: 属性值对象（包含 value 和 time）
  24 + * value: 属性值
26 25 */
27   - private Map<String, PropertyValue> properties;
  26 + private Map<String, Object> properties;
28 27  
29 28 /**
30 29 * 网关自身事件
... ... @@ -40,24 +39,6 @@ public class IotDevicePropertyPackPostReqDTO {
40 39 private List<SubDeviceData> subDevices;
41 40  
42 41 /**
43   - * 属性值对象
44   - */
45   - @Data
46   - public static class PropertyValue {
47   -
48   - /**
49   - * 属性值
50   - */
51   - private Object value;
52   -
53   - /**
54   - * 上报时间（毫秒时间戳）
55   - */
56   - private Long time;
57   -
58   - }
59   -
60   - /**
61 42 * 事件值对象
62 43 */
63 44 @Data
... ... @@ -84,37 +65,24 @@ public class IotDevicePropertyPackPostReqDTO {
84 65 /**
85 66 * 子设备标识
86 67 */
87   - private DeviceIdentity identity;
  68 + private IotDeviceIdentity identity;
88 69  
89 70 /**
90 71 * 子设备属性
  72 + * <p>
  73 + * key: 属性标识符
  74 + * value: 属性值
91 75 */
92   - private Map<String, PropertyValue> properties;
  76 + private Map<String, Object> properties;
93 77  
94 78 /**
95 79 * 子设备事件
  80 + * <p>
  81 + * key: 事件标识符
  82 + * value: 事件值对象（包含 value 和 time）
96 83 */
97 84 private Map<String, EventValue> events;
98 85  
99 86 }
100 87  
101   - /**
102   - * 设备标识
103   - */
104   - @Data
105   - @Accessors(chain = true)
106   - public static class DeviceIdentity {
107   -
108   - /**
109   - * 产品标识
110   - */
111   - private String productKey;
112   -
113   - /**
114   - * 设备名称
115   - */
116   - private String deviceName;
117   -
118   - }
119   -
120 88 }
... ...
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/property/IotDevicePropertyPostReqDTO.java 0 → 100644
  View file @bfbc352
  1 +package cn.iocoder.yudao.module.iot.core.topic.property;
  2 +
  3 +import java.util.HashMap;
  4 +import java.util.Map;
  5 +
  6 +/**
  7 + * IoT 设备属性上报 Request DTO
  8 + * <p>
  9 + * 用于 thing.property.post 消息的 params 参数
  10 + * <p>
  11 + * 本质是一个 Map，key 为属性标识符，value 为属性值
  12 + *
  13 + * @author 芋道源码
  14 + * @see <a href="https://help.aliyun.com/zh/iot/user-guide/device-properties-events-and-services">阿里云 - 设备上报属性</a>
  15 + */
  16 +public class IotDevicePropertyPostReqDTO extends HashMap<String, Object> {
  17 +
  18 + public IotDevicePropertyPostReqDTO() {
  19 + super();
  20 + }
  21 +
  22 + public IotDevicePropertyPostReqDTO(Map<String, Object> properties) {
  23 + super(properties);
  24 + }
  25 +
  26 + /**
  27 + * 创建属性上报 DTO
  28 + *
  29 + * @param properties 属性数据
  30 + * @return DTO 对象
  31 + */
  32 + public static IotDevicePropertyPostReqDTO of(Map<String, Object> properties) {
  33 + return new IotDevicePropertyPostReqDTO(properties);
  34 + }
  35 +
  36 +}
... ...
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/topo/IotDeviceTopoDeleteReqDTO.java
  View file @bfbc352
1 1 package cn.iocoder.yudao.module.iot.core.topic.topo;
2 2  
  3 +import cn.iocoder.yudao.module.iot.core.topic.IotDeviceIdentity;
  4 +import jakarta.validation.Valid;
3 5 import jakarta.validation.constraints.NotEmpty;
4 6 import lombok.Data;
5 7  
... ... @@ -16,30 +18,11 @@ import java.util.List;
16 18 @Data
17 19 public class IotDeviceTopoDeleteReqDTO {
18 20  
19   - // TODO @AI：应该是数组；IotDeviceIdentify
20 21 /**
21 22 * 子设备标识列表
22 23 */
23   - private List<SubDevice> subDevices;
24   -
25   - /**
26   - * 子设备标识
27   - */
28   - @Data
29   - public static class SubDevice {
30   -
31   - /**
32   - * 子设备 ProductKey
33   - */
34   - @NotEmpty(message = "产品标识不能为空")
35   - private String productKey;
36   -
37   - /**
38   - * 子设备 DeviceName
39   - */
40   - @NotEmpty(message = "设备名称不能为空")
41   - private String deviceName;
42   -
43   - }
  24 + @Valid
  25 + @NotEmpty(message = "子设备标识列表不能为空")
  26 + private List<IotDeviceIdentity> subDevices;
44 27  
45 28 }
... ...
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/topo/IotDeviceTopoGetReqDTO.java 0 → 100644
  View file @bfbc352
  1 +package cn.iocoder.yudao.module.iot.core.topic.topo;
  2 +
  3 +import lombok.Data;
  4 +
  5 +/**
  6 + * IoT 设备拓扑关系获取 Request DTO
  7 + * <p>
  8 + * 用于 thing.topo.get 请求的 params 参数（目前为空，预留扩展）
  9 + *
  10 + * @author 芋道源码
  11 + * @see <a href="https://help.aliyun.com/zh/marketplace/obtain-topological-relationship">阿里云 - 获取拓扑关系</a>
  12 + */
  13 +@Data
  14 +public class IotDeviceTopoGetReqDTO {
  15 +
  16 +}
... ...
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/topo/IotDeviceTopoRespDTO.java renamed to yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/topo/IotDeviceTopoGetRespDTO.java
  View file @bfbc352
1 1 package cn.iocoder.yudao.module.iot.core.topic.topo;
2 2  
  3 +import cn.iocoder.yudao.module.iot.core.topic.IotDeviceIdentity;
3 4 import lombok.Data;
4 5  
5   -// TODO @AI：IotDeviceTopoGetRespDTO
  6 +import java.util.List;
  7 +
6 8 /**
7   - * IoT 设备拓扑关系 Response DTO
  9 + * IoT 设备拓扑关系获取 Response DTO
8 10 * <p>
9   - * 用于 thing.topo.get 响应的子设备信息
  11 + * 用于 thing.topo.get 响应
10 12 *
11 13 * @author 芋道源码
12 14 * @see <a href="https://help.aliyun.com/zh/marketplace/obtain-topological-relationship">阿里云 - 获取拓扑关系</a>
13 15 */
14 16 @Data
15   -public class IotDeviceTopoRespDTO {
16   -
17   - // TODO @AI：应该是数组；IotDeviceIdentify
18   - /**
19   - * 子设备 ProductKey
20   - */
21   - private String productKey;
  17 +public class IotDeviceTopoGetRespDTO {
22 18  
23 19 /**
24   - * 子设备 DeviceName
  20 + * 子设备列表
25 21 */
26   - private String deviceName;
  22 + private List<IotDeviceIdentity> subDevices;
27 23  
28 24 }
... ...