@ -1,410 +1,584 @@
# 平台对接设计文档
本文档按当前项目代码反补,描述已落地的平台对接模块。相关实现主要位于:
- `fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform`
- `fastbee-open-api/src/main/java/com/fastbee/controller/platform`
- `yack-web/src/views/platform`
- `sql/platform-integration.sql`
- `sql/platform-dcp-init-data.sql`
## 1. 背景与目标
FastBee 需要将设备数据推送至第三方监测平台(当前为 DCP 结构监测 SaaS) , 并支持未来对接多个平台。不同平台的认证方式、接口路径、字段格式各不相同, 需要一套可配置、可扩展的对接框架。
FastBee 需要将设备上报数据推送到第三方监测平台,当前已落地 DCP 结构监测 SaaS。模块目标是把“平台、接口、流程、设备绑定、字段映射、推送日志”配置化, 降低后续接入新平台或调整 DCP 接口的代码改动 。
**设计原则:**
- 平台、接口、流程、设备绑定全部可视化配置,无需改代码即可对接新平台
- 严格区分三类执行模式,避免过度设计
- 字段映射通过 UI 关联选择,不手填字符串
- 推送全程有日志,可追溯
当前实现采用两条主链路:
1. 流程执行链路: 用于平台初始化、Token 刷新、设备入网等多步骤调用。
2. 实时推送链路:设备上报后异步命中平台设备绑定,由 DCP 适配器组装请求体并推送。
设计边界:
- 流程编排只负责顺序 HTTP 调用、上下文传递、结果持久化。
- 实时数据推送不走流程引擎,按设备绑定上的 `push_interface_id` 、`field_mapping`、`extra_mapping` 直接组装请求。
- 当前只有 DCP 适配器实现,平台路由按 `platform_instance.code` 分发,`code=dcp` 时进入 `DcpPlatformAdapter` 。
- 网关多子设备场景通过独立子表 `platform_device_sub_mapping` 管理,不再只依赖 `extra_mapping` 。
---
## 2. 三种执行模式
### 模式一:流程编排(多步串联)
### 2.1 流程编排
**适用场景:** 多个接口必须按顺序调用,且上一步的返回值是下一步的入参 。
适用于多个接口必须顺序调用,且后续步骤依赖前序步骤返回值的场景 。
| 流程 | 触发方式 | 步骤 | 频率 |
| 流程 | 触发方式 | 典型步骤 | 结果落库 |
|---|---|---|---|
| 平台初始化 | 手动点击一次 | Token → userId → projectId | 一次性 |
| 设备入网 | 手动点击(每设备) | 创建测点 → 绑定传感器 → 获得 DCP deviceId | 每设备一次 |
| 平台初始化 | 平台管理页手动初始化,或手动执行流程 | `dcp_auth` -> `dcp_get_user` -> `dcp_get_projects` | `access_token` 、`token_expires_at`、`extra_config.userId`、`extra_config.projectId`、`extra_config.projectName` |
| Token 刷新 | HTTP 调用前自动触发,也可手动执行 | `dcp_auth` | `access_token` 、`token_expires_at` |
| 直连设备入网 | 设备绑定页点击入网 | `dcp_create_point` -> 对应传感器绑定接口 | 主表 `platform_point_id` 、`platform_device_id`、`net_status` |
| 网关设备入网 | 设备绑定页点击入网 | 对每个 monitor 子设备逐个执行入网流程 | 子表 `platform_point_id` 、`platform_device_id`、`net_status`;全部 monitor 成功后主表 `net_status=1` |
执行结果自动持久化: userId/projectId 写入 `platform_instance.extra_config` , DCP deviceId 回填到 `platform_device_mapping` 。
入口代码:
### 模式二:直接调用(实时推送)
- `PlatformInstanceServiceImpl.initialize(Long id)`
- `PlatformFlowServiceImpl.executeManually(Long id, Long mappingId)`
- `PlatformDeviceMappingServiceImpl.onboard(Long mappingId)`
- `PlatformFlowExecutor.execute(PlatformFlow flow, FlowContext context)`
**适用场景:** 设备上报数据时,按字段映射转换后调用单个推送接口。
### 2.2 直接调用实时推送
流程:设备上报 → 字段映射转换 → 直接 HTTP 调用推送接口 → 写推送日志
设备上报后由 `DeviceReportTrigger.onDeviceReport(serialNumber, payload)` 异步触发:
**不走流程引擎**,最短路径,不引入额外开销。
1. `PlatformDeviceMappingMapper.selectBySerialNumber` 同时匹配直连主设备和网关子设备。
2. 过滤未启用的平台和绑定关系。
3. 按平台 `code` 分发,目前 `dcp` 进入 `DcpPlatformAdapter` 。
4. 适配器按推送接口的 `request_schema` 与设备绑定字段映射组装请求体。
5. 检查并刷新 Token。
6. 发送 HTTP POST, 写入 `platform_push_log` 。
### 模式三: Token 自动刷新(横切关注点)
实时推送不经过 `PlatformFlowExecutor` ,因此不会使用流程步骤的 `param_mapping` 、`result_key`、`result_save`。
每次 HTTP 调用前自动检查 `token_expires_at` ,提前 5 分钟视为过期,过期则先调用认证接口刷新 Token, 对业务逻辑完全透明。
### 2.3 Token 自动刷新
流程执行器和 DCP 实时推送适配器各自调用 `ensureToken` :
- `access_token` 为空,或 `token_expires_at` 距当前时间不足 5 分钟时视为过期。
- 自动查找同平台 `flow_type=3` 且 `enable='1'` 的 Token 刷新流程。
- 刷新流程本身不注入 Authorization, 避免递归。
- 刷新成功后重新读取或更新 `platform_instance.access_token` 和 `token_expires_at` 。
---
## 3. 数据库设计( 6 张表)
## 3. 数据库设计(7 张表)
### 3.1 `platform_instance` — 平台实体
DDL 以 `sql/platform-integration.sql` 为准。
```sql
CREATE TABLE `platform_instance` (
`id` bigint NOT NULL AUTO_INCREMENT,
`name` varchar(100) NOT NULL COMMENT '平台名称,如 DCP结构监测',
`code` varchar(50) NOT NULL UNIQUE COMMENT '平台唯一标识,如 dcp',
`base_url` varchar(255) NOT NULL COMMENT '基础URL',
`auth_type` tinyint NOT NULL DEFAULT 1 COMMENT '认证类型(1=密码Token,2=ApiKey,3=OAuth2)',
`auth_config` json DEFAULT NULL COMMENT '认证凭证JSON {userName,password,clientId,scope}',
`extra_config` json DEFAULT NULL COMMENT '初始化后保存的平台数据 {userId,projectId,...}',
`access_token` varchar(2000) DEFAULT NULL COMMENT '当前有效Token',
`token_expires_at` datetime DEFAULT NULL COMMENT 'Token过期时间',
`init_status` tinyint NOT NULL DEFAULT 0 COMMENT '初始化状态(0=未初始化,1=已初始化)',
`enable` char(1) NOT NULL DEFAULT '1',
`tenant_id` bigint DEFAULT NULL,
`remark` varchar(500) DEFAULT NULL,
`create_by` varchar(64) DEFAULT '',
`create_time` datetime DEFAULT CURRENT_TIMESTAMP,
`update_by` varchar(64) DEFAULT '',
`update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`)
) COMMENT='第三方平台实体';
```
### 3.1 `platform_instance` 平台实体
### 3.2 `platform_interface` — 接口定义
保存第三方平台实例、认证配置和初始化结果。
```sql
CREATE TABLE `platform_interface` (
`id` bigint NOT NULL AUTO_INCREMENT,
`platform_id` bigint NOT NULL COMMENT '所属平台',
`name` varchar(100) NOT NULL COMMENT '接口名称',
`code` varchar(100) NOT NULL COMMENT '接口标识',
`method` varchar(10) NOT NULL COMMENT 'HTTP方法',
`path` varchar(255) NOT NULL COMMENT '接口路径,支持路径变量 {userId}',
`content_type` varchar(100) DEFAULT 'application/json',
`request_schema` json DEFAULT NULL COMMENT '入参字段定义',
`response_schema` json DEFAULT NULL COMMENT '出参字段定义',
`sort_order` int DEFAULT 0,
`remark` varchar(500) DEFAULT NULL,
`create_by` varchar(64) DEFAULT '',
`create_time` datetime DEFAULT CURRENT_TIMESTAMP,
`update_by` varchar(64) DEFAULT '',
`update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
KEY `idx_platform_id` (`platform_id`)
) COMMENT='平台接口定义';
```
关键字段:
**`request_schema` 字段 source 枚举:**
- `fixed` — 固定值( value 字段)
- `auth` — 从 `auth_config` 取( field 字段指定 key)
- `extra` — 从 `extra_config` 取(初始化后保存的 userId/projectId 等)
- `context` — 从流程上下文或运行时上下文取( contextKey 指定,如 `step1.access_token` )
- `device` — 从设备绑定关系取(如 `platform_device_id` )
- `thingsModel` — 从设备物模型实时值取(设备绑定页选择 identifier)
- `ts` — 自动取设备上报时间
| 字段 | 说明 |
|---|---|
| `name` | 平台名称 |
| `code` | 平台唯一标识,当前 DCP 为 `dcp` ,有唯一索引 |
| `base_url` | 平台基础 URL |
| `auth_type` | 认证类型:`1=密码Token`、`2=ApiKey`、`3=OAuth2` |
| `auth_config` | 认证配置 JSON, 例如 `userName` 、`password`、`clientId`、`scope` |
| `extra_config` | 初始化后保存的业务上下文,例如 `userId` 、`projectId`、`projectName` |
| `access_token` | 当前 Token |
| `token_expires_at` | Token 过期时间 |
| `init_status` | 初始化状态 |
| `enable` | 是否启用 |
| `tenant_id` | 租户 ID |
**数组字段补充:**
- `type=array` 的字段通过 `itemSchema` 描述数组元素结构
- `itemSchema.source` 额外支持 `subDeviceId` ,用于数组元素内绑定子设备平台 ID( 如静力水准 `sensorId` )
### 3.2 `platform_interface` 接口定义
### 3.3 `platform_flow` — 流程定义
保存第三方 HTTP 接口定义。
```sql
CREATE TABLE `platform_flow` (
`id` bigint NOT NULL AUTO_INCREMENT,
`platform_id` bigint NOT NULL,
`name` varchar(100) NOT NULL,
`code` varchar(100) NOT NULL,
`flow_type` tinyint NOT NULL COMMENT '流程类型(1=平台初始化,2=设备入网,3=Token刷新)',
`sensor_type` varchar(50) DEFAULT NULL COMMENT '传感器类型, flow_type=2时有效',
`enable` char(1) NOT NULL DEFAULT '1',
`remark` varchar(500) DEFAULT NULL,
`create_by` varchar(64) DEFAULT '',
`create_time` datetime DEFAULT CURRENT_TIMESTAMP,
`update_by` varchar(64) DEFAULT '',
`update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
KEY `idx_platform_type` (`platform_id`, `flow_type` )
) COMMENT='平台对接流程';
```
关键字段:
**flow_type 说明:**
- `1` 平台初始化:手动触发,获取 userId/projectId 等基础数据
- `2` 设备入网:手动触发,创建测点+绑定传感器,获得 DCP deviceId
- `3` Token 刷新:系统自动触发,不在 UI 中手动执行
| 字段 | 说明 |
|---|---|
| `platform_id` | 所属平台 |
| `name` / `code` | 接口名称与接口标识 |
| `method` | `GET` 、`POST`、`PUT`、`DELETE` |
| `path` | 接口路径,支持 `{userId}` 形式路径变量 |
| `content_type` | 默认 `application/json` ,认证接口可用 `application/x-www-form-urlencoded` |
| `request_schema` | 入参字段定义 |
| `response_schema` | 出参字段定义,支持按路径抽取 |
| `sort_order` | 排序 |
### 3.4 `platform_flow_step` — 流程步骤
`request_schema` 在流程执行器中支持的 `source` :
```sql
CREATE TABLE `platform_flow_step` (
`id` bigint NOT NULL AUTO_INCREMENT,
`flow_id` bigint NOT NULL,
`step_order` int NOT NULL DEFAULT 1 COMMENT '执行顺序, 从1开始',
`interface_id` bigint NOT NULL COMMENT '关联接口',
`param_mapping` json DEFAULT NULL COMMENT '入参来源覆盖配置',
`result_key` varchar(100) DEFAULT NULL COMMENT '出参写入上下文的key, 如 step1',
`result_save` json DEFAULT NULL COMMENT '需持久化的出参字段 [{field,saveTo}]',
`remark` varchar(255) DEFAULT NULL,
PRIMARY KEY (`id`),
KEY `idx_flow_id` (`flow_id`)
) COMMENT='流程步骤';
```
| source | 含义 |
|---|---|
| `fixed` | 固定值,取 `value` |
| `uuid` | 每次请求生成新的 UUID |
| `auth` | 从 `platform_instance.auth_config` 按 `field` 取值 |
| `extra` | 从 `platform_instance.extra_config` 按 `field` 取值 |
| `context` | 从 `FlowContext` 按 `contextKey` 取值 |
| `device` | 从 `PlatformDeviceMapping` 取 `platform_device_id` 、`platform_point_id`、`serial_number`、`gateway_serial_number`、`monitor_serial_number` |
| `onboard_config` | 从当前子映射 `onboard_config` 取值 |
| `subDeviceSensorIds` | 收集子设备 sensorId 列表 |
| `referenceSubDeviceSensorIdPairs` | 收集 `[基准点sensorId, 监测点sensorId]` 配对列表,可能展开为多次请求 |
**`result_save` 示例:** 初始化流程 Step2 获取 userId 后自动保存
```json
[
{"field": "id", "saveTo": "extra_config.userId"}
]
```
实时推送适配器中支持的 `source` :
### 3.5 `platform_device_mapping` — 设备绑定
| source | 含义 |
|---|---|
| `fixed` | 固定值 |
| `device` | 从绑定关系取平台设备/测点/序列号 |
| `auth` | 从认证 JSON 取值 |
| `extra` | 从扩展 JSON 取值 |
| `thingsModel` / `identifier` | 按 `field_mapping` 或 `itemFieldMappings` 从上报 payload 取物模型值 |
| `ts` | 从物模型值的 `__ts` 派生时间,缺省为当前时间 |
| `context` | 优先从 payload 的 `contextKey` 取值;`reportTime`/`epoch` 会按时间字段处理 |
| `subDeviceId` | 数组元素内的子设备平台 ID, 如 DCP `sensorId` |
```sql
CREATE TABLE `platform_device_mapping` (
`id` bigint NOT NULL AUTO_INCREMENT,
`platform_id` bigint NOT NULL,
`device_id` bigint NOT NULL COMMENT 'FastBee设备ID',
`serial_number` varchar(64) NOT NULL COMMENT 'FastBee设备序列号',
`platform_device_id` varchar(100) DEFAULT NULL COMMENT 'DCP侧deviceId, 入网后自动回填',
`platform_point_id` varchar(100) DEFAULT NULL COMMENT 'DCP侧pointId',
`sensor_type` varchar(50) NOT NULL COMMENT '传感器类型: inclino/stress/hyeto/static_level',
`push_interface_id` bigint DEFAULT NULL COMMENT '推送接口ID, 决定设备绑定页动态字段',
`field_mapping` json DEFAULT NULL COMMENT '字段映射 [{identifier,platformField,transform}]',
`extra_mapping` json DEFAULT NULL COMMENT '扩展映射,如数组字段子设备映射/基准点配置',
`net_status` tinyint NOT NULL DEFAULT 0 COMMENT '入网状态(0=未入网,1=已入网)',
`enable` char(1) NOT NULL DEFAULT '1',
`remark` varchar(255) DEFAULT NULL,
`create_by` varchar(64) DEFAULT '',
`create_time` datetime DEFAULT CURRENT_TIMESTAMP,
`update_by` varchar(64) DEFAULT '',
`update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
KEY `idx_serial` (`serial_number`),
KEY `idx_platform_device` (`platform_id`, `device_id` )
) COMMENT='平台设备绑定关系';
```
`response_schema.path` 支持对象路径、数组下标路径,例如 `id` 、`[0].id`、`$.data.id`。
### 3.6 `platform_push_log` — 推送日志
### 3.3 `platform_flow` 流程定义
```sql
CREATE TABLE `platform_push_log` (
`id` bigint NOT NULL AUTO_INCREMENT,
`platform_id` bigint NOT NULL,
`flow_id` bigint DEFAULT NULL,
`step_order` int DEFAULT NULL,
`device_id` bigint DEFAULT NULL,
`serial_number` varchar(64) DEFAULT NULL,
`interface_code` varchar(100) DEFAULT NULL,
`request_url` varchar(500) DEFAULT NULL,
`request_body` mediumtext DEFAULT NULL,
`response_code` int DEFAULT NULL,
`response_body` mediumtext DEFAULT NULL,
`status` tinyint NOT NULL DEFAULT 0 COMMENT '0=进行中,1=成功,2=失败',
`error_msg` varchar(1000) DEFAULT NULL,
`idempotency_key` varchar(100) DEFAULT NULL COMMENT '幂等键',
`retry_count` int NOT NULL DEFAULT 0,
`cost_ms` int DEFAULT NULL,
`create_time` datetime DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
KEY `idx_platform_status` (`platform_id`, `status` ),
KEY `idx_device_time` (`device_id`, `create_time` )
) COMMENT='平台推送日志';
```
关键字段:
| 字段 | 说明 |
|---|---|
| `platform_id` | 所属平台 |
| `name` / `code` | 流程名称与标识 |
| `flow_type` | `1=平台初始化` 、`2=设备入网`、`3=Token刷新` |
| `sensor_type` | 入网流程适用传感器类型,如 `inclino` 、`stress`、`hyeto`、`static_level` |
| `enable` | 是否启用 |
### 3.4 `platform_flow_step` 流程步骤
关键字段:
| 字段 | 说明 |
|---|---|
| `flow_id` | 所属流程 |
| `step_order` | 执行顺序 |
| `interface_id` | 关联接口 |
| `param_mapping` | 覆盖接口默认入参来源 |
| `result_key` | 步骤结果写入上下文的 key |
| `result_save` | 将响应字段持久化到平台或设备绑定 |
`result_save` 支持:
- `extra_config.xxx`
- `platform_instance.access_token`
- `platform_instance.token_expires_at` ,值按 `expires_in` 秒数转换为 Date
- `device_mapping.platform_device_id`
- `device_mapping.platform_point_id`
### 3.5 `platform_device_mapping` 设备绑定主表
保存 FastBee 设备与第三方平台的绑定关系。
关键字段:
| 字段 | 说明 |
|---|---|
| `platform_id` | 所属平台 |
| `device_id` / `serial_number` | FastBee 设备 ID 与序列号 |
| `platform_device_id` | 平台侧 deviceId, 直连设备入网后回填 |
| `platform_point_id` | 平台侧 pointId, 直连设备入网后回填 |
| `sensor_type` | `inclino` 、`stress`、`hyeto`、`static_level` |
| `device_type` | `0=直连设备` 、`1=网关设备` |
| `push_interface_id` | 推送接口 ID, 用于动态字段映射和推送路径 |
| `field_mapping` | 标量字段映射,格式如 `[{identifier, platformField, transform}]` |
| `extra_mapping` | 数组字段、静力水准、上报间隔等扩展配置 |
| `net_status` | `0=未入网` 、`1=已入网` |
| `enable` | 是否启用 |
### 3.6 `platform_device_sub_mapping` 子设备映射表
用于网关多子设备场景,尤其是静力水准。
关键字段:
| 字段 | 说明 |
|---|---|
| `parent_mapping_id` | 主绑定 ID |
| `sub_serial_number` | 子设备序列号 |
| `device_role` | `reference=基准点` 、`monitor=监测点` |
| `platform_device_id` | 平台分配给该监测点的 deviceId |
| `platform_point_id` | 平台分配给该监测点的 pointId |
| `onboard_config` | 入网参数 JSON, 例如 `sensorId` 、`refSensorId`、`refSerialNumber` |
| `net_status` | 子设备入网状态 |
唯一约束:`parent_mapping_id + sub_serial_number`。
### 3.7 `platform_push_log` 推送日志
记录流程步骤调用、实时数据推送和失败信息。
关键字段:
| 字段 | 说明 |
|---|---|
| `platform_id` | 平台 ID |
| `flow_id` / `step_order` | 流程调用时记录流程与步骤 |
| `device_id` / `serial_number` | 设备维度 |
| `interface_code` | 接口标识 |
| `request_url` / `request_body` | 请求信息 |
| `response_code` / `response_body` | 响应信息 |
| `status` | `0=进行中` 、`1=成功`、`2=失败` |
| `error_msg` | 错误信息 |
| `idempotency_key` | 幂等键,当前表结构保留 |
| `log_type` | `1=数据推送` 、`2=操作日志` |
| `retry_count` | 重试次数 |
| `cost_ms` | 耗时 |
---
## 4. 后端代码结构
```text
fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/
├── domain/
│ ├── PlatformInstance
│ ├── PlatformInterface
│ ├── PlatformFlow
│ ├── PlatformFlowStep
│ ├── PlatformDeviceMapping
│ ├── PlatformDeviceSubMapping
│ └── PlatformPushLog
├── mapper/
│ ├── PlatformInstanceMapper
│ ├── PlatformInterfaceMapper
│ ├── PlatformFlowMapper
│ ├── PlatformFlowStepMapper
│ ├── PlatformDeviceMappingMapper
│ ├── PlatformDeviceSubMappingMapper
│ └── PlatformPushLogMapper
├── service/
│ ├── IPlatformInstanceService
│ ├── IPlatformFlowService
│ ├── IPlatformDeviceMappingService
│ ├── IPlatformDeviceSubMappingService
│ ├── IPlatformPushLogService
│ └── impl/
├── flow/
│ ├── PlatformFlowExecutor
│ └── FlowContext
├── adapter/
│ └── dcp/
│ ├── DcpPlatformAdapter
│ └── StaticLevelBufferService
└── trigger/
└── DeviceReportTrigger
```
fastbee-service/fastbee-iot-service/
└── com.fastbee.iot.platform/
├── domain/ # 6个实体类
├── mapper/ # MyBatis-Plus Mapper
├── service/
│ ├── IPlatformInstanceService
│ ├── IPlatformFlowService # 含手动触发、流程执行
│ ├── IPlatformDeviceMappingService
│ ├── IPlatformPushLogService
│ └── impl/
├── flow/
│ ├── PlatformFlowExecutor # 核心流程执行引擎
│ └── FlowContext # 步骤间传递的上下文
├── adapter/
│ ├── AbstractPlatformAdapter # Token管理、HTTP调用、日志
│ └── dcp/
│ ├── DcpPlatformAdapter # DCP实现
│ └── strategy/
│ ├── DcpSensorStrategy # 接口
│ ├── InclinoStrategy # 倾角
│ ├── StressStrategy # 轴力计(递增约束)
│ ├── HyetoStrategy # 雨量计(递增约束)
│ └── StaticLevelStrategy # 静力水准
└── trigger/
├── DeviceReportTrigger # 挂接 RuleProcess event=1
└── ScheduledPushJob # Quartz 定时批量推送
fastbee-open-api/
└── com.fastbee.controller.platform/
├── PlatformInstanceController # /platform/instance
├── PlatformInterfaceController # /platform/interface
├── PlatformFlowController # /platform/flow( 含手动触发)
├── PlatformDeviceMappingController # /platform/device-mapping( 含入网)
└── PlatformPushLogController # /platform/push-log
```text
fastbee-open-api/src/main/java/com/fastbee/controller/platform/
├── PlatformInstanceController # /platform/instance
├── PlatformInterfaceController # /platform/interface
├── PlatformFlowController # /platform/flow
├── PlatformDeviceMappingController # /platform/device-mapping
└── PlatformPushLogController # /platform/push-log
```
---
## 5. 核心逻辑
### 5.1 流程执行引擎( PlatformFlowExecutor)
### 5.1 流程执行器
```
FlowContext ctx = new FlowContext(platformInstance, deviceMapping?)
`PlatformFlowExecutor` 的执行顺序:
for each step (ordered by step_order):
1. 按 request_schema + param_mapping 解析入参
- fixed: 直接取 value
- auth: 从 auth_config JSON 取
- extra: 从 extra_config JSON 取( userId/projectId)
- context: 从 ctx.get(contextKey) 取(上一步出参)
- device: 从 deviceMapping 取( platform_device_id 等)
2. 替换 path 变量,如 /project/by-user-id/{userId}
3. 检查 Token, 过期则先刷新
4. HttpClientFactory.instance(config).executeAsString()
5. 将出参写入 ctx( key = step.resultKey)
6. 按 result_save 将指定字段持久化到 platform_instance 或 platform_device_mapping
7. 异步写 platform_push_log
```text
execute(flow, context)
查询 flow 下所有 step, 按 step_order 顺序执行
对每个 step:
1. 读取关联 platform_interface
2. 合并 request_schema 与 step.param_mapping
3. 按 source 解析参数
4. 若参数中存在 list-of-lists, 展开为多批请求
5. 每批请求刷新 uuid 参数
6. 替换 path 中的 {变量}
7. GET 参数放 query; form 接口编码为 x-www-form-urlencoded; 其他接口发 JSON body
8. 非认证步骤自动注入 Authorization: Bearer {token}
9. 执行 HTTP
10. 按 response_schema 抽取响应,写入 FlowContext
11. 按 result_save 持久化
12. 异步写 platform_push_log, log_type=2
```
### 5.2 Token 自动刷新( AbstractPlatformAdapter)
批量展开用于 `referenceSubDeviceSensorIdPairs` 这类参数。例如静力水准入网时,一个网关下多个 monitor 子设备会拆成多次绑定请求。
```
before any HTTP call:
if token is null OR now > token_expires_at - 5min:
find flow where flow_type=3 for this platform
execute flow → get new access_token + expires_in
update platform_instance.access_token + token_expires_at
inject "Authorization: Bearer {token}" into request headers
### 5.2 设备入网
入口:`PlatformDeviceMappingServiceImpl.onboard`。
直连设备:
1. 查找同平台、`flow_type=2`、`sensor_type` 匹配且启用的流程。
2. 创建 `FlowContext(platform, mapping, null)` 。
3. 执行流程。
4. 若主表已有 `platform_device_id` ,将主绑定 `net_status` 改为 `1` 。
网关设备:
1. 读取 `platform_device_sub_mapping` 。
2. 筛选 `device_role=monitor` 且未入网的子设备。
3. 对每个 monitor 创建独立 `FlowContext` :
- `currentMonitorSerialNumber = monitor.sub_serial_number`
- `subMappingOnboardConfig = monitor.onboard_config`
4. 执行同一个入网流程。
5. 从上下文 `step1.id` 、`step2.id` 取 pointId/deviceId, 写回当前 monitor 子表。
6. 所有 monitor 都入网后,主绑定 `net_status=1` 。
7. 若部分成功部分失败,已成功子设备会保留结果,接口返回明确失败信息。
### 5.3 设备上报触发
`DeviceReportTrigger.onDeviceReport` 使用异步线程池 `FastBeeConstant.TASK.PUSH_TO_THIRD_PLATFORM_ASYNC` 。
匹配逻辑在 `PlatformDeviceMappingMapper.xml` :
```sql
SELECT m.* FROM platform_device_mapping m
WHERE m.serial_number = #{serialNumber} AND m.enable = '1'
UNION
SELECT m.* FROM platform_device_mapping m
INNER JOIN platform_device_sub_mapping s ON s.parent_mapping_id = m.id
WHERE s.sub_serial_number = #{serialNumber} AND m.enable = '1'
```
### 5.3 实时推送( DeviceReportTrigger)
分发规则:
```
// 挂接在 RuleProcess.processRuleScript() 调用方之后,异步执行
List< PlatformDeviceMapping > mappings = findBySerialNumber(serialNumber)
// 既匹配主设备 serial_number, 也匹配 extra_mapping 中登记的子设备 serial_number
for each mapping where enable=1 and net_status=1:
adapter = getAdapter(mapping.platformId) // 按 platform.code 路由
adapter.pushData(platform, mapping, reportSerialNumber, payload)
// pushData 内部:按 request_schema / itemSchema 动态组装 → Token检查 → HTTP调用 → 写日志
- `platform.code=dcp` 且 `device_type=0` :调用 `DcpPlatformAdapter.pushData` 。
- `platform.code=dcp` 且 `device_type=1` :先在子表中按上报序列号找 `PlatformDeviceSubMapping` ,找到后调用 `pushGatewayData` 。
- 未知平台 code 只记录 warning, 不推送。
### 5.4 DCP 普通推送
DCP 推送接口优先使用设备绑定的 `push_interface_id` :
- 有推送接口:使用接口 `path` 、`code`、`request_schema`。
- 未配置或解析失败:按 `sensor_type` 回退到内置路径。
内置路径:
| sensor_type | 路径 |
|---|---|
| `inclino` | `/api/app/inclino-data/many` |
| `stress` | `/api/app/stress-data/many` |
| `hyeto` | `/api/app/hyeto-data/many` |
| `static_level` | `/api/app/static-level-data/many` |
请求体构建规则:
- payload 支持 JSON 对象和数组两种格式。
- 数组格式按 `id` 或 `identifier` 转成 `{identifier: value}` ,并把 `ts` 保存为 `{identifier}__ts` 。
- 有 `request_schema` 时按 schema 组装。
- 若 schema 组装结果为空,或没有任何 payload 驱动字段,则回退到 legacy 组装:
- `deviceId = platform_device_id`
- `epoch = 当前时间`
- 其他字段来自 `field_mapping`
- 普通推送请求体最终是数组:`[ {...} ]`。
时间字段会统一转为带时区的 ISO-8601 毫秒格式, 支持输入毫秒时间戳、秒时间戳、ISO 字符串、本地时间字符串。
### 5.5 网关与数组字段推送
网关推送入口:`DcpPlatformAdapter.pushGatewayData`。
当推送接口没有 `type=array` 字段时:
- 按普通子设备推送处理。
- `deviceId` 优先使用子表 `platform_device_id` 。
- 字段映射仍复用主表 `field_mapping` 。
当推送接口包含 `type=array` 字段时:
- 当前实现以静力水准为主要落地场景。
- `extra_mapping` 可配置:
- `fieldName` :数组字段名,默认 `sensorData`
- `itemFieldMappings` :数组元素物模型字段映射
- `reportIntervalSec` :缓存有效时间,默认 300 秒
- 子设备 `onboard_config.sensorId` 作为数组元素的 `sensorId` 。
- monitor 子设备 `platform_device_id` 作为请求体顶层 `deviceId` 。
静力水准缓存逻辑:
1. 基准点 `device_role=reference` 上报时,只缓存当前 item, 不立即推送。
2. 监测点 `device_role=monitor` 上报时,查找其 `onboard_config.refSerialNumber` 对应的基准点缓存。
3. 若基准点缓存新鲜,推送:
```json
[
{
"deviceId": "监测点平台deviceId",
"sensorData": [
{"sensorId": "基准点sensorId", "height": 0.0, "temperature": 0.0, "epoch": "..."},
{"sensorId": "监测点sensorId", "height": 0.0, "temperature": 0.0, "epoch": "..."}
],
"ct": 1
}
]
```
### 5.4 DCP 传感器推送策略
4. 若监测点先到或基准点缓存过期,监测点数据会写入 Redis monitor 缓存。当前代码已有 monitor 缓存读写能力,但推送路径主要依赖“监测点到达时基准点已就绪”的分支。
| sensor_type | 推送接口 | 关键字段 | 特殊约束 |
|---|---|---|---|
| inclino | POST /api/app/inclino-data/many | xDeg, yDeg, temperature, epoch | 无 |
| stress | POST /api/app/stress-data/many | e, epoch | e 必须严格递增 |
| hyeto | POST /api/app/hyeto-data/many | value, epoch | value 必须严格递增 |
| static_level | POST /api/app/static-level-data/many | sensorData[{epoch,height,temperature,sensorId}] | 网关下子设备成对推送 |
缓存键:
递增约束处理:推送前查询 `platform_push_log` 中上次成功推送的值,若当前值 ≤ 上次值则跳过并记录警告。
- `static_level:ref:{platformId}:{fieldName}:{serialNumber}`
- `static_level:monitor:{platformId}:{fieldName}:{serialNumber}`
### 5.5 数组入参通用方案
Redis TTL 为 `reportIntervalSec * 2` ,读取时超过 `reportIntervalSec * 1.5` 视为过期。
当前实现不再把静力水准写成硬编码特例,而是抽象成“数组字段推送”:
### 5.6 日志与重试
1. 推送接口在 `platform_interface.request_schema` 中定义 `type=array` 字段及 `itemSchema`
2. 设备绑定页读取所选推送接口的 `request_schema`
3. 标量字段中 `source=thingsModel` 的字段,渲染为普通字段映射
4. 数组字段中 `itemSchema.source=thingsModel` 的字段,渲染为数组项字段映射
5. 数组字段中 `itemSchema.source=subDeviceId` 的字段,渲染为子设备与平台字段映射
6. 运行时由 `DeviceReportTrigger` 按实际上报的子设备序列号命中绑定关系,再由适配器组装完整数组请求体
静力水准是第一类落地场景:
- 绑定的是网关级设备关系,`platform_device_id` 也是网关级
- `extra_mapping.arrayFields[].subDeviceMappings[]` 保存子设备序列号与 DCP `sensorId` 的对应关系
- 其中一个子设备标记为基准点
- 基准点上报时,推送 `[基准点, 基准点]`
- 监测点上报时,若基准点缓存仍在有效期内,则推送 `[基准点, 当前监测点]`
- 若监测点先到,则先暂存,待下一次基准点上报后补推
- 流程步骤日志:`log_type=2`。
- 实时数据推送日志:`log_type=1`。
- HTTP 成功时记录 URL、请求体、响应体、耗时。
- HTTP 失败时记录 `status=2` 和 `error_msg` 。
- `PlatformPushLogController` 暴露 `POST /platform/push-log/{id}/retry` ,重试逻辑在 `IPlatformPushLogService.retry` 实现中扩展。
---
## 6. 前端页面( yack-web)
菜单:**系统管理 → 平台对接**
菜单初始化见 `sql/platform-menu.sql` ,页面位于 `yack-web/src/views/platform` 。
### 平台管理
- 列表: 名称、标识、baseURL、Token状态( 有效/过期/未认证)、初始化状态、启用
- 操作:新增/编辑、**初始化**(执行 flow_type=1 流程)、启用/禁用
### 6.1 平台管理
### 流程编排
- 顶部选择平台,列表展示该平台所有流程
- 编辑页:步骤列表(可排序),每步选择接口 + 配置字段映射
- 字段来源下拉:固定值 / 认证配置 / 上下文(下拉展示前序步骤出参)/ 平台已保存数据
- 操作:保存、**手动执行**(展示实时执行日志)
路径:`yack-web/src/views/platform/instance/index.vue`
### 设备绑定
- 顶部选择平台,列表展示绑定关系
- 列表关联展示: 设备名称( 非ID) 、入网状态标签
- 新增弹窗:
- 设备搜索选择器(输入名称/序列号模糊搜索)
- 传感器类型下拉
- 推送接口选择器:从页面选择平台下已定义的推送接口
- 字段映射:根据所选推送接口 `request_schema` 动态渲染,不再按传感器类型写死
- 若 `request_schema` 含 `type=array` 字段,则自动出现“数组字段配置”区域
- 数组字段配置包含:
- 子设备序列号 ↔ 平台子设备字段映射(如 `sensorId` )
- 数组元素中 `source=thingsModel` 字段的物模型映射(如 `height` 、`temperature`)
- 基准点选择与缓存有效间隔
- 操作:**入网**(执行 flow_type=2 流程,成功后自动回填 DCP deviceId) 、启用/禁用
后端接口:`/platform/instance`
### 推送日志
- 筛选:平台、设备(搜索选择)、状态、时间范围
- 列表:时间、平台、设备名称、接口、状态标签、耗时
- 详情弹窗: 请求URL/Body、响应码/Body( JSON格式化) 、错误信息
主要能力:
- 查询平台列表。
- 新增、编辑、删除平台。
- 平台详情接口会清空 `access_token` ,避免前端拿到 Token。
- 初始化平台:`POST /platform/instance/{id}/initialize`。
### 6.2 接口定义
路径:`yack-web/src/views/platform/components/InterfaceDrawer.vue`、`HttpRequestDesigner.vue`。
后端接口:`/platform/interface`
主要能力:
- 按平台维护接口。
- 配置 method、path、content-type、请求 schema、响应 schema。
- 请求 schema 用于流程执行和设备绑定页动态字段映射。
### 6.3 流程编排
路径:`yack-web/src/views/platform/flow/index.vue`
后端接口:`/platform/flow`
主要能力:
- 按平台查询流程。
- 新增、编辑流程及步骤。
- 查看流程详情时返回 `{flow, steps}` 。
- 手动执行:`POST /platform/flow/{id}/execute`,可选 `mappingId` 。
### 6.4 设备绑定
路径:`yack-web/src/views/platform/device-mapping/index.vue`
后端接口:`/platform/device-mapping`
主要能力:
- 管理平台、设备、传感器类型、设备类型、推送接口、字段映射。
- 入网:`POST /platform/device-mapping/{id}/onboard`。
- 子设备映射:
- 查询:`GET /platform/device-mapping/{id}/sub-mappings`
- 全量同步:`POST /platform/device-mapping/{id}/sub-mappings`
网关设备必须配置子映射, monitor 子设备才会执行入网并获得平台 deviceId。
### 6.5 推送日志
路径:`yack-web/src/views/platform/push-log/index.vue`
后端接口:`/platform/push-log`
主要能力:
- 查询推送/操作日志。
- 查看日志详情。
- 手动重推入口:`POST /platform/push-log/{id}/retry`。
---
## 7. DCP 内置初始化数据
系统内置 DCP 平台的接口定义和流程配置( SQL 初始化数据),用户只需填入账号密码即可使用。
初始化 SQL: `sql/platform-dcp-init-data.sql` 。
**内置接口( 12个) : **
1. `dcp_auth` — POST /connect/token
2. `dcp_get_user` — GET /api/app/user/user-id
3. `dcp_get_projects` — GET /api/app/project/by-user-id/{userId}
4. `dcp_create_point` — POST /api/app/point
5. `dcp_bind_inclino` — POST /api/app/inclino-sensor
6. `dcp_bind_stress` — POST /api/app/stress-sensor
7. `dcp_bind_hyeto` — POST /api/app/hyeto-sensor
8. `dcp_bind_static_level` — POST /api/app/static-level-device
9. `dcp_push_inclino` — POST /api/app/inclino-data/many
10. `dcp_push_stress` — POST /api/app/stress-data/many
11. `dcp_push_hyeto` — POST /api/app/hyeto-data/many
12. `dcp_push_static_level` — POST /api/app/static-level-data/many
使用前需要先创建 DCP 平台实例:
**内置流程( 6个) : **
1. `dcp_init` ( flow_type=1) : Step1=dcp_auth → Step2=dcp_get_user → Step3=dcp_get_projects
2. `dcp_token_refresh` ( flow_type=3) : Step1=dcp_auth
3. `dcp_onboard_inclino` ( flow_type=2, sensor_type=inclino) : Step1=dcp_create_point → Step2=dcp_bind_inclino
4. `dcp_onboard_stress` ( flow_type=2, sensor_type=stress) : Step1=dcp_create_point → Step2=dcp_bind_stress
5. `dcp_onboard_hyeto` ( flow_type=2, sensor_type=hyeto) : Step1=dcp_create_point → Step2=dcp_bind_hyeto
6. `dcp_onboard_static_level` ( flow_type=2, sensor_type=static_level) : Step1=dcp_create_point → Step2=dcp_bind_static_level
```sql
SET @dcp_platform_id = (SELECT id FROM platform_instance WHERE code = 'dcp' LIMIT 1);
```
补充:
- 上述接口和流程既可通过 SQL 初始化,也可直接在页面维护
- 当前已具备接口管理页、流程编排页、设备绑定页,不需要再靠改 SQL 才能新增运行系统
- 接口入参支持 `uuid` 来源:运行时每次请求生成新的 UUID, 避免把 `auto` 字符串直接发送给目标平台
- 接口入参支持 `referenceSubDeviceSensorIdPairs` 来源:按设备绑定页配置的基准点,将子设备拆成 `[基准点, 监测点]` 多个请求批次,适合静力水准等成组设备绑定场景
### 7.1 内置接口( 12 个)
| code | 方法 | 路径 | 用途 |
|---|---|---|---|
| `dcp_auth` | POST | `/connect/token` | 获取 Token |
| `dcp_get_user` | GET | `/api/app/user/user-id` | 获取当前用户 |
| `dcp_get_projects` | GET | `/api/app/project/by-user-id/{userId}` | 获取项目列表,默认取第一个 |
| `dcp_create_point` | POST | `/api/app/point` | 创建测点 |
| `dcp_bind_inclino` | POST | `/api/app/inclino-sensor` | 绑定倾角传感器 |
| `dcp_bind_stress` | POST | `/api/app/stress-sensor` | 绑定轴力计 |
| `dcp_bind_hyeto` | POST | `/api/app/hyeto-sensor` | 绑定雨量计 |
| `dcp_bind_static_level` | POST | `/api/app/static-level-device` | 绑定静力水准 |
| `dcp_push_inclino` | POST | `/api/app/inclino-data/many` | 推送倾角数据 |
| `dcp_push_stress` | POST | `/api/app/stress-data/many` | 推送轴力计数据 |
| `dcp_push_hyeto` | POST | `/api/app/hyeto-data/many` | 推送雨量计数据 |
| `dcp_push_static_level` | POST | `/api/app/static-level-data/many` | 推送静力水准数据 |
### 7.2 内置流程( 6 个)
| code | flow_type | sensor_type | 步骤 |
|---|---:|---|---|
| `dcp_init` | 1 | - | `dcp_auth` -> `dcp_get_user` -> `dcp_get_projects` |
| `dcp_token_refresh` | 3 | - | `dcp_auth` |
| `dcp_onboard_inclino` | 2 | `inclino` | `dcp_create_point` -> `dcp_bind_inclino` |
| `dcp_onboard_stress` | 2 | `stress` | `dcp_create_point` -> `dcp_bind_stress` |
| `dcp_onboard_hyeto` | 2 | `hyeto` | `dcp_create_point` -> `dcp_bind_hyeto` |
| `dcp_onboard_static_level` | 2 | `static_level` | `dcp_create_point` -> `dcp_bind_static_level` |
### 7.3 DCP 字段约定
- 入网接口中的 `moduleId` 、`virtualId` 使用 `source=uuid` ,每次请求生成新 UUID。
- 静力水准绑定接口的 `sensorIds` 使用 `source=referenceSubDeviceSensorIdPairs` 。
- 静力水准推送接口定义了数组字段 `sensorData` ,其 `itemSchema` 包含 `epoch` 、`height`、`temperature`、`sensorId`。
- 轴力计和雨量计 DCP 文档要求累计值递增;当前代码未在适配器中实现“查询上次成功值并跳过非递增值”的通用校验,若目标平台严格校验,需要在推送前补充该逻辑或依赖平台返回失败日志。
---
## 8. 关键复用点
| 现有资产 | 位置 | 用途 |
| 资产 | 位置 | 用途 |
|---|---|---|
| `HttpClientFactory` | `fastbee-plugs/fastbee-http` | 发起 HTTP 请求 |
| `HttpClientConfig` | `fastbee-plugs/fastbee-http` | 请求配置 DTO |
| `RuleProcess` | `fastbee-iot-service` | 设备上报触发挂接点( event=1) |
| `sys_job` / Quartz | `fastbee-plugs/quartz` | 定时批量推送 Job |
| `iot_device_log` | DB | 批量推送数据来源 |
| `iot_things_model` | DB | 设备绑定页面物模型字段下拉 |
| MyBatis-Plus ServiceImpl | 已有 | 所有 Service 基类 |
| `HttpClientConfig` | `fastbee-plugs/fastbee-http` | 构建请求 URL、method、headers、body/query |
| Forest HTTP | `com.dtflys.forest` | 底层 HTTP 执行 |
| `RuleProcess` 调用方 | IoT 上报处理链路 | 设备上报后触发 `DeviceReportTrigger` |
| Redis | `StaticLevelBufferService` | 静力水准基准点/监测点缓存 |
| MyBatis-Plus `ServiceImpl` | 平台 service 实现 | 常规 CRUD |
| `iot_things_model` | 物模型表 | 前端字段映射可选项来源 |
---
## 9. 当前实现限制与注意事项
- 当前适配器只有 DCP。新增平台需要新增 adapter, 并在 `DeviceReportTrigger.dispatchPush` 中按平台 `code` 路由。
- 文档中的“可视化配置”已覆盖平台、接口、流程、设备绑定、子映射和日志,但平台适配器路由仍需要代码扩展。
- 实时推送只支持 POST JSON, 流程执行支持 GET、form、JSON。
- `idempotency_key` 字段已建表,但实时推送未生成幂等键。
- `PlatformInstanceController.getInfo` 只清空 `access_token` ,注释提到密码脱敏,但当前代码未改写 `auth_config` 内密码。
- `platform_device_mapping.platform_device_id` 对网关设备不是主要数据来源;网关 monitor 的平台 deviceId 保存在 `platform_device_sub_mapping.platform_device_id` 。
- 静力水准 monitor 先到时会缓存 monitor 数据,但当前主推送分支是在 monitor 到达时读取基准点缓存;若要实现“基准点后到自动补推所有等待 monitor”, 需要在基准点上报分支补充扫描 monitor 缓存并推送。