From ec3308617909c8d6280e0d00ffb18350e372fbdf Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期日, 30 八月 2026 17:36:29 +0800
Subject: [PATCH] 实现采集器断线恢复与来源异常持久化
---
ruoyi-admin/src/test/java/com/ruoyi/config/DevelopmentCollectorStatusSynchronizerTest.java | 7
ruoyi-system/src/main/java/com/ruoyi/system/domain/TTestCollectorExceptionInterval.java | 54 +++++++
ruoyi-system/src/main/java/com/ruoyi/system/service/ICollectorDisconnectRecoveryService.java | 11 +
ruoyi-admin/src/test/java/com/ruoyi/config/FlywayMigrationContractTest.java | 7 +
ruoyi-system/src/main/java/com/ruoyi/system/service/impl/CollectorDisconnectRecoveryServiceImpl.java | 62 ++++++++
ruoyi-admin/src/main/java/com/ruoyi/config/DevelopmentCollectorStatusSynchronizer.java | 18 ++
ruoyi-admin/src/test/java/com/ruoyi/system/service/impl/CollectorDisconnectRecoveryServiceImplTest.java | 67 +++++++++
ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TTestLiveServiceImpl.java | 35 ++++
ruoyi-system/src/main/resources/mapper/system/TTestMapper.xml | 61 ++++++++
ruoyi-admin/src/main/resources/db/migration/V18__collector_disconnect_recovery.sql | 26 +++
ruoyi-system/src/main/java/com/ruoyi/system/mapper/TTestMapper.java | 13 +
ruoyi-system/src/main/java/com/ruoyi/system/domain/TTestCollectorSource.java | 25 +++
12 files changed, 376 insertions(+), 10 deletions(-)
diff --git a/ruoyi-admin/src/main/java/com/ruoyi/config/DevelopmentCollectorStatusSynchronizer.java b/ruoyi-admin/src/main/java/com/ruoyi/config/DevelopmentCollectorStatusSynchronizer.java
index 431da82..e827d08 100644
--- a/ruoyi-admin/src/main/java/com/ruoyi/config/DevelopmentCollectorStatusSynchronizer.java
+++ b/ruoyi-admin/src/main/java/com/ruoyi/config/DevelopmentCollectorStatusSynchronizer.java
@@ -10,6 +10,7 @@
import com.ruoyi.device.discovery.DeviceDiscoveryCandidateStore;
import com.ruoyi.device.discovery.DeviceDiscoveryCandidateStore.CollectorCandidate;
import com.ruoyi.system.mapper.TCollectorMapper;
+import com.ruoyi.system.service.ICollectorDisconnectRecoveryService;
/** 将模拟采集器的发现写入 Redis，并同步已登记采集器的实时在线事实。 */
@Component
@@ -18,14 +19,24 @@
private final BtsDevelopmentSimulator simulator;
private final DeviceDiscoveryCandidateStore candidateStore;
private final TCollectorMapper collectorMapper;
+ private final ICollectorDisconnectRecoveryService disconnectRecovery;
@Autowired
public DevelopmentCollectorStatusSynchronizer(BtsDevelopmentSimulator simulator,
- DeviceDiscoveryCandidateStore candidateStore, TCollectorMapper collectorMapper)
+ DeviceDiscoveryCandidateStore candidateStore, TCollectorMapper collectorMapper,
+ ICollectorDisconnectRecoveryService disconnectRecovery)
{
this.simulator = simulator;
this.candidateStore = candidateStore;
this.collectorMapper = collectorMapper;
+ this.disconnectRecovery = disconnectRecovery;
+ }
+
+ /** 保留单元测试及独立模拟器装配使用的旧构造入口。 */
+ public DevelopmentCollectorStatusSynchronizer(BtsDevelopmentSimulator simulator,
+ DeviceDiscoveryCandidateStore candidateStore, TCollectorMapper collectorMapper)
+ {
+ this(simulator, candidateStore, collectorMapper, (codes, at) -> { });
}
/** 完整替换 dev-sim 采集器快照，删除配置后候选立即消失。 */
@@ -33,7 +44,8 @@
public synchronized void synchronizeOnlineStatus()
{
Instant observedAt = Instant.now();
- List<CollectorCandidate> candidates = simulator.snapshot().collectors().stream()
+ var simulatorState = simulator.snapshot();
+ List<CollectorCandidate> candidates = simulatorState.collectors().stream()
.map(item -> new CollectorCandidate(BtsDevelopmentSimulator.ADAPTER_CODE, item.deviceCode(),
item.ip(), item.mac(), item.name(),
item.manufacturer(), item.model(), item.channelCount(), item.online(), observedAt,
@@ -42,5 +54,7 @@
candidateStore.replaceCollectors(BtsDevelopmentSimulator.ADAPTER_CODE, candidates);
candidates.forEach(candidate -> collectorMapper.updateRegisteredObservation(candidate.deviceCode(),
candidate.online() ? "ONLINE" : "OFFLINE"));
+ disconnectRecovery.reconcile(simulatorState.collectors().stream()
+ .filter(item -> item.online()).map(item -> item.deviceCode()).collect(java.util.stream.Collectors.toSet()), observedAt);
}
}
diff --git a/ruoyi-admin/src/main/resources/db/migration/V18__collector_disconnect_recovery.sql b/ruoyi-admin/src/main/resources/db/migration/V18__collector_disconnect_recovery.sql
new file mode 100644
index 0000000..ddbe27c
--- /dev/null
+++ b/ruoyi-admin/src/main/resources/db/migration/V18__collector_disconnect_recovery.sql
@@ -0,0 +1,26 @@
+-- Ticket11：持久化采集器来源级断线、缺失范围和恢复边界。
+CREATE TABLE t_test_collector_exception_interval (
+ interval_id BIGINT NOT NULL AUTO_INCREMENT,
+ test_id BIGINT NOT NULL,
+ battery_relation_id BIGINT NOT NULL,
+ collector_id BIGINT NOT NULL,
+ collector_channel_code VARCHAR(64) NOT NULL,
+ exception_type VARCHAR(64) NOT NULL DEFAULT 'COLLECTOR_DISCONNECTED',
+ opened_at_utc DATETIME(3) NOT NULL,
+ missing_from_utc DATETIME(3) NOT NULL,
+ recovered_at_utc DATETIME(3) NULL,
+ missing_to_utc DATETIME(3) NULL,
+ interval_status VARCHAR(16) NOT NULL DEFAULT 'OPEN',
+ detail VARCHAR(1000) NULL,
+ PRIMARY KEY (interval_id),
+ KEY idx_collector_exception_test (test_id, interval_status, opened_at_utc),
+ KEY idx_collector_exception_source (collector_id, collector_channel_code, interval_status),
+ CONSTRAINT fk_collector_exception_test FOREIGN KEY (test_id) REFERENCES t_test (test_id),
+ CONSTRAINT fk_collector_exception_battery FOREIGN KEY (battery_relation_id) REFERENCES t_test_battery (relation_id),
+ CONSTRAINT fk_collector_exception_collector FOREIGN KEY (collector_id) REFERENCES t_collector (collector_id),
+ CONSTRAINT chk_collector_exception_status CHECK (interval_status IN ('OPEN', 'CLOSED')),
+ CONSTRAINT chk_collector_exception_range CHECK (missing_to_utc IS NULL OR missing_to_utc >= missing_from_utc)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='采集器来源级断线缺失区间';
+
+ALTER TABLE t_test_collector_exception_interval
+ ADD UNIQUE KEY uk_collector_exception_open (test_id, battery_relation_id, collector_id, collector_channel_code, exception_type, interval_status);
diff --git a/ruoyi-admin/src/test/java/com/ruoyi/config/DevelopmentCollectorStatusSynchronizerTest.java b/ruoyi-admin/src/test/java/com/ruoyi/config/DevelopmentCollectorStatusSynchronizerTest.java
index 042f79a..5473902 100644
--- a/ruoyi-admin/src/test/java/com/ruoyi/config/DevelopmentCollectorStatusSynchronizerTest.java
+++ b/ruoyi-admin/src/test/java/com/ruoyi/config/DevelopmentCollectorStatusSynchronizerTest.java
@@ -16,9 +16,10 @@
void productionConstructorIsExplicitlyAutowired()
{
var constructors = DevelopmentCollectorStatusSynchronizer.class.getDeclaredConstructors();
- assertEquals(1, constructors.length);
- assertTrue(constructors[0].isAnnotationPresent(Autowired.class));
- assertEquals(3, constructors[0].getParameterCount());
+ assertEquals(2, constructors.length);
+ var production = java.util.Arrays.stream(constructors)
+ .filter(item -> item.isAnnotationPresent(Autowired.class)).findFirst().orElseThrow();
+ assertEquals(4, production.getParameterCount());
}
@Test
void synchronizesOnlyConfiguredCollectorsWithoutRegisteringThem() throws Exception
diff --git a/ruoyi-admin/src/test/java/com/ruoyi/config/FlywayMigrationContractTest.java b/ruoyi-admin/src/test/java/com/ruoyi/config/FlywayMigrationContractTest.java
index d3c6e01..5106b4b 100644
--- a/ruoyi-admin/src/test/java/com/ruoyi/config/FlywayMigrationContractTest.java
+++ b/ruoyi-admin/src/test/java/com/ruoyi/config/FlywayMigrationContractTest.java
@@ -128,6 +128,13 @@
assertTrue(v17.contains("AFTER ended_at_utc"));
assertFalse(v17.matches("(?s).*\\bDROP\\s+TABLE\\b.*"));
+ String v18 = resource("db/migration/V18__collector_disconnect_recovery.sql");
+ assertTrue(v18.contains("CREATE TABLE t_test_collector_exception_interval"));
+ assertTrue(v18.contains("missing_from_utc"));
+ assertTrue(v18.contains("recovered_at_utc"));
+ assertTrue(v18.contains("uk_collector_exception_open"));
+ assertFalse(v18.matches("(?s).*\\bDROP\\s+TABLE\\b.*"));
+
String testMapper = resource("mapper/system/TTestMapper.xml");
assertTrue(testMapper.contains("selectReceivingChannelsForAdapter"));
assertTrue(testMapper.contains("RECEPTION_RESUMED"));
diff --git a/ruoyi-admin/src/test/java/com/ruoyi/system/service/impl/CollectorDisconnectRecoveryServiceImplTest.java b/ruoyi-admin/src/test/java/com/ruoyi/system/service/impl/CollectorDisconnectRecoveryServiceImplTest.java
new file mode 100644
index 0000000..dd11996
--- /dev/null
+++ b/ruoyi-admin/src/test/java/com/ruoyi/system/service/impl/CollectorDisconnectRecoveryServiceImplTest.java
@@ -0,0 +1,67 @@
+package com.ruoyi.system.service.impl;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Set;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import com.ruoyi.system.domain.TTestCollectorExceptionInterval;
+import com.ruoyi.system.domain.TTestCollectorSource;
+import com.ruoyi.system.mapper.TTestMapper;
+
+/** 采集器断线对账的来源级幂等与恢复边界测试。 */
+@ExtendWith(MockitoExtension.class)
+class CollectorDisconnectRecoveryServiceImplTest
+{
+ @Mock private TTestMapper mapper;
+ @InjectMocks private CollectorDisconnectRecoveryServiceImpl service;
+
+ @Test
+ void opensOnlyOfflineSourceAndKeepsOtherSourcesIndependent()
+ {
+ TTestCollectorSource offline = source(10L, 20L, 8L, "COL-1", "S-01");
+ TTestCollectorSource online = source(10L, 20L, 9L, "COL-2", "S-01");
+ when(mapper.selectReceivingCollectorSources()).thenReturn(List.of(offline, online));
+ when(mapper.selectOpenCollectorException(10L, 20L, 8L, "S-01")).thenReturn(null);
+ when(mapper.selectOpenCollectorException(10L, 20L, 9L, "S-01")).thenReturn(null);
+
+ service.reconcile(Set.of("COL-2"), Instant.parse("2026-08-30T09:00:00Z"));
+
+ verify(mapper).insertCollectorExceptionInterval(any(TTestCollectorExceptionInterval.class));
+ verify(mapper, never()).closeCollectorExceptionInterval(any(), any(), any());
+ }
+
+ @Test
+ void closesOpenIntervalAtRecoveryBoundary()
+ {
+ TTestCollectorSource source = source(10L, 20L, 8L, "COL-1", "S-01");
+ TTestCollectorExceptionInterval open = new TTestCollectorExceptionInterval();
+ open.setIntervalId(99L);
+ open.setIntervalStatus("OPEN");
+ when(mapper.selectReceivingCollectorSources()).thenReturn(List.of(source));
+ when(mapper.selectOpenCollectorException(10L, 20L, 8L, "S-01")).thenReturn(open);
+
+ service.reconcile(Set.of("COL-1"), Instant.parse("2026-08-30T09:01:00Z"));
+
+ verify(mapper).closeCollectorExceptionInterval(eq(99L), any(), any());
+ verify(mapper, never()).insertCollectorExceptionInterval(any());
+ }
+
+ private TTestCollectorSource source(Long testId, Long relationId, Long collectorId,
+ String collectorCode, String channelCode)
+ {
+ TTestCollectorSource source = new TTestCollectorSource();
+ source.setTestId(testId); source.setBatteryRelationId(relationId); source.setCollectorId(collectorId);
+ source.setCollectorCode(collectorCode); source.setBatteryNo("BAT-01"); source.setChannelCode(channelCode);
+ return source;
+ }
+}
diff --git a/ruoyi-system/src/main/java/com/ruoyi/system/domain/TTestCollectorExceptionInterval.java b/ruoyi-system/src/main/java/com/ruoyi/system/domain/TTestCollectorExceptionInterval.java
new file mode 100644
index 0000000..798f595
--- /dev/null
+++ b/ruoyi-system/src/main/java/com/ruoyi/system/domain/TTestCollectorExceptionInterval.java
@@ -0,0 +1,54 @@
+package com.ruoyi.system.domain;
+
+import java.util.Date;
+
+/** 采集器来源断线期间的只追加事实。关闭动作只补写恢复边界。 */
+public class TTestCollectorExceptionInterval
+{
+ private Long intervalId;
+ private Long testId;
+ private Long batteryRelationId;
+ private Long collectorId;
+ private String collectorCode;
+ private String collectorName;
+ private String collectorChannelCode;
+ private String measurementName;
+ private String exceptionType;
+ private Date openedAtUtc;
+ private Date missingFromUtc;
+ private Date recoveredAtUtc;
+ private Date missingToUtc;
+ private String intervalStatus;
+ private String detail;
+
+ public Long getIntervalId() { return intervalId; }
+ public void setIntervalId(Long intervalId) { this.intervalId = intervalId; }
+ public Long getTestId() { return testId; }
+ public void setTestId(Long testId) { this.testId = testId; }
+ public Long getBatteryRelationId() { return batteryRelationId; }
+ public void setBatteryRelationId(Long batteryRelationId) { this.batteryRelationId = batteryRelationId; }
+ public Long getCollectorId() { return collectorId; }
+ public void setCollectorId(Long collectorId) { this.collectorId = collectorId; }
+ public String getCollectorCode() { return collectorCode; }
+ public void setCollectorCode(String collectorCode) { this.collectorCode = collectorCode; }
+ public String getCollectorName() { return collectorName; }
+ public void setCollectorName(String collectorName) { this.collectorName = collectorName; }
+ public String getCollectorChannelCode() { return collectorChannelCode; }
+ public void setCollectorChannelCode(String collectorChannelCode) { this.collectorChannelCode = collectorChannelCode; }
+ public String getMeasurementName() { return measurementName; }
+ public void setMeasurementName(String measurementName) { this.measurementName = measurementName; }
+ public String getExceptionType() { return exceptionType; }
+ public void setExceptionType(String exceptionType) { this.exceptionType = exceptionType; }
+ public Date getOpenedAtUtc() { return openedAtUtc; }
+ public void setOpenedAtUtc(Date openedAtUtc) { this.openedAtUtc = openedAtUtc; }
+ public Date getMissingFromUtc() { return missingFromUtc; }
+ public void setMissingFromUtc(Date missingFromUtc) { this.missingFromUtc = missingFromUtc; }
+ public Date getRecoveredAtUtc() { return recoveredAtUtc; }
+ public void setRecoveredAtUtc(Date recoveredAtUtc) { this.recoveredAtUtc = recoveredAtUtc; }
+ public Date getMissingToUtc() { return missingToUtc; }
+ public void setMissingToUtc(Date missingToUtc) { this.missingToUtc = missingToUtc; }
+ public String getIntervalStatus() { return intervalStatus; }
+ public void setIntervalStatus(String intervalStatus) { this.intervalStatus = intervalStatus; }
+ public String getDetail() { return detail; }
+ public void setDetail(String detail) { this.detail = detail; }
+}
diff --git a/ruoyi-system/src/main/java/com/ruoyi/system/domain/TTestCollectorSource.java b/ruoyi-system/src/main/java/com/ruoyi/system/domain/TTestCollectorSource.java
new file mode 100644
index 0000000..766915d
--- /dev/null
+++ b/ruoyi-system/src/main/java/com/ruoyi/system/domain/TTestCollectorSource.java
@@ -0,0 +1,25 @@
+package com.ruoyi.system.domain;
+
+/** 接收中的测试采集器来源，供断线恢复协调器使用。 */
+public class TTestCollectorSource
+{
+ private Long testId;
+ private Long batteryRelationId;
+ private String batteryNo;
+ private Long collectorId;
+ private String collectorCode;
+ private String channelCode;
+
+ public Long getTestId() { return testId; }
+ public void setTestId(Long testId) { this.testId = testId; }
+ public Long getBatteryRelationId() { return batteryRelationId; }
+ public void setBatteryRelationId(Long batteryRelationId) { this.batteryRelationId = batteryRelationId; }
+ public String getBatteryNo() { return batteryNo; }
+ public void setBatteryNo(String batteryNo) { this.batteryNo = batteryNo; }
+ public Long getCollectorId() { return collectorId; }
+ public void setCollectorId(Long collectorId) { this.collectorId = collectorId; }
+ public String getCollectorCode() { return collectorCode; }
+ public void setCollectorCode(String collectorCode) { this.collectorCode = collectorCode; }
+ public String getChannelCode() { return channelCode; }
+ public void setChannelCode(String channelCode) { this.channelCode = channelCode; }
+}
diff --git a/ruoyi-system/src/main/java/com/ruoyi/system/mapper/TTestMapper.java b/ruoyi-system/src/main/java/com/ruoyi/system/mapper/TTestMapper.java
index 76f31d2..ee5884e 100644
--- a/ruoyi-system/src/main/java/com/ruoyi/system/mapper/TTestMapper.java
+++ b/ruoyi-system/src/main/java/com/ruoyi/system/mapper/TTestMapper.java
@@ -11,6 +11,8 @@
import com.ruoyi.system.domain.TTestPlanStep;
import com.ruoyi.system.domain.TTestResult;
import com.ruoyi.system.domain.TTestWarningThreshold;
+import com.ruoyi.system.domain.TTestCollectorSource;
+import com.ruoyi.system.domain.TTestCollectorExceptionInterval;
import com.ruoyi.system.domain.TDetectionChannel;
import com.ruoyi.system.domain.vo.TestBatteryOptionVo;
import com.ruoyi.system.domain.vo.TestChannelOptionVo;
@@ -87,4 +89,15 @@
@Param("updatedAtUtc") Date updatedAtUtc);
int updateReceptionStopped(@Param("testId") Long testId, @Param("updateBy") String updateBy,
@Param("endedAtUtc") Date endedAtUtc);
+
+ /** 查询仍在接收中的采集器来源。 */
+ List<TTestCollectorSource> selectReceivingCollectorSources();
+ /** 查询来源当前开放的断线区间。 */
+ TTestCollectorExceptionInterval selectOpenCollectorException(@Param("testId") Long testId,
+ @Param("batteryRelationId") Long batteryRelationId, @Param("collectorId") Long collectorId,
+ @Param("channelCode") String channelCode);
+ int insertCollectorExceptionInterval(TTestCollectorExceptionInterval interval);
+ int closeCollectorExceptionInterval(@Param("intervalId") Long intervalId,
+ @Param("recoveredAtUtc") Date recoveredAtUtc, @Param("missingToUtc") Date missingToUtc);
+ List<TTestCollectorExceptionInterval> selectCollectorExceptionIntervals(Long testId);
}
diff --git a/ruoyi-system/src/main/java/com/ruoyi/system/service/ICollectorDisconnectRecoveryService.java b/ruoyi-system/src/main/java/com/ruoyi/system/service/ICollectorDisconnectRecoveryService.java
new file mode 100644
index 0000000..e75128d
--- /dev/null
+++ b/ruoyi-system/src/main/java/com/ruoyi/system/service/ICollectorDisconnectRecoveryService.java
@@ -0,0 +1,11 @@
+package com.ruoyi.system.service;
+
+import java.time.Instant;
+import java.util.Set;
+
+/** 维护接收中采集器来源的断线异常与恢复边界。 */
+public interface ICollectorDisconnectRecoveryService
+{
+ /** 根据当前模拟器在线设备编码对账所有接收中来源。 */
+ void reconcile(Set<String> onlineCollectorCodes, Instant observedAt);
+}
diff --git a/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/CollectorDisconnectRecoveryServiceImpl.java b/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/CollectorDisconnectRecoveryServiceImpl.java
new file mode 100644
index 0000000..d7b264e
--- /dev/null
+++ b/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/CollectorDisconnectRecoveryServiceImpl.java
@@ -0,0 +1,62 @@
+package com.ruoyi.system.service.impl;
+
+import java.time.Instant;
+import java.util.Collections;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+import com.alibaba.fastjson2.JSON;
+import com.ruoyi.system.domain.TTestCollectorExceptionInterval;
+import com.ruoyi.system.domain.TTestCollectorSource;
+import com.ruoyi.system.mapper.TTestMapper;
+import com.ruoyi.system.service.ICollectorDisconnectRecoveryService;
+
+/** 将采集器在线事实对账为可恢复的来源级异常区间。 */
+@Service
+public class CollectorDisconnectRecoveryServiceImpl implements ICollectorDisconnectRecoveryService
+{
+ private final TTestMapper mapper;
+
+ public CollectorDisconnectRecoveryServiceImpl(TTestMapper mapper) { this.mapper = mapper; }
+
+ @Override
+ @Transactional(rollbackFor = Exception.class)
+ public void reconcile(Set<String> onlineCollectorCodes, Instant observedAt)
+ {
+ Set<String> online = onlineCollectorCodes == null ? Collections.emptySet()
+ : onlineCollectorCodes.stream().filter(code -> code != null && !code.isBlank()).collect(Collectors.toSet());
+ for (TTestCollectorSource source : mapper.selectReceivingCollectorSources())
+ {
+ TTestCollectorExceptionInterval open = mapper.selectOpenCollectorException(source.getTestId(),
+ source.getBatteryRelationId(), source.getCollectorId(), source.getChannelCode());
+ boolean isOnline = online.contains(source.getCollectorCode());
+ if (!isOnline && open == null)
+ {
+ TTestCollectorExceptionInterval interval = new TTestCollectorExceptionInterval();
+ interval.setTestId(source.getTestId());
+ interval.setBatteryRelationId(source.getBatteryRelationId());
+ interval.setCollectorId(source.getCollectorId());
+ interval.setCollectorChannelCode(source.getChannelCode());
+ interval.setExceptionType("COLLECTOR_DISCONNECTED");
+ java.util.Date at = java.util.Date.from(observedAt);
+ interval.setOpenedAtUtc(at);
+ interval.setMissingFromUtc(at);
+ interval.setDetail(JSON.toJSONString(java.util.Map.of("batteryNo", source.getBatteryNo(),
+ "collectorCode", source.getCollectorCode(), "message", "采集器离线，检测柜及其他来源继续接收")));
+ mapper.insertCollectorExceptionInterval(interval);
+ mapper.insertTestAudit(source.getTestId(), "COLLECTOR_DISCONNECTED", null, "system", null,
+ JSON.toJSONString(java.util.Map.of("collectorId", source.getCollectorId(),
+ "channelCode", source.getChannelCode(), "missingFrom", observedAt.toString())));
+ }
+ else if (isOnline && open != null)
+ {
+ java.util.Date at = java.util.Date.from(observedAt);
+ mapper.closeCollectorExceptionInterval(open.getIntervalId(), at, at);
+ mapper.insertTestAudit(source.getTestId(), "COLLECTOR_RECONNECTED", null, "system", null,
+ JSON.toJSONString(java.util.Map.of("collectorId", source.getCollectorId(),
+ "channelCode", source.getChannelCode(), "recoveredAt", observedAt.toString())));
+ }
+ }
+ }
+}
diff --git a/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TTestLiveServiceImpl.java b/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TTestLiveServiceImpl.java
index 02f04cf..1765131 100644
--- a/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TTestLiveServiceImpl.java
+++ b/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TTestLiveServiceImpl.java
@@ -24,6 +24,8 @@
import com.ruoyi.device.model.Measurement;
import com.ruoyi.system.domain.TTest;
import com.ruoyi.system.domain.TTestBattery;
+import com.ruoyi.system.domain.TTestCollectorExceptionInterval;
+import com.ruoyi.system.mapper.TTestMapper;
import com.ruoyi.system.service.ITestLiveService;
import com.ruoyi.system.service.ITTestService;
@@ -34,12 +36,13 @@
private final ITTestService tests;
private final DeviceDiscoveryCandidateStore candidates;
private final BtsDevelopmentSimulator simulator;
+ private final TTestMapper testMapper;
private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
private final ConcurrentMap<String, LiveSource> lastSources = new ConcurrentHashMap<>();
public TTestLiveServiceImpl(ITTestService tests, DeviceDiscoveryCandidateStore candidates,
- BtsDevelopmentSimulator simulator)
- { this.tests = tests; this.candidates = candidates; this.simulator = simulator; }
+ BtsDevelopmentSimulator simulator, TTestMapper testMapper)
+ { this.tests = tests; this.candidates = candidates; this.simulator = simulator; this.testMapper = testMapper; }
@Override public LiveState current(Long testId)
{
@@ -55,12 +58,34 @@
if (battery.getCollectorId() != null)
sources.add(snapshotFor(testId, collectorSource(battery), freezeSnapshots));
}
- List<Map<String, Object>> exceptions = sources.stream().filter(source -> !"ONLINE".equals(source.deviceStatus())
+ List<TTestCollectorExceptionInterval> persisted = testMapper.selectCollectorExceptionIntervals(testId);
+ if (persisted == null) persisted = List.of();
+ List<Map<String, Object>> exceptions = new ArrayList<>(persisted.stream().map(interval -> {
+ Map<String, Object> item = new LinkedHashMap<>();
+ item.put("sourceType", "COLLECTOR");
+ item.put("sourceId", interval.getCollectorId());
+ item.put("collectorCode", interval.getCollectorCode());
+ item.put("collectorName", interval.getCollectorName());
+ item.put("channelCode", interval.getCollectorChannelCode());
+ item.put("measurementName", interval.getMeasurementName());
+ item.put("exceptionType", interval.getExceptionType());
+ item.put("status", interval.getIntervalStatus());
+ item.put("openedAt", interval.getOpenedAtUtc());
+ item.put("missingFrom", interval.getMissingFromUtc());
+ item.put("recoveredAt", interval.getRecoveredAtUtc());
+ item.put("missingTo", interval.getMissingToUtc());
+ item.put("detail", interval.getDetail());
+ return item;
+ }).toList());
+ exceptions.addAll(sources.stream().filter(source -> !"COLLECTOR".equals(source.sourceType())
+ && !"ONLINE".equals(source.deviceStatus())
&& !"WORKING".equals(source.deviceStatus()) && !"PAUSED".equals(source.deviceStatus()))
.map(source -> Map.<String, Object>of("sourceType", source.sourceType(), "sourceId", source.sourceId(),
"batteryNo", source.batteryNo(), "status", source.deviceStatus(), "message", "数据来源当前不可观测"))
- .toList();
- return new LiveState(testId, test.getTestStatus(), test.getReceptionStatus(), Instant.now(), sources, exceptions);
+ .toList());
+ boolean abnormal = persisted.stream().anyMatch(item -> "OPEN".equals(item.getIntervalStatus()));
+ String testStatus = abnormal && "RUNNING".equals(test.getTestStatus()) ? "RUNNING_ABNORMAL" : test.getTestStatus();
+ return new LiveState(testId, testStatus, test.getReceptionStatus(), Instant.now(), sources, exceptions);
}
private LiveSource snapshotFor(Long testId, LiveSource fresh, boolean freeze)
diff --git a/ruoyi-system/src/main/resources/mapper/system/TTestMapper.xml b/ruoyi-system/src/main/resources/mapper/system/TTestMapper.xml
index 657dab0..4bd94cf 100644
--- a/ruoyi-system/src/main/resources/mapper/system/TTestMapper.xml
+++ b/ruoyi-system/src/main/resources/mapper/system/TTestMapper.xml
@@ -108,6 +108,23 @@
<result property="exceptionInfo" column="exception_info" />
<result property="delFlag" column="del_flag" />
</resultMap>
+ <resultMap id="TTestCollectorExceptionIntervalResultMap" type="TTestCollectorExceptionInterval">
+ <id property="intervalId" column="interval_id" />
+ <result property="testId" column="test_id" />
+ <result property="batteryRelationId" column="battery_relation_id" />
+ <result property="collectorId" column="collector_id" />
+ <result property="collectorCode" column="collector_code" />
+ <result property="collectorName" column="collector_name" />
+ <result property="collectorChannelCode" column="collector_channel_code" />
+ <result property="measurementName" column="measurement_name" />
+ <result property="exceptionType" column="exception_type" />
+ <result property="openedAtUtc" column="opened_at_utc" />
+ <result property="missingFromUtc" column="missing_from_utc" />
+ <result property="recoveredAtUtc" column="recovered_at_utc" />
+ <result property="missingToUtc" column="missing_to_utc" />
+ <result property="intervalStatus" column="interval_status" />
+ <result property="detail" column="detail" />
+ </resultMap>
<sql id="selectTestBase">
select t.test_id, t.test_name, t.creator_id, creator.nick_name creator_name,
@@ -458,4 +475,48 @@
update_by = #{updateBy}, update_time = #{endedAtUtc}
where test_id = #{testId} and reception_status in ('RECEIVING', 'PAUSED') and del_flag = 0
</update>
+
+ <select id="selectReceivingCollectorSources" resultType="com.ruoyi.system.domain.TTestCollectorSource">
+ select test_fact.test_id testId, battery.relation_id batteryRelationId, battery.battery_no batteryNo,
+ collector.collector_id collectorId, collector.collector_code collectorCode,
+ mapping.channel_code channelCode
+ from t_test test_fact
+ join t_test_battery battery on battery.test_id = test_fact.test_id and battery.del_flag = 0
+ join t_collector collector on collector.collector_id = battery.collector_id
+ and collector.del_flag = 0
+ join t_test_collector_channel mapping on mapping.battery_relation_id = battery.relation_id
+ and mapping.test_id = test_fact.test_id
+ where test_fact.reception_status = 'RECEIVING' and test_fact.del_flag = 0
+ order by test_fact.test_id, battery.relation_id, mapping.mapping_id
+ </select>
+ <select id="selectOpenCollectorException" resultMap="TTestCollectorExceptionIntervalResultMap">
+ select * from t_test_collector_exception_interval
+ where test_id = #{testId} and battery_relation_id = #{batteryRelationId}
+ and collector_id = #{collectorId} and collector_channel_code = #{channelCode}
+ and exception_type = 'COLLECTOR_DISCONNECTED' and interval_status = 'OPEN'
+ order by interval_id desc limit 1
+ </select>
+ <insert id="insertCollectorExceptionInterval" parameterType="TTestCollectorExceptionInterval" useGeneratedKeys="true" keyProperty="intervalId">
+ insert into t_test_collector_exception_interval
+ (test_id, battery_relation_id, collector_id, collector_channel_code, exception_type,
+ opened_at_utc, missing_from_utc, detail)
+ values (#{testId}, #{batteryRelationId}, #{collectorId}, #{collectorChannelCode},
+ #{exceptionType}, #{openedAtUtc}, #{missingFromUtc}, #{detail})
+ </insert>
+ <update id="closeCollectorExceptionInterval">
+ update t_test_collector_exception_interval
+ set recovered_at_utc = #{recoveredAtUtc}, missing_to_utc = #{missingToUtc}, interval_status = 'CLOSED'
+ where interval_id = #{intervalId} and interval_status = 'OPEN'
+ </update>
+ <select id="selectCollectorExceptionIntervals" parameterType="Long" resultMap="TTestCollectorExceptionIntervalResultMap">
+ select exception_fact.*, collector.collector_code, collector.collector_name,
+ mapping.measurement_name
+ from t_test_collector_exception_interval exception_fact
+ left join t_collector collector on collector.collector_id = exception_fact.collector_id
+ left join t_test_collector_channel mapping on mapping.battery_relation_id = exception_fact.battery_relation_id
+ and mapping.collector_id = exception_fact.collector_id
+ and mapping.channel_code = exception_fact.collector_channel_code
+ where exception_fact.test_id = #{testId}
+ order by exception_fact.opened_at_utc, exception_fact.interval_id
+ </select>
</mapper>
--
Gitblit v1.7.1