| | |
|---|
| | | if (adapter == null) return; TTestChannelBinding binding = battery.getBinding(); if (binding == null || binding.getDevid() == null || binding.getChlid() == null) return; |
|---|
| | | String key = test.getTestId() + ":" + battery.getTestBatteryId(); |
|---|
| | | // 接收调度每秒跑一次，而来源按自己的写库节奏产出工步：不按来源节奏限流会每秒重查厂商库。 |
|---|
| | | // 汇总用来源声明的汇总间隔：它可能比实时行的间隔长（推式来源的落地表在本地）。 |
|---|
| | | Instant now = clock.instant(); Instant previous = lastAggregateAt.get(key); |
|---|
| | | if (previous != null && now.isBefore(previous.plusMillis(persistenceInterval(binding)))) return; |
|---|
| | | if (previous != null && now.isBefore(previous.plusMillis(aggregateInterval(binding)))) return; |
|---|
| | | lastAggregateAt.put(key, now); |
|---|
| | | // STEP 2: 读取来源的工步与循环汇总；来源不可达时本轮放弃，下一轮重试。 |
|---|
| | | ChannelIdentity channel = new ChannelIdentity(binding.getAdapterCode(), binding.getDevtype(), binding.getDevid(), binding.getSubdevid(), binding.getChlid()); |
|---|
| | |
|---|
| | | return String.join(":", parts); |
|---|
| | | } |
|---|
| | | private long persistenceInterval(TTestChannelBinding b) { try { return Math.max(100L, adapter.persistenceIntervalMillis(channelIdentity(b))); } catch (RuntimeException ex) { return 60000L; } } |
|---|
| | | private long aggregateInterval(TTestChannelBinding b) { try { return Math.max(100L, adapter.aggregateIntervalMillis(channelIdentity(b))); } catch (RuntimeException ex) { return 60000L; } } |
|---|
| | | private static String sourceIdentity(TTestChannelBinding b) { return String.join(":", String.valueOf(b.getAdapterCode()), String.valueOf(b.getDevtype()), String.valueOf(b.getDevid()), String.valueOf(b.getSubdevid()), String.valueOf(b.getChlid())); } |
|---|
| | | private static String timezone(TTestChannelBinding b) { return b.getAdapterTimezoneId() == null || b.getAdapterTimezoneId().isBlank() ? "Asia/Hong_Kong" : b.getAdapterTimezoneId(); } |
|---|
| | | private static String blankToNull(String value) { return value == null || value.isBlank() ? null : value; } |
|---|
