diff --git a/fastbee-common/src/main/java/com/fastbee/common/config/DeviceTask.java b/fastbee-common/src/main/java/com/fastbee/common/config/DeviceTask.java index be51c4db..eeb46936 100644 --- a/fastbee-common/src/main/java/com/fastbee/common/config/DeviceTask.java +++ b/fastbee-common/src/main/java/com/fastbee/common/config/DeviceTask.java @@ -97,6 +97,12 @@ public class DeviceTask { return builder(FastBeeConstant.TASK.TRIGGER_POWER_ON_TASK_ASYNC); } + /*第三方平台推送*/ + @Bean(FastBeeConstant.TASK.PUSH_TO_THIRD_PLATFORM_ASYNC) + public Executor pushToThirdPlatformExecutor() { + return builder(FastBeeConstant.TASK.PUSH_TO_THIRD_PLATFORM_ASYNC); + } + /*组装线程池*/ private ThreadPoolTaskExecutor builder(String threadNamePrefix) { diff --git a/fastbee-common/src/main/java/com/fastbee/common/constant/FastBeeConstant.java b/fastbee-common/src/main/java/com/fastbee/common/constant/FastBeeConstant.java index b3a1aff4..48d88c51 100644 --- a/fastbee-common/src/main/java/com/fastbee/common/constant/FastBeeConstant.java +++ b/fastbee-common/src/main/java/com/fastbee/common/constant/FastBeeConstant.java @@ -99,6 +99,9 @@ public interface FastBeeConstant { // 设备上电即触发异步任务 TriggerPowerOnTaskAsync String TRIGGER_POWER_ON_TASK_ASYNC = "triggerPowerOnTaskAsync"; + + // 设备上报数据-推送到第三方平台 + String PUSH_TO_THIRD_PLATFORM_ASYNC = "pushToThirdPlatformAsync"; } interface MQTT { diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DataHandlerImpl.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DataHandlerImpl.java index 31aa7e89..9fa9938b 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DataHandlerImpl.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DataHandlerImpl.java @@ -20,6 +20,7 @@ import com.fastbee.common.utils.gateway.mq.TopicsUtils; import com.fastbee.iot.cache.SceneModelTagCache; import com.fastbee.iot.data.executor.ConditionTriggerTaskExecutor; import com.fastbee.iot.data.service.IDataHandler; +import com.fastbee.iot.platform.trigger.DeviceReportTrigger; import com.fastbee.iot.data.service.IMqttMessagePublish; import com.fastbee.iot.data.service.IRuleEngine; import com.fastbee.iot.domain.*; @@ -84,6 +85,9 @@ public class DataHandlerImpl implements IDataHandler { @Resource private IFunctionLogService functionLogService; + @Resource + private DeviceReportTrigger deviceReportTrigger; + /** * 上报属性或功能处理 * @@ -116,6 +120,8 @@ public class DataHandlerImpl implements IDataHandler { // payload 与 MQTT 上报格式一致:ThingsModelSimpleItem JSON 数组(含计算物模型) String bridgePayload = JSON.toJSONString(result); ruleProcess.processRuleScript(serialNumber, 1, "tcp/property/post", bridgePayload); + // 平台对接推送(异步,不阻塞主链路) + deviceReportTrigger.onDeviceReport(serialNumber, bridgePayload); } catch (Exception e) { log.error("数据桥接规则引擎执行异常 sn={}, msg={}", serialNumber, e.getMessage(), e); } diff --git a/fastbee-open-api/src/main/java/com/fastbee/controller/platform/PlatformPushLogController.java b/fastbee-open-api/src/main/java/com/fastbee/controller/platform/PlatformPushLogController.java index 242d5877..9744a971 100644 --- a/fastbee-open-api/src/main/java/com/fastbee/controller/platform/PlatformPushLogController.java +++ b/fastbee-open-api/src/main/java/com/fastbee/controller/platform/PlatformPushLogController.java @@ -33,4 +33,12 @@ public class PlatformPushLogController extends BaseController { public AjaxResult getInfo(@PathVariable Long id) { return success(pushLogService.getById(id)); } + + @PreAuthorize("@ss.hasPermi('platform:pushLog:edit')") + @PostMapping("/{id}/retry") + @ApiOperation("手动重推") + public AjaxResult retry(@PathVariable Long id) { + pushLogService.retry(id); + return success("重推成功"); + } } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/adapter/dcp/DcpPlatformAdapter.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/adapter/dcp/DcpPlatformAdapter.java index b9e061da..41dd215b 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/adapter/dcp/DcpPlatformAdapter.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/adapter/dcp/DcpPlatformAdapter.java @@ -71,6 +71,7 @@ public class DcpPlatformAdapter { logEntry.setInterfaceCode("dcp_push_" + sensorType); logEntry.setStatus(0); logEntry.setRetryCount(0); + logEntry.setLogType(1); // 数据推送日志 try { // 1. 字段映射转换 @@ -126,7 +127,12 @@ public class DcpPlatformAdapter { return null; } JSONArray fieldMappings = JSON.parseArray(mapping.getFieldMapping()); - JSONObject reportData = JSON.parseObject(payload); + + // payload 可能是数组 [{id,value,ts},...] 或对象 {key:value,...} + JSONObject reportData = parsePayloadToMap(payload); + if (reportData == null || reportData.isEmpty()) { + return null; + } JSONObject dcpData = new JSONObject(); dcpData.put("deviceId", mapping.getPlatformDeviceId()); @@ -158,6 +164,35 @@ public class DcpPlatformAdapter { return arr; } + /** + * 将设备上报 payload 解析为 {identifier: value} 的扁平 map + * 支持两种格式: + * 1. 数组: [{"id":"AngX","value":"-1.55","ts":"..."},...] + * 2. 对象: {"AngX":"-1.55","T":"21.6",...} + */ + private JSONObject parsePayloadToMap(String payload) { + String trimmed = payload.trim(); + if (trimmed.startsWith("[")) { + JSONArray arr = JSON.parseArray(trimmed); + JSONObject map = new JSONObject(); + for (int i = 0; i < arr.size(); i++) { + JSONObject item = arr.getJSONObject(i); + String id = item.getString("id"); + if (id == null) id = item.getString("identifier"); + if (id == null) continue; + Object value = item.get("value"); + map.put(id, value); + // 同时保存 ts 供时间转换使用 + if (item.containsKey("ts")) { + map.put(id + "__ts", item.get("ts")); + } + } + return map; + } else { + return JSON.parseObject(trimmed); + } + } + /** 确保 Token 有效,过期则执行 flow_type=3 的 Token 刷新流程 */ private String ensureToken(PlatformInstance platform) { Date now = new Date(); diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/domain/PlatformPushLog.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/domain/PlatformPushLog.java index bd5a704f..230c438b 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/domain/PlatformPushLog.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/domain/PlatformPushLog.java @@ -57,6 +57,9 @@ public class PlatformPushLog implements Serializable { @ApiModelProperty("状态(0=进行中,1=成功,2=失败)") private Integer status; + @ApiModelProperty("日志类型(1=数据推送,2=操作日志)") + private Integer logType; + @ApiModelProperty("错误信息") private String errorMsg; diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/flow/PlatformFlowExecutor.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/flow/PlatformFlowExecutor.java index 07faa88a..d5fa48a1 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/flow/PlatformFlowExecutor.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/flow/PlatformFlowExecutor.java @@ -69,6 +69,7 @@ public class PlatformFlowExecutor { logEntry.setInterfaceCode(iface.getCode()); logEntry.setStatus(0); logEntry.setRetryCount(0); + logEntry.setLogType(2); // 操作日志(认证/入网等流程步骤) try { // 1. 解析入参 diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/IPlatformPushLogService.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/IPlatformPushLogService.java index 9dcdf9f6..6eb0d253 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/IPlatformPushLogService.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/IPlatformPushLogService.java @@ -14,4 +14,7 @@ public interface IPlatformPushLogService extends IService { /** 查询某设备某接口最后一次成功推送的响应体(用于递增约束校验) */ String selectLastSuccessResponseBody(Long deviceId, String interfaceCode); + + /** 手动重推:重新发送已记录的请求,更新日志状态 */ + void retry(Long id); } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformPushLogServiceImpl.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformPushLogServiceImpl.java index ad102f42..89bb03be 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformPushLogServiceImpl.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformPushLogServiceImpl.java @@ -3,7 +3,13 @@ package com.fastbee.iot.platform.service.impl; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.baomidou.mybatisplus.core.toolkit.Wrappers; import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl; +import com.dtflys.forest.http.ForestRequest; +import com.dtflys.forest.http.ForestRequestType; +import com.fastbee.http.model.HttpClientConfig; +import com.fastbee.http.service.HttpClientFactory; +import com.fastbee.iot.platform.domain.PlatformInstance; import com.fastbee.iot.platform.domain.PlatformPushLog; +import com.fastbee.iot.platform.mapper.PlatformInstanceMapper; import com.fastbee.iot.platform.mapper.PlatformPushLogMapper; import com.fastbee.iot.platform.service.IPlatformPushLogService; import org.springframework.beans.factory.annotation.Autowired; @@ -11,7 +17,9 @@ import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Service; import org.springframework.util.StringUtils; +import java.util.HashMap; import java.util.List; +import java.util.Map; @Service public class PlatformPushLogServiceImpl extends ServiceImpl @@ -19,6 +27,8 @@ public class PlatformPushLogServiceImpl extends ServiceImpl selectList(PlatformPushLog query) { @@ -26,6 +36,7 @@ public class PlatformPushLogServiceImpl extends ServiceImpl headers = new HashMap<>(); + headers.put("Content-Type", "application/json"); + if (StringUtils.hasText(token)) { + headers.put("Authorization", "Bearer " + token); + } + + HttpClientConfig config = HttpClientConfig.builder() + .url(log.getRequestUrl()) + .method(ForestRequestType.POST) + .headers(headers) + .body(log.getRequestBody()) + .build(); + + ForestRequest request = HttpClientFactory.instance(config); + String responseBody = request.executeAsString(); + + PlatformPushLog success = new PlatformPushLog(); + success.setId(id); + success.setStatus(1); + success.setResponseBody(responseBody); + success.setResponseCode(200); + success.setCostMs((int) (System.currentTimeMillis() - startMs)); + this.updateById(success); + + } catch (Exception e) { + PlatformPushLog fail = new PlatformPushLog(); + fail.setId(id); + fail.setStatus(2); + fail.setErrorMsg(e.getMessage()); + fail.setCostMs((int) (System.currentTimeMillis() - startMs)); + this.updateById(fail); + throw new RuntimeException("重推失败: " + e.getMessage(), e); + } + } } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/trigger/DeviceReportTrigger.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/trigger/DeviceReportTrigger.java index 87661aca..9305e014 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/trigger/DeviceReportTrigger.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/trigger/DeviceReportTrigger.java @@ -1,5 +1,6 @@ package com.fastbee.iot.platform.trigger; +import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.iot.platform.adapter.dcp.DcpPlatformAdapter; import com.fastbee.iot.platform.domain.PlatformDeviceMapping; import com.fastbee.iot.platform.domain.PlatformInstance; @@ -36,7 +37,7 @@ public class DeviceReportTrigger { * @param serialNumber 设备序列号 * @param payload 上报数据 JSON */ - @Async + @Async(FastBeeConstant.TASK.PUSH_TO_THIRD_PLATFORM_ASYNC) public void onDeviceReport(String serialNumber, String payload) { try { List mappings = mappingService.selectBySerialNumber(serialNumber); diff --git a/sql/platform-dcp-init-data.sql b/sql/platform-dcp-init-data.sql index 45868bd7..2660958b 100644 --- a/sql/platform-dcp-init-data.sql +++ b/sql/platform-dcp-init-data.sql @@ -236,7 +236,9 @@ SET @flow_inclino_id = LAST_INSERT_ID(); INSERT INTO `platform_flow_step` (`flow_id`,`step_order`,`interface_id`,`result_key`,`result_save`,`remark`) VALUES (@flow_inclino_id, 1, (SELECT id FROM platform_interface WHERE platform_id=@dcp_platform_id AND code='dcp_create_point' LIMIT 1), - 'step1', NULL, '创建测点,获取pointId'); + 'step1', + '[{"field":"id","saveTo":"device_mapping.platform_point_id"}]', + '创建测点,获取pointId并回填'); INSERT INTO `platform_flow_step` (`flow_id`,`step_order`,`interface_id`,`result_key`,`result_save`,`remark`) VALUES (@flow_inclino_id, 2, @@ -252,10 +254,12 @@ VALUES (@dcp_platform_id, 'DCP轴力计入网', 'dcp_onboard_stress', 2, 'stress SET @flow_stress_id = LAST_INSERT_ID(); -INSERT INTO `platform_flow_step` (`flow_id`,`step_order`,`interface_id`,`result_key`,`remark`) +INSERT INTO `platform_flow_step` (`flow_id`,`step_order`,`interface_id`,`result_key`,`result_save`,`remark`) VALUES (@flow_stress_id, 1, (SELECT id FROM platform_interface WHERE platform_id=@dcp_platform_id AND code='dcp_create_point' LIMIT 1), - 'step1', '创建测点'); + 'step1', + '[{"field":"id","saveTo":"device_mapping.platform_point_id"}]', + '创建测点'); INSERT INTO `platform_flow_step` (`flow_id`,`step_order`,`interface_id`,`result_key`,`result_save`,`remark`) VALUES (@flow_stress_id, 2, @@ -271,10 +275,12 @@ VALUES (@dcp_platform_id, 'DCP雨量计入网', 'dcp_onboard_hyeto', 2, 'hyeto', SET @flow_hyeto_id = LAST_INSERT_ID(); -INSERT INTO `platform_flow_step` (`flow_id`,`step_order`,`interface_id`,`result_key`,`remark`) +INSERT INTO `platform_flow_step` (`flow_id`,`step_order`,`interface_id`,`result_key`,`result_save`,`remark`) VALUES (@flow_hyeto_id, 1, (SELECT id FROM platform_interface WHERE platform_id=@dcp_platform_id AND code='dcp_create_point' LIMIT 1), - 'step1', '创建测点'); + 'step1', + '[{"field":"id","saveTo":"device_mapping.platform_point_id"}]', + '创建测点'); INSERT INTO `platform_flow_step` (`flow_id`,`step_order`,`interface_id`,`result_key`,`result_save`,`remark`) VALUES (@flow_hyeto_id, 2, diff --git a/sql/platform-integration.sql b/sql/platform-integration.sql index 557164be..68a84dca 100644 --- a/sql/platform-integration.sql +++ b/sql/platform-integration.sql @@ -128,6 +128,7 @@ CREATE TABLE IF NOT EXISTS `platform_push_log` ( `status` tinyint(1) NOT NULL DEFAULT 0 COMMENT '状态(0=进行中,1=成功,2=失败)', `error_msg` varchar(1000) DEFAULT NULL COMMENT '错误信息', `idempotency_key` varchar(100) DEFAULT NULL COMMENT '幂等键,防重复推送', + `log_type` tinyint(1) NOT NULL DEFAULT 1 COMMENT '日志类型(1=数据推送,2=操作日志)', `retry_count` int(11) NOT NULL DEFAULT 0 COMMENT '重试次数', `cost_ms` int(11) DEFAULT NULL COMMENT '耗时毫秒', `create_time` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', diff --git a/yack-web/src/api/platform/pushLog.js b/yack-web/src/api/platform/pushLog.js index 7199632f..084a688e 100644 --- a/yack-web/src/api/platform/pushLog.js +++ b/yack-web/src/api/platform/pushLog.js @@ -7,3 +7,7 @@ export function listPlatformPushLog(params) { export function getPlatformPushLog(id) { return request({ url: `/platform/push-log/${id}`, method: 'get' }) } + +export function retryPlatformPushLog(id) { + return request({ url: `/platform/push-log/${id}/retry`, method: 'post' }) +} diff --git a/yack-web/src/views/platform/push-log/index.vue b/yack-web/src/views/platform/push-log/index.vue index 121a5648..be7149a2 100644 --- a/yack-web/src/views/platform/push-log/index.vue +++ b/yack-web/src/views/platform/push-log/index.vue @@ -28,13 +28,18 @@ + + + + + - +