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 9017cb60..4d4f6b1c 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 @@ -51,12 +51,15 @@ import java.util.Map; @Component public class DcpPlatformAdapter { - private static final DateTimeFormatter ISO_OFFSET_MILLIS = - DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSXXX"); - private static final DateTimeFormatter DATE_TIME_MILLIS = - DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS"); - private static final DateTimeFormatter DATE_TIME_SECONDS = - DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); + // 组包形势 + // 成对 + private static final String ARRAY_MODE_REFERENCE_PAIR = "reference_pair"; + // 单个 + private static final String ARRAY_MODE_SINGLE = "single"; + + private static final DateTimeFormatter ISO_OFFSET_MILLIS = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSXXX"); + private static final DateTimeFormatter DATE_TIME_MILLIS = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS"); + private static final DateTimeFormatter DATE_TIME_SECONDS = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); @Autowired private IPlatformInstanceService instanceService; @@ -105,7 +108,8 @@ public class DcpPlatformAdapter { * 推送网关子设备上报数据到 DCP *
* deviceId 直接从子映射行获取,不再解析 JSON。
- * reference 角色仅缓存基准数据,monitor 角色组装 [refItem, monitorItem] 推送。
+ * 数组字段由 requestSchema.arrayMode 控制组包方式:
+ * single 只推当前子设备,reference_pair 组装 [refItem, monitorItem] 推送。
*/
public void pushGatewayData(PlatformInstance platform, PlatformDeviceMapping mapping,
PlatformDeviceSubMapping subMapping,
@@ -140,15 +144,18 @@ public class DcpPlatformAdapter {
// 解析 extra_mapping 中的 itemFieldMappings 和 reportIntervalSec
int reportIntervalSec = 300;
JSONArray itemFieldMappings = null;
- String fieldName = "sensorData";
+ String fieldName = resolveArrayFieldName(requestSchema, null);
+ JSONObject extra = null;
if (StringUtils.hasText(mapping.getExtraMapping())) {
- JSONObject extra = JSON.parseObject(mapping.getExtraMapping());
+ extra = JSON.parseObject(mapping.getExtraMapping());
reportIntervalSec = extra.getIntValue("reportIntervalSec", 300);
itemFieldMappings = extra.getJSONArray("itemFieldMappings");
if (StringUtils.hasText(extra.getString("fieldName"))) {
fieldName = extra.getString("fieldName");
}
}
+ JSONObject arrayFieldSchema = findArrayFieldSchema(requestSchema, fieldName);
+ String arrayMode = resolveArrayMode(arrayFieldSchema, extra);
// 从 onboard_config 取 sensorId
String sensorId = reportSerialNumber;
@@ -170,6 +177,13 @@ public class DcpPlatformAdapter {
JSONObject currentBodyFields = buildObjectBySchema(platform, mapping, requestSchema, reportData, fieldMappings);
+ // 单点数组推送:sensorData 中只包含当前上报子设备这一条数据。
+ if (ARRAY_MODE_SINGLE.equals(arrayMode)) {
+ pushSingleArrayData(platform, mapping, subMapping, reportSerialNumber, baseUrl,
+ interfaceCode, fieldName, currentItem, currentBodyFields);
+ return;
+ }
+
if (subMapping.isReference()) {
// 基准点上报:缓存当前基准点,并补推此前已缓存的监测点
bufferService.saveReferenceData(
@@ -216,6 +230,25 @@ public class DcpPlatformAdapter {
}
}
+ private void pushSingleArrayData(PlatformInstance platform, PlatformDeviceMapping mapping,
+ PlatformDeviceSubMapping subMapping, String reportSerialNumber,
+ String baseUrl, String interfaceCode, String fieldName,
+ JSONObject currentItem, JSONObject currentBodyFields) {
+ String deviceId = subMapping.getPlatformDeviceId();
+ if (!StringUtils.hasText(deviceId)) {
+ log.warn("[DCP] 子设备 {} 未分配平台设备ID,跳过单点数组推送", reportSerialNumber);
+ return;
+ }
+ JSONArray sensorData = new JSONArray();
+ sensorData.add(currentItem);
+ JSONArray pushBody = buildArrayPushBody(mapping, currentBodyFields, fieldName, sensorData, deviceId);
+
+ PlatformPushLog logEntry = createLogEntry(platform, mapping, interfaceCode, reportSerialNumber);
+ doHttpPost(platform, baseUrl, JSON.toJSONString(pushBody), logEntry);
+ pushLogService.asyncSave(logEntry);
+ log.info("[DCP] 子设备 {} 单点数组已推送 deviceId={}", reportSerialNumber, deviceId);
+ }
+
/**
* 请求 schema 是否包含数组字段
*/
@@ -232,6 +265,33 @@ public class DcpPlatformAdapter {
return false;
}
+ private String resolveArrayFieldName(JSONArray requestSchema, String fallback) {
+ if (StringUtils.hasText(fallback)) {
+ return fallback;
+ }
+ JSONObject firstArrayField = findFirstArrayFieldSchema(requestSchema);
+ if (firstArrayField != null && StringUtils.hasText(firstArrayField.getString("name"))) {
+ return firstArrayField.getString("name");
+ }
+ return "sensorData";
+ }
+
+ private String resolveArrayMode(JSONObject arrayFieldSchema, JSONObject extra) {
+ String mode = null;
+ if (extra != null && StringUtils.hasText(extra.getString("arrayMode"))) {
+ mode = extra.getString("arrayMode");
+ } else if (arrayFieldSchema != null && StringUtils.hasText(arrayFieldSchema.getString("arrayMode"))) {
+ mode = arrayFieldSchema.getString("arrayMode");
+ }
+ if (!StringUtils.hasText(mode)) {
+ return ARRAY_MODE_REFERENCE_PAIR;
+ }
+ if (ARRAY_MODE_SINGLE.equals(mode) || ARRAY_MODE_REFERENCE_PAIR.equals(mode)) {
+ return mode;
+ }
+ throw new RuntimeException("不支持的数组组包模式 arrayMode=" + mode);
+ }
+
/** 解析基准点序列号 */
private String resolveRefSerialNumber(PlatformDeviceSubMapping subMapping, PlatformDeviceMapping mapping) {
try {
@@ -633,6 +693,19 @@ public class DcpPlatformAdapter {
return null;
}
+ private JSONObject findFirstArrayFieldSchema(JSONArray requestSchema) {
+ if (requestSchema == null || requestSchema.isEmpty()) {
+ return null;
+ }
+ for (int i = 0; i < requestSchema.size(); i++) {
+ JSONObject schemaField = requestSchema.getJSONObject(i);
+ if ("array".equalsIgnoreCase(schemaField.getString("type"))) {
+ return schemaField;
+ }
+ }
+ return null;
+ }
+
private JSONObject copyJsonObject(JSONObject source) {
if (source == null || source.isEmpty()) {
return new JSONObject();
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 aa9d68c4..1887e701 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
@@ -295,7 +295,7 @@ public class PlatformFlowExecutor {
if (StringUtils.hasText(ctx.getCurrentMonitorSerialNumber())) {
return ctx.getCurrentMonitorSerialNumber();
}
- return resolveGatewaySerialNumber(ctx.getDeviceMapping());
+ return normalizeValueByType(fd.getString("type"), resolveGatewaySerialNumber(ctx.getDeviceMapping()));
}
if ("gateway_serial_number".equals(df)) {
return resolveGatewaySerialNumber(ctx.getDeviceMapping());
diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformDeviceMappingServiceImpl.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformDeviceMappingServiceImpl.java
index 4cd95611..0036ad56 100644
--- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformDeviceMappingServiceImpl.java
+++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformDeviceMappingServiceImpl.java
@@ -127,13 +127,7 @@ public class PlatformDeviceMappingServiceImpl extends ServiceImpl