| | |
|---|
| | | 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) |
|---|
| | | { |
|---|
| | |
|---|
| | | 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)); |
|---|
