reactor：【IoT 物联网】调整数据流转的包结构

yudao-module-iot/yudao-module-iot-api/src/main/java/cn/iocoder/yudao/module/iot/enums/rule/IotDataBridgeDirectionEnum.java

@@ -1,30 +0,0 @@
-package cn.iocoder.yudao.module.iot.enums.rule;
-
-import cn.iocoder.yudao.framework.common.core.ArrayValuable;
-import lombok.Getter;
-import lombok.RequiredArgsConstructor;
-
-import java.util.Arrays;
-
-/**
- * IoT 数据桥接的方向枚举
- *
- * @author 芋道源码
- */
-@RequiredArgsConstructor
-@Getter
-public enum IotDataBridgeDirectionEnum implements ArrayValuable<Integer> {
-
-    INPUT(1), // 输入
-    OUTPUT(2); // 输出
-
-    private final Integer type;
-
-    public static final Integer[] ARRAYS = Arrays.stream(values()).map(IotDataBridgeDirectionEnum::getType).toArray(Integer[]::new);
-
-    @Override
-    public Integer[] array() {
-        return ARRAYS;
-    }
-
-}

yudao-module-iot/yudao-module-iot-api/src/main/java/cn/iocoder/yudao/module/iot/enums/rule/IotDataSinkTypeEnum.java

@@ -7,32 +7,33 @@ import lombok.RequiredArgsConstructor;
 import java.util.Arrays;
 
 /**
- * IoT 数据流转的数据目的的类型枚举
+ * IoT 数据目的的类型枚举
  *
  * @author 芋道源码
  */
 @RequiredArgsConstructor
 @Getter
-public enum IotDataRuleSinkTypeEnum implements ArrayValuable<Integer> {
+public enum IotDataSinkTypeEnum implements ArrayValuable<Integer> {
 
     HTTP(1, "HTTP"),
     TCP(2, "TCP"),
-    WEBSOCKET(3, "WEBSOCKET"),
+    WEBSOCKET(3, "WebSocket"),
 
     MQTT(10, "MQTT"),
 
-    DATABASE(20, "DATABASE"),
+    DATABASE(20, "Database"),
-    REDIS_STREAM(21, "REDIS_STREAM"),
+    // TODO @芋艿：改成 Redis；通过 execute 通用化；
+    REDIS_STREAM(21, "Redis Stream"),
 
-    ROCKETMQ(30, "ROCKETMQ"),
+    ROCKETMQ(30, "RocketMQ"),
-    RABBITMQ(31, "RABBITMQ"),
+    RABBITMQ(31, "RabbitMQ"),
-    KAFKA(32, "KAFKA");
+    KAFKA(32, "Kafka");
 
     private final Integer type;
 
     private final String name;
 
-    public static final Integer[] ARRAYS = Arrays.stream(values()).map(IotDataRuleSinkTypeEnum::getType).toArray(Integer[]::new);
+    public static final Integer[] ARRAYS = Arrays.stream(values()).map(IotDataSinkTypeEnum::getType).toArray(Integer[]::new);
 
     @Override
     public Integer[] array() {

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/rule/IotDataBridgeController.java

@@ -1,84 +0,0 @@
-package cn.iocoder.yudao.module.iot.controller.admin.rule;
-
-import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum;
-import cn.iocoder.yudao.framework.common.pojo.CommonResult;
-import cn.iocoder.yudao.framework.common.pojo.PageResult;
-import cn.iocoder.yudao.framework.common.util.object.BeanUtils;
-import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.databridge.IotDataBridgePageReqVO;
-import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.databridge.IotDataBridgeRespVO;
-import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.databridge.IotDataBridgeSaveReqVO;
-import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataRuleSinkDO;
-import cn.iocoder.yudao.module.iot.service.rule.IotDataBridgeService;
-import io.swagger.v3.oas.annotations.Operation;
-import io.swagger.v3.oas.annotations.Parameter;
-import io.swagger.v3.oas.annotations.tags.Tag;
-import jakarta.annotation.Resource;
-import jakarta.validation.Valid;
-import org.springframework.security.access.prepost.PreAuthorize;
-import org.springframework.validation.annotation.Validated;
-import org.springframework.web.bind.annotation.*;
-
-import java.util.List;
-
-import static cn.iocoder.yudao.framework.common.pojo.CommonResult.success;
-import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertList;
-
-@Tag(name = "管理后台 - IoT 数据桥梁")
-@RestController
-@RequestMapping("/iot/data-bridge")
-@Validated
-public class IotDataBridgeController {
-
-    @Resource
-    private IotDataBridgeService dataBridgeService;
-
-    @PostMapping("/create")
-    @Operation(summary = "创建数据桥梁")
-    @PreAuthorize("@ss.hasPermission('iot:data-bridge:create')")
-    public CommonResult<Long> createDataBridge(@Valid @RequestBody IotDataBridgeSaveReqVO createReqVO) {
-        return success(dataBridgeService.createDataBridge(createReqVO));
-    }
-
-    @PutMapping("/update")
-    @Operation(summary = "更新数据桥梁")
-    @PreAuthorize("@ss.hasPermission('iot:data-bridge:update')")
-    public CommonResult<Boolean> updateDataBridge(@Valid @RequestBody IotDataBridgeSaveReqVO updateReqVO) {
-        dataBridgeService.updateDataBridge(updateReqVO);
-        return success(true);
-    }
-
-    @DeleteMapping("/delete")
-    @Operation(summary = "删除数据桥梁")
-    @Parameter(name = "id", description = "编号", required = true)
-    @PreAuthorize("@ss.hasPermission('iot:data-bridge:delete')")
-    public CommonResult<Boolean> deleteDataBridge(@RequestParam("id") Long id) {
-        dataBridgeService.deleteDataBridge(id);
-        return success(true);
-    }
-
-    @GetMapping("/get")
-    @Operation(summary = "获得数据桥梁")
-    @Parameter(name = "id", description = "编号", required = true, example = "1024")
-    @PreAuthorize("@ss.hasPermission('iot:data-bridge:query')")
-    public CommonResult<IotDataBridgeRespVO> getDataBridge(@RequestParam("id") Long id) {
-        IotDataRuleSinkDO dataBridge = dataBridgeService.getDataBridge(id);
-        return success(BeanUtils.toBean(dataBridge, IotDataBridgeRespVO.class));
-    }
-
-    @GetMapping("/page")
-    @Operation(summary = "获得数据桥梁分页")
-    @PreAuthorize("@ss.hasPermission('iot:data-bridge:query')")
-    public CommonResult<PageResult<IotDataBridgeRespVO>> getDataBridgePage(@Valid IotDataBridgePageReqVO pageReqVO) {
-        PageResult<IotDataRuleSinkDO> pageResult = dataBridgeService.getDataBridgePage(pageReqVO);
-        return success(BeanUtils.toBean(pageResult, IotDataBridgeRespVO.class));
-    }
-
-    @GetMapping("/simple-list")
-    @Operation(summary = "获取数据桥梁的精简信息列表", description = "主要用于前端的下拉选项")
-    public CommonResult<List<IotDataBridgeRespVO>> getSimpleDataBridgeList() {
-        List<IotDataRuleSinkDO> list = dataBridgeService.getDataBridgeList(CommonStatusEnum.ENABLE.getStatus());
-        return success(convertList(list, dataBridge -> // 只返回 id、name 字段
-                new IotDataBridgeRespVO().setId(dataBridge.getId()).setName(dataBridge.getName())));
-    }
-
-}

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/rule/IotDataSinkController.java

@@ -0,0 +1,84 @@
+package cn.iocoder.yudao.module.iot.controller.admin.rule;
+
+import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum;
+import cn.iocoder.yudao.framework.common.pojo.CommonResult;
+import cn.iocoder.yudao.framework.common.pojo.PageResult;
+import cn.iocoder.yudao.framework.common.util.object.BeanUtils;
+import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.data.sink.IotDataSinkPageReqVO;
+import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.data.sink.IotDataSinkRespVO;
+import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.data.sink.IotDataSinkSaveReqVO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataSinkDO;
+import cn.iocoder.yudao.module.iot.service.rule.data.IotDataSinkService;
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.Parameter;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import jakarta.annotation.Resource;
+import jakarta.validation.Valid;
+import org.springframework.security.access.prepost.PreAuthorize;
+import org.springframework.validation.annotation.Validated;
+import org.springframework.web.bind.annotation.*;
+
+import java.util.List;
+
+import static cn.iocoder.yudao.framework.common.pojo.CommonResult.success;
+import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertList;
+
+@Tag(name = "管理后台 - IoT 数据流转目的")
+@RestController
+@RequestMapping("/iot/data-sink")
+@Validated
+public class IotDataSinkController {
+
+    @Resource
+    private IotDataSinkService dataSinkService;
+
+    @PostMapping("/create")
+    @Operation(summary = "创建数据目的")
+    @PreAuthorize("@ss.hasPermission('iot:data-sink:create')")
+    public CommonResult<Long> createDataSink(@Valid @RequestBody IotDataSinkSaveReqVO createReqVO) {
+        return success(dataSinkService.createDataSink(createReqVO));
+    }
+
+    @PutMapping("/update")
+    @Operation(summary = "更新数据目的")
+    @PreAuthorize("@ss.hasPermission('iot:data-sink:update')")
+    public CommonResult<Boolean> updateDataSink(@Valid @RequestBody IotDataSinkSaveReqVO updateReqVO) {
+        dataSinkService.updateDataSink(updateReqVO);
+        return success(true);
+    }
+
+    @DeleteMapping("/delete")
+    @Operation(summary = "删除数据目的")
+    @Parameter(name = "id", description = "编号", required = true)
+    @PreAuthorize("@ss.hasPermission('iot:data-sink:delete')")
+    public CommonResult<Boolean> deleteDataSink(@RequestParam("id") Long id) {
+        dataSinkService.deleteDataSink(id);
+        return success(true);
+    }
+
+    @GetMapping("/get")
+    @Operation(summary = "获得数据目的")
+    @Parameter(name = "id", description = "编号", required = true, example = "1024")
+    @PreAuthorize("@ss.hasPermission('iot:data-sink:query')")
+    public CommonResult<IotDataSinkRespVO> getDataSink(@RequestParam("id") Long id) {
+        IotDataSinkDO sink = dataSinkService.getDataSink(id);
+        return success(BeanUtils.toBean(sink, IotDataSinkRespVO.class));
+    }
+
+    @GetMapping("/page")
+    @Operation(summary = "获得数据目的分页")
+    @PreAuthorize("@ss.hasPermission('iot:data-sink:query')")
+    public CommonResult<PageResult<IotDataSinkRespVO>> getDataSinkPage(@Valid IotDataSinkPageReqVO pageReqVO) {
+        PageResult<IotDataSinkDO> pageResult = dataSinkService.getDataSinkPage(pageReqVO);
+        return success(BeanUtils.toBean(pageResult, IotDataSinkRespVO.class));
+    }
+
+    @GetMapping("/simple-list")
+    @Operation(summary = "获取数据目的的精简信息列表", description = "主要用于前端的下拉选项")
+    public CommonResult<List<IotDataSinkRespVO>> getSimpleDataSinkList() {
+        List<IotDataSinkDO> list = dataSinkService.getDataSinkListByStatus(CommonStatusEnum.ENABLE.getStatus());
+        return success(convertList(list, sink -> // 只返回 id、name 字段
+                new IotDataSinkRespVO().setId(sink.getId()).setName(sink.getName())));
+    }
+
+}

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/rule/IotRuleSceneController.java

@@ -7,7 +7,7 @@ import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.scene.IotRuleScenePa
 import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.scene.IotRuleSceneRespVO;
 import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.scene.IotRuleSceneSaveReqVO;
 import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotRuleSceneDO;
-import cn.iocoder.yudao.module.iot.service.rule.IotRuleSceneService;
+import cn.iocoder.yudao.module.iot.service.rule.scene.IotRuleSceneService;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.Parameter;
 import io.swagger.v3.oas.annotations.tags.Tag;

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/rule/vo/data/package-info.java

@@ -0,0 +1 @@
+package cn.iocoder.yudao.module.iot.controller.admin.rule.vo.data;

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/rule/vo/data/sink/IotDataSinkPageReqVO.java

@@ -1,4 +1,4 @@
-package cn.iocoder.yudao.module.iot.controller.admin.rule.vo.databridge;
+package cn.iocoder.yudao.module.iot.controller.admin.rule.vo.data.sink;
 
 import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum;
 import cn.iocoder.yudao.framework.common.pojo.PageParam;
@@ -11,14 +11,14 @@ import java.time.LocalDateTime;
 
 import static cn.iocoder.yudao.framework.common.util.date.DateUtils.FORMAT_YEAR_MONTH_DAY_HOUR_MINUTE_SECOND;
 
-@Schema(description = "管理后台 - IoT 数据桥梁分页 Request VO")
+@Schema(description = "管理后台 - IoT 数据流转目的分页 Request VO")
 @Data
-public class IotDataBridgePageReqVO extends PageParam {
+public class IotDataSinkPageReqVO extends PageParam {
 
-    @Schema(description = "桥梁名称", example = "赵六")
+    @Schema(description = "数据目的名称", example = "赵六")
     private String name;
 
-    @Schema(description = "桥梁状态", example = "1")
+    @Schema(description = "数据目的状态", example = "2")
     @InEnum(CommonStatusEnum.class)
     private Integer status;
 

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/rule/vo/data/sink/IotDataSinkRespVO.java

@@ -0,0 +1,34 @@
+package cn.iocoder.yudao.module.iot.controller.admin.rule.vo.data.sink;
+
+import cn.iocoder.yudao.module.iot.dal.dataobject.rule.config.IotDataBridgeAbstractConfig;
+import io.swagger.v3.oas.annotations.media.Schema;
+import lombok.Data;
+
+import java.time.LocalDateTime;
+
+@Schema(description = "管理后台 - IoT 数据流转目的 Response VO")
+@Data
+public class IotDataSinkRespVO {
+
+    @Schema(description = "数据目的编号", requiredMode = Schema.RequiredMode.REQUIRED, example = "18564")
+    private Long id;
+
+    @Schema(description = "数据目的名称", requiredMode = Schema.RequiredMode.REQUIRED, example = "赵六")
+    private String name;
+
+    @Schema(description = "数据目的描述", example = "随便")
+    private String description;
+
+    @Schema(description = "数据目的状态", requiredMode = Schema.RequiredMode.REQUIRED, example = "1")
+    private Integer status;
+
+    @Schema(description = "数据目的类型", requiredMode = Schema.RequiredMode.REQUIRED, example = "1")
+    private Integer type;
+
+    @Schema(description = "数据目的配置")
+    private IotDataBridgeAbstractConfig config;
+
+    @Schema(description = "创建时间", requiredMode = Schema.RequiredMode.REQUIRED)
+    private LocalDateTime createTime;
+
+}

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/rule/vo/data/sink/IotDataSinkSaveReqVO.java

@@ -0,0 +1,40 @@
+package cn.iocoder.yudao.module.iot.controller.admin.rule.vo.data.sink;
+
+import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum;
+import cn.iocoder.yudao.framework.common.validation.InEnum;
+import cn.iocoder.yudao.module.iot.dal.dataobject.rule.config.IotDataBridgeAbstractConfig;
+import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum;
+import io.swagger.v3.oas.annotations.media.Schema;
+import jakarta.validation.constraints.NotEmpty;
+import jakarta.validation.constraints.NotNull;
+import lombok.Data;
+
+@Schema(description = "管理后台 - IoT 数据流转目的新增/修改 Request VO")
+@Data
+public class IotDataSinkSaveReqVO {
+
+    @Schema(description = "数据目的编号", example = "18564")
+    private Long id;
+
+    @Schema(description = "数据目的名称", requiredMode = Schema.RequiredMode.REQUIRED, example = "赵六")
+    @NotEmpty(message = "数据目的名称不能为空")
+    private String name;
+
+    @Schema(description = "数据目的描述", example = "随便")
+    private String description;
+
+    @Schema(description = "数据目的状态", requiredMode = Schema.RequiredMode.REQUIRED, example = "1")
+    @NotNull(message = "数据目的状态不能为空")
+    @InEnum(CommonStatusEnum.class)
+    private Integer status;
+
+    @Schema(description = "数据目的类型", requiredMode = Schema.RequiredMode.REQUIRED, example = "1")
+    @NotNull(message = "数据目的类型不能为空")
+    @InEnum(IotDataSinkTypeEnum.class)
+    private Integer type;
+
+    @Schema(description = "数据目的配置")
+    @NotNull(message = "数据目的配置不能为空")
+    private IotDataBridgeAbstractConfig config;
+
+}

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/rule/vo/databridge/IotDataBridgeRespVO.java

@@ -1,37 +0,0 @@
-package cn.iocoder.yudao.module.iot.controller.admin.rule.vo.databridge;
-
-import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.databridge.config.IotDataBridgeAbstractConfig;
-import io.swagger.v3.oas.annotations.media.Schema;
-import lombok.Data;
-
-import java.time.LocalDateTime;
-
-@Schema(description = "管理后台 - IoT 数据桥梁 Response VO")
-@Data
-public class IotDataBridgeRespVO {
-
-    @Schema(description = "桥梁编号", requiredMode = Schema.RequiredMode.REQUIRED, example = "18564")
-    private Long id;
-
-    @Schema(description = "桥梁名称", requiredMode = Schema.RequiredMode.REQUIRED, example = "赵六")
-    private String name;
-
-    @Schema(description = "桥梁描述", example = "随便")
-    private String description;
-
-    @Schema(description = "桥梁状态", requiredMode = Schema.RequiredMode.REQUIRED, example = "1")
-    private Integer status;
-
-    @Schema(description = "桥梁方向", requiredMode = Schema.RequiredMode.REQUIRED)
-    private Integer direction;
-
-    @Schema(description = "桥梁类型", requiredMode = Schema.RequiredMode.REQUIRED, example = "1")
-    private Integer type;
-
-    @Schema(description = "桥梁配置")
-    private IotDataBridgeAbstractConfig config;
-
-    @Schema(description = "创建时间", requiredMode = Schema.RequiredMode.REQUIRED)
-    private LocalDateTime createTime;
-
-}

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/rule/vo/databridge/IotDataBridgeSaveReqVO.java

@@ -1,46 +0,0 @@
-package cn.iocoder.yudao.module.iot.controller.admin.rule.vo.databridge;
-
-import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum;
-import cn.iocoder.yudao.framework.common.validation.InEnum;
-import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.databridge.config.IotDataBridgeAbstractConfig;
-import cn.iocoder.yudao.module.iot.enums.rule.IotDataBridgeDirectionEnum;
-import cn.iocoder.yudao.module.iot.enums.rule.IotDataRuleSinkTypeEnum;
-import io.swagger.v3.oas.annotations.media.Schema;
-import jakarta.validation.constraints.NotEmpty;
-import jakarta.validation.constraints.NotNull;
-import lombok.Data;
-
-@Schema(description = "管理后台 - IoT 数据桥梁新增/修改 Request VO")
-@Data
-public class IotDataBridgeSaveReqVO {
-
-    @Schema(description = "桥梁编号", requiredMode = Schema.RequiredMode.REQUIRED, example = "18564")
-    private Long id;
-
-    @Schema(description = "桥梁名称", requiredMode = Schema.RequiredMode.REQUIRED, example = "赵六")
-    @NotEmpty(message = "桥梁名称不能为空")
-    private String name;
-
-    @Schema(description = "桥梁描述", example = "随便")
-    private String description;
-
-    @Schema(description = "桥梁状态", requiredMode = Schema.RequiredMode.REQUIRED, example = "2")
-    @NotNull(message = "桥梁状态不能为空")
-    @InEnum(CommonStatusEnum.class)
-    private Integer status;
-
-    @Schema(description = "桥梁方向", requiredMode = Schema.RequiredMode.REQUIRED)
-    @NotNull(message = "桥梁方向不能为空")
-    @InEnum(IotDataBridgeDirectionEnum.class)
-    private Integer direction;
-
-    @Schema(description = "桥梁类型", requiredMode = Schema.RequiredMode.REQUIRED, example = "1")
-    @NotNull(message = "桥梁类型不能为空")
-    @InEnum(IotDataRuleSinkTypeEnum.class)
-    private Integer type;
-
-    @Schema(description = "桥梁配置")
-    @NotNull(message = "桥梁配置不能为空")
-    private IotDataBridgeAbstractConfig config;
-
-}

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/thingmodel/model/ThingModelEvent.java

@@ -9,6 +9,7 @@ import lombok.Data;
 
 import java.util.List;
 
+// TODO @puhui999：感觉这个，是不是放到 dal 里会好点？（讨论下，先不改哈）
 /**
  * IoT 物模型中的事件
  *

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/IotDataRuleDO.java

@@ -1,8 +1,15 @@
 package cn.iocoder.yudao.module.iot.dal.dataobject.rule;
 
 import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum;
+import cn.iocoder.yudao.framework.mybatis.core.type.LongListTypeHandler;
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.product.IotProductDO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.thingmodel.IotThingModelDO;
 import com.baomidou.mybatisplus.annotation.KeySequence;
+import com.baomidou.mybatisplus.annotation.TableField;
 import com.baomidou.mybatisplus.annotation.TableName;
+import com.baomidou.mybatisplus.extension.handlers.JacksonTypeHandler;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
@@ -11,9 +18,9 @@ import lombok.NoArgsConstructor;
 import java.util.List;
 
 /**
- * IoT 数据流转 DO
+ * IoT 数据流转规则 DO
  *
- * 组合 {@link IotDataRuleSourceDO} => {@link IotDataRuleSinkDO}
+ * 监听 {@link SourceConfig} 数据源，转发到 {@link IotDataSinkDO} 数据目的
  *
  * @author 芋道源码
  */
@@ -45,18 +52,54 @@ public class IotDataRuleDO {
     private Integer status;
 
     /**
-     * 数据源编号
+     * 数据源配置数组
-     *
-     * 关联 {@link IotDataRuleSourceDO#getId()}
      */
-    private List<Long> sourceIds;
+    @TableField(typeHandler = JacksonTypeHandler.class)
+    private List<SourceConfig> sourceConfigs;
     /**
      * 数据目的编号
      *
-     * 关联 {@link IotDataRuleSinkDO#getId()}
+     * 关联 {@link IotDataSinkDO#getId()}
      */
+    @TableField(typeHandler = LongListTypeHandler.class)
     private List<Long> sinkIds;
 
     // TODO @芋艿：未来考虑使用 groovy；支持数据处理；
 
+    /**
+     * 数据源配置
+     */
+    @Data
+    public static class SourceConfig {
+
+        /**
+         * 消息方法
+         *
+         * 枚举 {@link IotDeviceMessageMethodEnum} 中的 upstream 上行部分
+         */
+        private String method;
+
+        /**
+         * 产品编号
+         *
+         * 关联 {@link IotProductDO#getId()}
+         */
+        private Long productId;
+        /**
+         * 设备编号
+         *
+         * 关联 {@link IotDeviceDO#getId()}
+         * 特殊：如果为 {@link IotDeviceDO#DEVICE_ID_ALL} 时，则是全部设备
+         */
+        private Long deviceId;
+
+        /**
+         * 标识符
+         *
+         * 1. 物模型时，对应：{@link IotThingModelDO#getIdentifier()}
+         */
+        private String identifier;
+
+    }
+
 }

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/IotDataRuleSourceDO.java

@@ -1,79 +0,0 @@
-package cn.iocoder.yudao.module.iot.dal.dataobject.rule;
-
-import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
-import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
-import cn.iocoder.yudao.module.iot.dal.dataobject.product.IotProductDO;
-import cn.iocoder.yudao.module.iot.dal.dataobject.thingmodel.IotThingModelDO;
-import com.baomidou.mybatisplus.annotation.KeySequence;
-import com.baomidou.mybatisplus.annotation.TableName;
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.Data;
-import lombok.NoArgsConstructor;
-
-import java.util.List;
-
-/**
- * IoT 数据流转的数据源 DO
- *
- * @author 芋道源码
- */
-@TableName(value = "iot_data_flow_source", autoResultMap = true)
-@KeySequence("iot_data_flow_source_seq") // 用于 Oracle、PostgreSQL、Kingbase、DB2、H2 数据库的主键自增。如果是 MySQL 等数据库，可不写。
-@Data
-@Builder
-@NoArgsConstructor
-@AllArgsConstructor
-public class IotDataRuleSourceDO {
-
-    /**
-     * 数据源编号
-     */
-    private Long id;
-    /**
-     * 数据源名称
-     */
-    private String name;
-
-    /**
-     * 配置数组
-     */
-    private List<Config> configs;
-
-    /**
-     * 配置
-     */
-    @Data
-    public static class Config {
-
-        /**
-         * 消息方法
-         *
-         * 枚举 {@link IotDeviceMessageMethodEnum} 中的 upstream 上行部分
-         */
-        private String method;
-
-        /**
-         * 产品编号
-         *
-         * 关联 {@link IotProductDO#getId()}
-         */
-        private Long productId;
-        /**
-         * 设备编号
-         *
-         * 关联 {@link IotDeviceDO#getId()}
-         * 特殊：如果为 {@link IotDeviceDO#DEVICE_ID_ALL} 时，则是全部设备
-         */
-        private Long deviceId;
-
-        /**
-         * 标识符
-         *
-         * 1. 物模型时，对应：{@link IotThingModelDO#getIdentifier()}
-         */
-        private String identifier;
-
-    }
-
-}

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/IotDataSinkDO.java

@@ -2,28 +2,30 @@ package cn.iocoder.yudao.module.iot.dal.dataobject.rule;
 
 import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum;
 import cn.iocoder.yudao.framework.mybatis.core.dataobject.BaseDO;
-import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.databridge.config.IotDataBridgeAbstractConfig;
+import cn.iocoder.yudao.module.iot.dal.dataobject.rule.config.IotDataBridgeAbstractConfig;
-import cn.iocoder.yudao.module.iot.enums.rule.IotDataBridgeDirectionEnum;
+import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum;
-import cn.iocoder.yudao.module.iot.enums.rule.IotDataRuleSinkTypeEnum;
 import com.baomidou.mybatisplus.annotation.KeySequence;
 import com.baomidou.mybatisplus.annotation.TableField;
 import com.baomidou.mybatisplus.annotation.TableId;
 import com.baomidou.mybatisplus.annotation.TableName;
 import com.baomidou.mybatisplus.extension.handlers.JacksonTypeHandler;
-import lombok.*;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
 
 /**
- * IoT 数据流转的数据目的 DO
+ * IoT 数据流转目的 DO
  *
  * @author 芋道源码
  */
-@TableName(value = "iot_data_bridge", autoResultMap = true)
+@TableName(value = "iot_data_sink", autoResultMap = true)
 @KeySequence("iot_data_bridge_seq") // 用于 Oracle、PostgreSQL、Kingbase、DB2、H2 数据库的主键自增。如果是 MySQL 等数据库，可不写。
 @Data
 @Builder
 @NoArgsConstructor
 @AllArgsConstructor
-public class IotDataRuleSinkDO extends BaseDO {
+public class IotDataSinkDO extends BaseDO {
 
     /**
      * 数据目的编号
@@ -39,27 +41,18 @@ public class IotDataRuleSinkDO extends BaseDO {
      */
     private String description;
     /**
-     * 桥梁状态
+     * 数据目的状态
      *
-     * 枚举 {@link CommonStatusEnum}
+     *  枚举 {@link CommonStatusEnum}
      */
-    @Deprecated // TODO @puhui999：这个删除
     private Integer status;
-    /**
-     * 桥梁方向
-     *
-     * 枚举 {@link IotDataBridgeDirectionEnum}
-     */
-    @Deprecated // TODO @puhui999：这个删除
-    private Integer direction;
 
     /**
      * 数据目的类型
      *
-     * 枚举 {@link IotDataRuleSinkTypeEnum}
+     * 枚举 {@link IotDataSinkTypeEnum}
      */
     private Integer type;
-
     /**
      * 数据目的配置
      */

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/IotRuleSceneDO.java

@@ -201,7 +201,7 @@ public class IotRuleSceneDO extends TenantBaseDO {
          * 数据桥接编号
          *
          * 必填：当 {@link #type} 为 {@link IotRuleSceneActionTypeEnum#DATA_BRIDGE} 时
-         * 关联：{@link IotDataRuleSinkDO#getId()}
+         * 关联：{@link IotDataSinkDO#getId()}
          */
         private Long dataBridgeId;
 

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/config/IotDataBridgeAbstractConfig.java

@@ -1,6 +1,6 @@
-package cn.iocoder.yudao.module.iot.controller.admin.rule.vo.databridge.config;
+package cn.iocoder.yudao.module.iot.dal.dataobject.rule.config;
 
-import cn.iocoder.yudao.module.iot.enums.rule.IotDataRuleSinkTypeEnum;
+import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum;
 import com.fasterxml.jackson.annotation.JsonSubTypes;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import lombok.Data;
@@ -8,7 +8,7 @@ import lombok.Data;
 /**
  * IoT IotDataBridgeConfig 抽象类
  *
- * 用于表示数据桥梁配置数据的通用类型，根据具体的 "type" 字段动态映射到对应的子类
+ * 用于表示数据目的配置数据的通用类型，根据具体的 "type" 字段动态映射到对应的子类
  * 提供多态支持，适用于不同类型的数据结构序列化和反序列化场景。
  *
  * @author HUIHUI
@@ -28,7 +28,7 @@ public abstract class IotDataBridgeAbstractConfig {
     /**
      * 配置类型
      *
-     * 枚举 {@link IotDataRuleSinkTypeEnum#getType()}
+     * 枚举 {@link IotDataSinkTypeEnum#getType()}
      */
     private String type;
 

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/config/IotDataBridgeHttpConfig.java

@@ -1,4 +1,4 @@
-package cn.iocoder.yudao.module.iot.controller.admin.rule.vo.databridge.config;
+package cn.iocoder.yudao.module.iot.dal.dataobject.rule.config;
 
 import lombok.Data;
 

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/config/IotDataBridgeKafkaMQConfig.java

@@ -1,4 +1,4 @@
-package cn.iocoder.yudao.module.iot.controller.admin.rule.vo.databridge.config;
+package cn.iocoder.yudao.module.iot.dal.dataobject.rule.config;
 
 import lombok.Data;
 

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/config/IotDataBridgeMqttConfig.java

@@ -1,4 +1,4 @@
-package cn.iocoder.yudao.module.iot.controller.admin.rule.vo.databridge.config;
+package cn.iocoder.yudao.module.iot.dal.dataobject.rule.config;
 
 import lombok.Data;
 

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/config/IotDataBridgeRabbitMQConfig.java

@@ -1,4 +1,4 @@
-package cn.iocoder.yudao.module.iot.controller.admin.rule.vo.databridge.config;
+package cn.iocoder.yudao.module.iot.dal.dataobject.rule.config;
 
 import lombok.Data;
 

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/config/IotDataBridgeRedisStreamConfig.java

@@ -1,4 +1,4 @@
-package cn.iocoder.yudao.module.iot.controller.admin.rule.vo.databridge.config;
+package cn.iocoder.yudao.module.iot.dal.dataobject.rule.config;
 
 import lombok.Data;
 

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/config/IotDataBridgeRocketMQConfig.java

@@ -1,4 +1,4 @@
-package cn.iocoder.yudao.module.iot.controller.admin.rule.vo.databridge.config;
+package cn.iocoder.yudao.module.iot.dal.dataobject.rule.config;
 
 import lombok.Data;
 

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/rule/IotDataBridgeMapper.java

@@ -1,34 +0,0 @@
-package cn.iocoder.yudao.module.iot.dal.mysql.rule;
-
-import cn.iocoder.yudao.framework.common.pojo.PageResult;
-import cn.iocoder.yudao.framework.mybatis.core.mapper.BaseMapperX;
-import cn.iocoder.yudao.framework.mybatis.core.query.LambdaQueryWrapperX;
-import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.databridge.IotDataBridgePageReqVO;
-import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataRuleSinkDO;
-import org.apache.ibatis.annotations.Mapper;
-
-import java.util.List;
-
-/**
- * IoT 数据桥梁 Mapper
- *
- * @author HUIHUI
- */
-@Mapper
-public interface IotDataBridgeMapper extends BaseMapperX<IotDataRuleSinkDO> {
-
-    default PageResult<IotDataRuleSinkDO> selectPage(IotDataBridgePageReqVO reqVO) {
-        return selectPage(reqVO, new LambdaQueryWrapperX<IotDataRuleSinkDO>()
-                .likeIfPresent(IotDataRuleSinkDO::getName, reqVO.getName())
-                .eqIfPresent(IotDataRuleSinkDO::getStatus, reqVO.getStatus())
-                .betweenIfPresent(IotDataRuleSinkDO::getCreateTime, reqVO.getCreateTime())
-                .orderByDesc(IotDataRuleSinkDO::getId));
-    }
-
-    default List<IotDataRuleSinkDO> selectList(Integer status) {
-        return selectList(new LambdaQueryWrapperX<IotDataRuleSinkDO>()
-                .eqIfPresent(IotDataRuleSinkDO::getStatus, status)
-                .orderByDesc(IotDataRuleSinkDO::getId));
-    }
-
-}

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/rule/IotDataSinkMapper.java

@@ -0,0 +1,32 @@
+package cn.iocoder.yudao.module.iot.dal.mysql.rule;
+
+import cn.iocoder.yudao.framework.common.pojo.PageResult;
+import cn.iocoder.yudao.framework.mybatis.core.mapper.BaseMapperX;
+import cn.iocoder.yudao.framework.mybatis.core.query.LambdaQueryWrapperX;
+import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.data.sink.IotDataSinkPageReqVO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataSinkDO;
+import org.apache.ibatis.annotations.Mapper;
+
+import java.util.List;
+
+/**
+ * IoT 数据流转目的 Mapper
+ *
+ * @author HUIHUI
+ */
+@Mapper
+public interface IotDataSinkMapper extends BaseMapperX<IotDataSinkDO> {
+
+    default PageResult<IotDataSinkDO> selectPage(IotDataSinkPageReqVO reqVO) {
+        return selectPage(reqVO, new LambdaQueryWrapperX<IotDataSinkDO>()
+                .likeIfPresent(IotDataSinkDO::getName, reqVO.getName())
+                .eqIfPresent(IotDataSinkDO::getStatus, reqVO.getStatus())
+                .betweenIfPresent(IotDataSinkDO::getCreateTime, reqVO.getCreateTime())
+                .orderByDesc(IotDataSinkDO::getId));
+    }
+
+    default List<IotDataSinkDO> selectListByStatus(Integer status) {
+        return selectList(IotDataSinkDO::getStatus, status);
+    }
+
+}

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/job/rule/IotRuleSceneJob.java

@@ -2,7 +2,7 @@ package cn.iocoder.yudao.module.iot.job.rule;
 
 import cn.hutool.core.map.MapUtil;
 import cn.iocoder.yudao.module.iot.enums.rule.IotRuleSceneTriggerTypeEnum;
-import cn.iocoder.yudao.module.iot.service.rule.IotRuleSceneService;
+import cn.iocoder.yudao.module.iot.service.rule.scene.IotRuleSceneService;
 import jakarta.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
 import org.quartz.JobExecutionContext;

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/mq/consumer/rule/IotRuleSceneMessageHandler.java

@@ -3,7 +3,7 @@ package cn.iocoder.yudao.module.iot.mq.consumer.rule;
 import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
 import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber;
 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
-import cn.iocoder.yudao.module.iot.service.rule.IotRuleSceneService;
+import cn.iocoder.yudao.module.iot.service.rule.scene.IotRuleSceneService;
 import jakarta.annotation.PostConstruct;
 import jakarta.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/IotDataBridgeService.java

@@ -1,64 +0,0 @@
-package cn.iocoder.yudao.module.iot.service.rule;
-
-import cn.iocoder.yudao.framework.common.pojo.PageResult;
-import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.databridge.IotDataBridgePageReqVO;
-import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.databridge.IotDataBridgeSaveReqVO;
-import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataRuleSinkDO;
-import jakarta.validation.Valid;
-
-import java.util.List;
-
-/**
- * IoT 数据桥梁 Service 接口
- *
- * @author HUIHUI
- */
-public interface IotDataBridgeService {
-
-    /**
-     * 创建数据桥梁
-     *
-     * @param createReqVO 创建信息
-     * @return 编号
-     */
-    Long createDataBridge(@Valid IotDataBridgeSaveReqVO createReqVO);
-
-    /**
-     * 更新数据桥梁
-     *
-     * @param updateReqVO 更新信息
-     */
-    void updateDataBridge(@Valid IotDataBridgeSaveReqVO updateReqVO);
-
-    /**
-     * 删除数据桥梁
-     *
-     * @param id 编号
-     */
-    void deleteDataBridge(Long id);
-
-    /**
-     * 获得数据桥梁
-     *
-     * @param id 编号
-     * @return 数据桥梁
-     */
-    IotDataRuleSinkDO getDataBridge(Long id);
-
-    /**
-     * 获得数据桥梁分页
-     *
-     * @param pageReqVO 分页查询
-     * @return 数据桥梁分页
-     */
-    PageResult<IotDataRuleSinkDO> getDataBridgePage(IotDataBridgePageReqVO pageReqVO);
-
-    /**
-     * 获取数据桥梁列表
-     *
-     * @param status 状态，如果为空，则不进行筛选
-     * @return 数据桥梁列表
-     */
-    List<IotDataRuleSinkDO> getDataBridgeList(Integer status);
-
-}

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/IotDataBridgeServiceImpl.java

@@ -1,77 +0,0 @@
-package cn.iocoder.yudao.module.iot.service.rule;
-
-import cn.iocoder.yudao.framework.common.pojo.PageResult;
-import cn.iocoder.yudao.framework.common.util.object.BeanUtils;
-import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.databridge.IotDataBridgePageReqVO;
-import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.databridge.IotDataBridgeSaveReqVO;
-import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataRuleSinkDO;
-import cn.iocoder.yudao.module.iot.dal.mysql.rule.IotDataBridgeMapper;
-import jakarta.annotation.Resource;
-import org.springframework.stereotype.Service;
-import org.springframework.validation.annotation.Validated;
-
-import java.util.List;
-
-import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
-import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.DATA_BRIDGE_NOT_EXISTS;
-
-/**
- * IoT 数据桥梁 Service 实现类
- *
- * @author HUIHUI
- */
-@Service
-@Validated
-public class IotDataBridgeServiceImpl implements IotDataBridgeService {
-
-    @Resource
-    private IotDataBridgeMapper dataBridgeMapper;
-
-    @Override
-    public Long createDataBridge(IotDataBridgeSaveReqVO createReqVO) {
-        // 插入
-        IotDataRuleSinkDO dataBridge = BeanUtils.toBean(createReqVO, IotDataRuleSinkDO.class);
-        dataBridgeMapper.insert(dataBridge);
-        // 返回
-        return dataBridge.getId();
-    }
-
-    @Override
-    public void updateDataBridge(IotDataBridgeSaveReqVO updateReqVO) {
-        // 校验存在
-        validateDataBridgeExists(updateReqVO.getId());
-        // 更新
-        IotDataRuleSinkDO updateObj = BeanUtils.toBean(updateReqVO, IotDataRuleSinkDO.class);
-        dataBridgeMapper.updateById(updateObj);
-    }
-
-    @Override
-    public void deleteDataBridge(Long id) {
-        // 校验存在
-        validateDataBridgeExists(id);
-        // 删除
-        dataBridgeMapper.deleteById(id);
-    }
-
-    private void validateDataBridgeExists(Long id) {
-        if (dataBridgeMapper.selectById(id) == null) {
-            throw exception(DATA_BRIDGE_NOT_EXISTS);
-        }
-    }
-
-    @Override
-    public IotDataRuleSinkDO getDataBridge(Long id) {
-        return dataBridgeMapper.selectById(id);
-    }
-
-    @Override
-    public PageResult<IotDataRuleSinkDO> getDataBridgePage(IotDataBridgePageReqVO pageReqVO) {
-        return dataBridgeMapper.selectPage(pageReqVO);
-    }
-
-    @Override
-    public List<IotDataRuleSinkDO> getDataBridgeList(Integer status) {
-        return dataBridgeMapper.selectList(status);
-    }
-
-}

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataSinkService.java

@@ -0,0 +1,64 @@
+package cn.iocoder.yudao.module.iot.service.rule.data;
+
+import cn.iocoder.yudao.framework.common.pojo.PageResult;
+import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.data.sink.IotDataSinkPageReqVO;
+import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.data.sink.IotDataSinkSaveReqVO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataSinkDO;
+import jakarta.validation.Valid;
+
+import java.util.List;
+
+/**
+ * IoT 数据流转目的 Service 接口
+ *
+ * @author HUIHUI
+ */
+public interface IotDataSinkService {
+
+    /**
+     * 创建数据目的
+     *
+     * @param createReqVO 创建信息
+     * @return 编号
+     */
+    Long createDataSink(@Valid IotDataSinkSaveReqVO createReqVO);
+
+    /**
+     * 更新数据目的
+     *
+     * @param updateReqVO 更新信息
+     */
+    void updateDataSink(@Valid IotDataSinkSaveReqVO updateReqVO);
+
+    /**
+     * 删除数据目的
+     *
+     * @param id 编号
+     */
+    void deleteDataSink(Long id);
+
+    /**
+     * 获得数据目的
+     *
+     * @param id 编号
+     * @return 数据目的
+     */
+    IotDataSinkDO getDataSink(Long id);
+
+    /**
+     * 获得数据目的分页
+     *
+     * @param pageReqVO 分页查询
+     * @return 数据目的分页
+     */
+    PageResult<IotDataSinkDO> getDataSinkPage(IotDataSinkPageReqVO pageReqVO);
+
+    /**
+     * 获取数据目的列表
+     *
+     * @param status 状态，如果为空，则不进行筛选
+     * @return 数据目的列表
+     */
+    List<IotDataSinkDO> getDataSinkListByStatus(Integer status);
+
+}

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotDataSinkServiceImpl.java

@@ -0,0 +1,75 @@
+package cn.iocoder.yudao.module.iot.service.rule.data;
+
+import cn.iocoder.yudao.framework.common.pojo.PageResult;
+import cn.iocoder.yudao.framework.common.util.object.BeanUtils;
+import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.data.sink.IotDataSinkPageReqVO;
+import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.data.sink.IotDataSinkSaveReqVO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataSinkDO;
+import cn.iocoder.yudao.module.iot.dal.mysql.rule.IotDataSinkMapper;
+import jakarta.annotation.Resource;
+import org.springframework.stereotype.Service;
+import org.springframework.validation.annotation.Validated;
+
+import java.util.List;
+
+import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
+import static cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants.DATA_BRIDGE_NOT_EXISTS;
+
+/**
+ * IoT 数据流转目的 Service 实现类
+ *
+ * @author HUIHUI
+ */
+@Service
+@Validated
+public class IotDataSinkServiceImpl implements IotDataSinkService {
+
+    @Resource
+    private IotDataSinkMapper dataSinkMapper;
+
+    @Override
+    public Long createDataSink(IotDataSinkSaveReqVO createReqVO) {
+        IotDataSinkDO dataBridge = BeanUtils.toBean(createReqVO, IotDataSinkDO.class);
+        dataSinkMapper.insert(dataBridge);
+        return dataBridge.getId();
+    }
+
+    @Override
+    public void updateDataSink(IotDataSinkSaveReqVO updateReqVO) {
+        // 校验存在
+        validateDataBridgeExists(updateReqVO.getId());
+        // 更新
+        IotDataSinkDO updateObj = BeanUtils.toBean(updateReqVO, IotDataSinkDO.class);
+        dataSinkMapper.updateById(updateObj);
+    }
+
+    @Override
+    public void deleteDataSink(Long id) {
+        // 校验存在
+        validateDataBridgeExists(id);
+        // 删除
+        dataSinkMapper.deleteById(id);
+    }
+
+    private void validateDataBridgeExists(Long id) {
+        if (dataSinkMapper.selectById(id) == null) {
+            throw exception(DATA_BRIDGE_NOT_EXISTS);
+        }
+    }
+
+    @Override
+    public IotDataSinkDO getDataSink(Long id) {
+        return dataSinkMapper.selectById(id);
+    }
+
+    @Override
+    public PageResult<IotDataSinkDO> getDataSinkPage(IotDataSinkPageReqVO pageReqVO) {
+        return dataSinkMapper.selectPage(pageReqVO);
+    }
+
+    @Override
+    public List<IotDataSinkDO> getDataSinkListByStatus(Integer status) {
+        return dataSinkMapper.selectListByStatus(status);
+    }
+
+}

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/IotRuleSceneDataBridgeAction.java

@@ -1,13 +1,13 @@
-package cn.iocoder.yudao.module.iot.service.rule.action;
+package cn.iocoder.yudao.module.iot.service.rule.data;
 
 import cn.hutool.core.lang.Assert;
 import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum;
 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
-import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataRuleSinkDO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataSinkDO;
 import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotRuleSceneDO;
 import cn.iocoder.yudao.module.iot.enums.rule.IotRuleSceneActionTypeEnum;
-import cn.iocoder.yudao.module.iot.service.rule.IotDataBridgeService;
+import cn.iocoder.yudao.module.iot.service.rule.data.action.IotDataBridgeExecute;
-import cn.iocoder.yudao.module.iot.service.rule.action.databridge.IotDataBridgeExecute;
+import cn.iocoder.yudao.module.iot.service.rule.scene.action.IotRuleSceneAction;
 import jakarta.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
@@ -19,12 +19,13 @@ import java.util.List;
  *
  * @author 芋道源码
  */
+@Deprecated
 @Component
 @Slf4j
 public class IotRuleSceneDataBridgeAction implements IotRuleSceneAction {
 
     @Resource
-    private IotDataBridgeService dataBridgeService;
+    private IotDataSinkService dataBridgeService;
     @Resource
     private List<IotDataBridgeExecute<?>> dataBridgeExecutes;
 
@@ -36,7 +37,7 @@ public class IotRuleSceneDataBridgeAction implements IotRuleSceneAction {
         }
         // 1.2 获得数据桥梁
         Assert.notNull(config.getDataBridgeId(), "数据桥梁编号不能为空");
-        IotDataRuleSinkDO dataBridge = dataBridgeService.getDataBridge(config.getDataBridgeId());
+        IotDataSinkDO dataBridge = dataBridgeService.getDataSink(config.getDataBridgeId());
         if (dataBridge == null || dataBridge.getConfig() == null) {
             log.error("[execute][message({}) config({}) 对应的数据桥梁不存在]", message, config);
             return;

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/AbstractCacheableDataBridgeExecute.java

@@ -1,8 +1,8 @@
-package cn.iocoder.yudao.module.iot.service.rule.action.databridge;
+package cn.iocoder.yudao.module.iot.service.rule.data.action;
 
 import cn.hutool.core.util.ObjUtil;
 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
-import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataRuleSinkDO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataSinkDO;
 import com.google.common.cache.CacheBuilder;
 import com.google.common.cache.CacheLoader;
 import com.google.common.cache.LoadingCache;
@@ -100,7 +100,7 @@ public abstract class AbstractCacheableDataBridgeExecute<Config, Producer> imple
 
     @Override
     @SuppressWarnings({"unchecked"})
-    public void execute(IotDeviceMessage message, IotDataRuleSinkDO dataBridge) {
+    public void execute(IotDeviceMessage message, IotDataSinkDO dataBridge) {
         if (ObjUtil.notEqual(dataBridge.getType(), getType())) {
             return;
         }

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotDataBridgeExecute.java

@@ -1,7 +1,7 @@
-package cn.iocoder.yudao.module.iot.service.rule.action.databridge;
+package cn.iocoder.yudao.module.iot.service.rule.data.action;
 
 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
-import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataRuleSinkDO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataSinkDO;
 
 /**
  * IoT 数据桥梁的执行器 execute 接口
@@ -24,7 +24,7 @@ public interface IotDataBridgeExecute<Config> {
      * @param dataBridge 数据桥梁
      */
     @SuppressWarnings({"unchecked"})
-    default void execute(IotDeviceMessage message, IotDataRuleSinkDO dataBridge) throws Exception {
+    default void execute(IotDeviceMessage message, IotDataSinkDO dataBridge) throws Exception {
         // 1.1 校验数据桥梁类型
         if (!getType().equals(dataBridge.getType())) {
             return;

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotHttpDataBridgeExecute.java

@@ -1,11 +1,11 @@
-package cn.iocoder.yudao.module.iot.service.rule.action.databridge;
+package cn.iocoder.yudao.module.iot.service.rule.data.action;
 
 import cn.hutool.core.collection.CollUtil;
 import cn.iocoder.yudao.framework.common.util.http.HttpUtils;
 import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
-import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.databridge.config.IotDataBridgeHttpConfig;
+import cn.iocoder.yudao.module.iot.dal.dataobject.rule.config.IotDataBridgeHttpConfig;
 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
-import cn.iocoder.yudao.module.iot.enums.rule.IotDataRuleSinkTypeEnum;
+import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum;
 import jakarta.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.http.*;
@@ -30,7 +30,7 @@ public class IotHttpDataBridgeExecute implements IotDataBridgeExecute<IotDataBri
 
     @Override
     public Integer getType() {
-        return IotDataRuleSinkTypeEnum.HTTP.getType();
+        return IotDataSinkTypeEnum.HTTP.getType();
     }
 
     @Override

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotKafkaMQDataBridgeExecute.java

@@ -1,8 +1,8 @@
-package cn.iocoder.yudao.module.iot.service.rule.action.databridge;
+package cn.iocoder.yudao.module.iot.service.rule.data.action;
 
-import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.databridge.config.IotDataBridgeKafkaMQConfig;
+import cn.iocoder.yudao.module.iot.dal.dataobject.rule.config.IotDataBridgeKafkaMQConfig;
 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
-import cn.iocoder.yudao.module.iot.enums.rule.IotDataRuleSinkTypeEnum;
+import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.serialization.StringSerializer;
@@ -31,7 +31,7 @@ public class IotKafkaMQDataBridgeExecute extends
 
     @Override
     public Integer getType() {
-        return IotDataRuleSinkTypeEnum.KAFKA.getType();
+        return IotDataSinkTypeEnum.KAFKA.getType();
     }
 
     @Override

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRabbitMQDataBridgeExecute.java

@@ -1,8 +1,8 @@
-package cn.iocoder.yudao.module.iot.service.rule.action.databridge;
+package cn.iocoder.yudao.module.iot.service.rule.data.action;
 
-import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.databridge.config.IotDataBridgeRabbitMQConfig;
+import cn.iocoder.yudao.module.iot.dal.dataobject.rule.config.IotDataBridgeRabbitMQConfig;
 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
-import cn.iocoder.yudao.module.iot.enums.rule.IotDataRuleSinkTypeEnum;
+import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
@@ -26,7 +26,7 @@ public class IotRabbitMQDataBridgeExecute extends
 
     @Override
     public Integer getType() {
-        return IotDataRuleSinkTypeEnum.RABBITMQ.getType();
+        return IotDataSinkTypeEnum.RABBITMQ.getType();
     }
 
     @Override

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRedisStreamDataBridgeExecute.java

@@ -1,9 +1,9 @@
-package cn.iocoder.yudao.module.iot.service.rule.action.databridge;
+package cn.iocoder.yudao.module.iot.service.rule.data.action;
 
 import cn.hutool.core.util.StrUtil;
-import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.databridge.config.IotDataBridgeRedisStreamConfig;
+import cn.iocoder.yudao.module.iot.dal.dataobject.rule.config.IotDataBridgeRedisStreamConfig;
 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
-import cn.iocoder.yudao.module.iot.enums.rule.IotDataRuleSinkTypeEnum;
+import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum;
 import lombok.extern.slf4j.Slf4j;
 import org.redisson.Redisson;
 import org.redisson.api.RedissonClient;
@@ -29,7 +29,7 @@ public class IotRedisStreamDataBridgeExecute extends
 
     @Override
     public Integer getType() {
-        return IotDataRuleSinkTypeEnum.REDIS_STREAM.getType();
+        return IotDataSinkTypeEnum.REDIS_STREAM.getType();
     }
 
     @Override

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotRocketMQDataBridgeExecute.java

@@ -1,8 +1,8 @@
-package cn.iocoder.yudao.module.iot.service.rule.action.databridge;
+package cn.iocoder.yudao.module.iot.service.rule.data.action;
 
-import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.databridge.config.IotDataBridgeRocketMQConfig;
+import cn.iocoder.yudao.module.iot.dal.dataobject.rule.config.IotDataBridgeRocketMQConfig;
 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
-import cn.iocoder.yudao.module.iot.enums.rule.IotDataRuleSinkTypeEnum;
+import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.rocketmq.client.producer.DefaultMQProducer;
 import org.apache.rocketmq.client.producer.SendResult;
@@ -25,7 +25,7 @@ public class IotRocketMQDataBridgeExecute extends
 
     @Override
     public Integer getType() {
-        return IotDataRuleSinkTypeEnum.ROCKETMQ.getType();
+        return IotDataSinkTypeEnum.ROCKETMQ.getType();
     }
 
     @Override

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/IotRuleSceneService.java

@@ -1,4 +1,4 @@
-package cn.iocoder.yudao.module.iot.service.rule;
+package cn.iocoder.yudao.module.iot.service.rule.scene;
 
 import cn.iocoder.yudao.framework.common.pojo.PageResult;
 import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.scene.IotRuleScenePageReqVO;

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/IotRuleSceneServiceImpl.java

@@ -1,4 +1,4 @@
-package cn.iocoder.yudao.module.iot.service.rule;
+package cn.iocoder.yudao.module.iot.service.rule.scene;
 
 import cn.hutool.core.collection.CollUtil;
 import cn.hutool.core.collection.ListUtil;
@@ -27,7 +27,7 @@ import cn.iocoder.yudao.module.iot.enums.rule.IotRuleSceneConditionOperatorEnum;
 import cn.iocoder.yudao.module.iot.enums.rule.IotRuleSceneTriggerTypeEnum;
 import cn.iocoder.yudao.module.iot.framework.job.core.IotSchedulerManager;
 import cn.iocoder.yudao.module.iot.job.rule.IotRuleSceneJob;
-import cn.iocoder.yudao.module.iot.service.rule.action.IotRuleSceneAction;
+import cn.iocoder.yudao.module.iot.service.rule.scene.action.IotRuleSceneAction;
 import jakarta.annotation.Resource;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/action/IotRuleSceneAction.java

@@ -1,4 +1,4 @@
-package cn.iocoder.yudao.module.iot.service.rule.action;
+package cn.iocoder.yudao.module.iot.service.rule.scene.action;
 
 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
 import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotRuleSceneDO;

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/action/IotRuleSceneAlertAction.java

@@ -1,4 +1,4 @@
-package cn.iocoder.yudao.module.iot.service.rule.action;
+package cn.iocoder.yudao.module.iot.service.rule.scene.action;
 
 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
 import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotRuleSceneDO;

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/action/IotRuleSceneDeviceControlAction.java

@@ -1,4 +1,4 @@
-package cn.iocoder.yudao.module.iot.service.rule.action;
+package cn.iocoder.yudao.module.iot.service.rule.scene.action;
 
 import cn.hutool.core.lang.Assert;
 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;

yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/action/databridge/IotDataBridgeExecuteTest.java

@@ -1,9 +1,10 @@
 package cn.iocoder.yudao.module.iot.service.rule.action.databridge;
 
 import cn.iocoder.yudao.framework.test.core.ut.BaseMockitoUnitTest;
-import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.databridge.config.*;
 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
-import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataRuleSinkDO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotDataSinkDO;
+import cn.iocoder.yudao.module.iot.dal.dataobject.rule.config.*;
+import cn.iocoder.yudao.module.iot.service.rule.data.action.*;
 import lombok.extern.slf4j.Slf4j;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Disabled;
@@ -129,7 +130,7 @@ public class IotDataBridgeExecuteTest extends BaseMockitoUnitTest {
 
         // 3. 执行测试
         log.info("[testHttpDataBridge][执行HTTP数据桥接测试]");
-        httpDataBridgeExecute.execute(message, new IotDataRuleSinkDO()
+        httpDataBridgeExecute.execute(message, new IotDataSinkDO()
                 .setType(httpDataBridgeExecute.getType()).setConfig(config));
     }
 
@@ -144,10 +145,10 @@ public class IotDataBridgeExecuteTest extends BaseMockitoUnitTest {
     private void executeAndVerifyCache(IotDataBridgeExecute<?> action, IotDataBridgeAbstractConfig config, String type)
             throws Exception {
         log.info("[test{}DataBridge][第一次执行，应该会创建新的 producer]", type);
-        action.execute(message, new IotDataRuleSinkDO().setType(action.getType()).setConfig(config));
+        action.execute(message, new IotDataSinkDO().setType(action.getType()).setConfig(config));
 
         log.info("[test{}DataBridge][第二次执行，应该会复用缓存的 producer]", type);
-        action.execute(message, new IotDataRuleSinkDO().setType(action.getType()).setConfig(config));
+        action.execute(message, new IotDataSinkDO().setType(action.getType()).setConfig(config));
     }
 
 }