From 99935242ecccd94e76978e2b71cbc46b87d31dc6 Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期五, 09 十月 2026 11:17:44 +0800
Subject: [PATCH] 推式来源的工步与循环改为从采样段推导
---
battery-business/src/test/java/com/battery/business/service/impl/FakeIngestLanding.java | 28 ++++++++++++++++++++++++++++
1 files changed, 28 insertions(+), 0 deletions(-)
diff --git a/battery-business/src/test/java/com/battery/business/service/impl/FakeIngestLanding.java b/battery-business/src/test/java/com/battery/business/service/impl/FakeIngestLanding.java
index 8e8923e..21b6586 100644
--- a/battery-business/src/test/java/com/battery/business/service/impl/FakeIngestLanding.java
+++ b/battery-business/src/test/java/com/battery/business/service/impl/FakeIngestLanding.java
@@ -32,6 +32,22 @@
return 1;
}
+ /** 拓扑是声明：撞键即刷新声明内容，不追加新行（与生产 SQL 的 upsert 语义一致）。 */
+ @Override
+ public int upsertTopology(TDeviceIngestRecord record)
+ {
+ int index = -1;
+ for (int i = 0; i < rows.size(); i++)
+ if (same(rows.get(i), record)) index = i;
+ if (index < 0)
+ {
+ rows.add(record);
+ return 1;
+ }
+ rows.set(index, record);
+ return 1;
+ }
+
@Override
public String selectIngestRecordHashByKey(String adapterCode, String payloadKind, String recordKey)
{
@@ -59,6 +75,7 @@
TDeviceIngestChannel channel = new TDeviceIngestChannel();
channel.setDevtype(row.getDevtype()); channel.setDevid(row.getDevid());
channel.setSubdevid(row.getSubdevid()); channel.setChlid(row.getChlid());
+ channel.setVendorChannelNo(row.getVendorChannelNo());
channel.setLastDataAt(rows.stream().filter(sample -> isSample(sample, adapterCode, row))
.map(TDeviceIngestRecord::getDeviceTimeUtc).filter(java.util.Objects::nonNull)
.max(Date::compareTo).orElse(null));
@@ -80,6 +97,17 @@
}
@Override
+ public List<TDeviceIngestRecord> selectSamplesOfSource(String adapterCode, int devtype, int devid, int subdevid,
+ int chlid, String sourceRef)
+ {
+ return samples(adapterCode, devtype, devid, subdevid, chlid).stream()
+ .filter(row -> sourceRef.equals(row.getSourceRef()))
+ .sorted(java.util.Comparator.comparing(TDeviceIngestRecord::getDeviceTimeUtc)
+ .thenComparing(TDeviceIngestRecord::getRecordSequence))
+ .toList();
+ }
+
+ @Override
public List<TDeviceIngestRecord> selectSamplesSince(String adapterCode, int devtype, int devid, int subdevid,
int chlid, Date watermarkTime, long watermarkSequence, int limit)
{
--
Gitblit v1.7.1