package com.battery.business.service.impl;
|
|
import java.util.ArrayList;
|
import java.util.Date;
|
import java.util.HexFormat;
|
import java.util.List;
|
import com.battery.business.domain.TDeviceIngestRecord;
|
import com.battery.business.domain.vo.TDeviceIngestChannel;
|
import com.battery.business.mapper.TDeviceIngestRecordMapper;
|
|
/**
|
* 落地表的内存替身：按来源行身份判重、心跳按来源覆盖、水位按 (设备时间, 协议序号) 取严格更晚的行——
|
* 与生产 Mapper 的 SQL 语义一致，使接收端点与适配器可以在没有数据库的情况下接起来测。
|
*/
|
final class FakeIngestLanding implements TDeviceIngestRecordMapper
|
{
|
private final List<TDeviceIngestRecord> rows = new ArrayList<>();
|
|
@Override
|
public int insertIngestRecord(TDeviceIngestRecord record)
|
{
|
if (rows.stream().anyMatch(existing -> same(existing, record))) return 0;
|
rows.add(record);
|
return 1;
|
}
|
|
@Override
|
public int upsertHeartbeat(TDeviceIngestRecord record)
|
{
|
rows.removeIf(existing -> same(existing, record));
|
rows.add(record);
|
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)
|
{
|
return rows.stream().filter(row -> same(row, adapterCode, payloadKind, recordKey)).findFirst()
|
.map(row -> HexFormat.of().formatHex(row.getContentSha256())).orElse(null);
|
}
|
|
@Override
|
public Date selectHeartbeatAt(String adapterCode)
|
{
|
return rows.stream()
|
.filter(row -> adapterCode.equals(row.getAdapterCode())
|
&& TDeviceIngestRecord.KIND_HEARTBEAT.equals(row.getPayloadKind()))
|
.map(TDeviceIngestRecord::getReceivedAtUtc).max(Date::compareTo).orElse(null);
|
}
|
|
@Override
|
public List<TDeviceIngestChannel> selectTopologyChannels(String adapterCode)
|
{
|
List<TDeviceIngestChannel> channels = new ArrayList<>();
|
for (TDeviceIngestRecord row : rows)
|
{
|
if (!adapterCode.equals(row.getAdapterCode())
|
|| !TDeviceIngestRecord.KIND_TOPOLOGY.equals(row.getPayloadKind())) continue;
|
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));
|
channels.add(channel);
|
}
|
channels.sort(java.util.Comparator.comparing(TDeviceIngestChannel::getDevid)
|
.thenComparing(TDeviceIngestChannel::getSubdevid).thenComparing(TDeviceIngestChannel::getChlid));
|
return channels;
|
}
|
|
@Override
|
public TDeviceIngestRecord selectLatestSample(String adapterCode, int devtype, int devid, int subdevid,
|
int chlid)
|
{
|
return samples(adapterCode, devtype, devid, subdevid, chlid).stream()
|
.max(java.util.Comparator.comparing(TDeviceIngestRecord::getDeviceTimeUtc)
|
.thenComparing(TDeviceIngestRecord::getRecordSequence))
|
.orElse(null);
|
}
|
|
@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)
|
{
|
List<TDeviceIngestRecord> later = samples(adapterCode, devtype, devid, subdevid, chlid).stream()
|
.filter(row -> after(row, watermarkTime, watermarkSequence))
|
.sorted(java.util.Comparator.comparing(TDeviceIngestRecord::getDeviceTimeUtc)
|
.thenComparing(TDeviceIngestRecord::getRecordSequence))
|
.toList();
|
return later.size() > limit ? later.subList(0, limit) : later;
|
}
|
|
/** 全部落地行，供断言"拒绝时没有落任何数据"使用。 */
|
List<TDeviceIngestRecord> rows() { return List.copyOf(rows); }
|
|
List<TDeviceIngestRecord> samples(String adapterCode, int devtype, int devid, int subdevid, int chlid)
|
{
|
return rows.stream().filter(row -> adapterCode.equals(row.getAdapterCode())
|
&& TDeviceIngestRecord.KIND_SAMPLE.equals(row.getPayloadKind())
|
&& row.getDevtype() == devtype && row.getDevid() == devid && row.getSubdevid() == subdevid
|
&& row.getChlid() == chlid).toList();
|
}
|
|
private static boolean isSample(TDeviceIngestRecord row, String adapterCode, TDeviceIngestRecord channel)
|
{
|
return adapterCode.equals(row.getAdapterCode()) && TDeviceIngestRecord.KIND_SAMPLE.equals(row.getPayloadKind())
|
&& row.getDevtype().equals(channel.getDevtype()) && row.getDevid().equals(channel.getDevid())
|
&& row.getSubdevid().equals(channel.getSubdevid()) && row.getChlid().equals(channel.getChlid());
|
}
|
|
private static boolean after(TDeviceIngestRecord row, Date watermarkTime, long watermarkSequence)
|
{
|
if (row.getDeviceTimeUtc() == null) return false;
|
if (watermarkTime == null) return true;
|
int byTime = row.getDeviceTimeUtc().compareTo(watermarkTime);
|
return byTime > 0 || (byTime == 0 && row.getRecordSequence() > watermarkSequence);
|
}
|
|
private static boolean same(TDeviceIngestRecord row, TDeviceIngestRecord other)
|
{
|
return same(row, other.getAdapterCode(), other.getPayloadKind(), other.getRecordKey());
|
}
|
|
private static boolean same(TDeviceIngestRecord row, String adapterCode, String payloadKind, String recordKey)
|
{
|
return adapterCode.equals(row.getAdapterCode()) && payloadKind.equals(row.getPayloadKind())
|
&& recordKey.equals(row.getRecordKey());
|
}
|
}
|
