完成测试生命周期与接收边界

yeg

2026-09-05 acfb83068486766f63284b479d30f78e3ec17dca

 ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TestDataReceptionServiceImpl.java

@@ -61,9 +61,11 @@
        for (Long testId : testIds)
        {
            TTest locked = mapper.selectTestByIdForUpdate(testId);
            if (locked == null || !"RECEIVING".equals(locked.getReceptionStatus())) continue;
            if (locked == null || !"RUNNING".equals(locked.getTestStatus())
                    || !"RECEIVING".equals(locked.getReceptionStatus())) continue;
            TTest test = tests.selectTestById(testId);
            if (test == null || !"RECEIVING".equals(test.getReceptionStatus())) continue;
            if (test == null || !"RUNNING".equals(test.getTestStatus())
                    || !"RECEIVING".equals(test.getReceptionStatus())) continue;
            // 完整快照与原始事实在同一接收事务内写入；原始事实按发现频率限流，避免 100ms 开发调度重复扫描历史。
            Instant lastRaw = lastRawCaptureAt.get(testId);
            if (lastRaw == null || !now.toInstant().isBefore(lastRaw.plusSeconds(5)))
@@ -95,7 +97,8 @@
        try { collector = simulator.collectorRealtime(battery.getCollectorId().intValue()); } catch (IllegalArgumentException ignored) { return; }
        if (cabinet == null || !collector.online()) return;
        if (test.getStartedAtUtc() != null && cabinet.deviceTime().isBefore(test.getStartedAtUtc().toInstant())) return;
        if (test.getUpdateTime() != null && cabinet.deviceTime().isBefore(test.getUpdateTime().toInstant())) return;
        if (test.getReceptionBoundaryAtUtc() != null && cabinet.serverReceivedAt() != null
                && cabinet.serverReceivedAt().isBefore(test.getReceptionBoundaryAtUtc().toInstant())) return;
        Map<String, CollectorRealtimeMeasurement> metrics = new LinkedHashMap<>();
        for (CollectorRealtimeMeasurement item : collector.measurements())
            if (battery.getCollectorChannels() != null && battery.getCollectorChannels().stream().anyMatch(mapping ->
@@ -109,9 +112,10 @@
        if (test.getStartedAtUtc() != null && (temp.deviceTime().isBefore(test.getStartedAtUtc().toInstant())
                || pos.deviceTime().isBefore(test.getStartedAtUtc().toInstant())
                || neg.deviceTime().isBefore(test.getStartedAtUtc().toInstant()))) return;
        if (test.getUpdateTime() != null && (temp.serverReceivedAt().isBefore(test.getUpdateTime().toInstant())
                || pos.serverReceivedAt().isBefore(test.getUpdateTime().toInstant())
                || neg.serverReceivedAt().isBefore(test.getUpdateTime().toInstant()))) return;
        if (test.getReceptionBoundaryAtUtc() != null
                && (temp.serverReceivedAt().isBefore(test.getReceptionBoundaryAtUtc().toInstant())
                || pos.serverReceivedAt().isBefore(test.getReceptionBoundaryAtUtc().toInstant())
                || neg.serverReceivedAt().isBefore(test.getReceptionBoundaryAtUtc().toInstant()))) return;
        Date slot = new Date((now.getTime() / interval) * interval);
        byte[] hash = snapshotHash(cabinet, temp, pos, neg, slot.toInstant());
        mapper.insertRealtimeSnapshot(test.getTestId(), binding.getBindingId(), binding.getChannelId(), battery.getCollectorId(), slot,
@@ -143,9 +147,11 @@
        {
            // 锁住测试行直到本轮采样写入完成，暂停/停止动作只能在本轮接收提交后生效。
            TTest locked = mapper.selectTestByIdForUpdate(testId);
            if (locked == null || !"RECEIVING".equals(locked.getReceptionStatus())) continue;
            if (locked == null || !"RUNNING".equals(locked.getTestStatus())
                    || !"RECEIVING".equals(locked.getReceptionStatus())) continue;
            TTest test = tests.selectTestById(testId);
            if (test == null || !"RECEIVING".equals(test.getReceptionStatus())) continue;
            if (test == null || !"RUNNING".equals(test.getTestStatus())
                    || !"RECEIVING".equals(test.getReceptionStatus())) continue;
            captureTest(test);
        }
    }
@@ -160,9 +166,11 @@
        for (Long testId : testIds)
        {
            TTest locked = mapper.selectTestByIdForUpdate(testId);
            if (locked == null || !"RECEIVING".equals(locked.getReceptionStatus())) continue;
            if (locked == null || !"RUNNING".equals(locked.getTestStatus())
                    || !"RECEIVING".equals(locked.getReceptionStatus())) continue;
            TTest test = tests.selectTestById(testId);
            if (test == null || !"RECEIVING".equals(test.getReceptionStatus())) continue;
            if (test == null || !"RUNNING".equals(test.getTestStatus())
                    || !"RECEIVING".equals(test.getReceptionStatus())) continue;
            for (TTestBattery battery : test.getBatteries() == null ? List.<TTestBattery>of() : test.getBatteries())
                captureCollectorMeasurements(test, battery);
        }
@@ -182,6 +190,8 @@
            {
                // 模拟器保留通道历史；只接收本次测试开始后产生的设备时间记录，避免顺序测试串入旧事实。
                if (startedAt != null && record.deviceTime().isBefore(startedAt.toInstant())) continue;
                if (test.getReceptionBoundaryAtUtc() != null && record.serverReceivedAt() != null
                        && record.serverReceivedAt().isBefore(test.getReceptionBoundaryAtUtc().toInstant())) continue;
                if (save(test, binding, record)) {
                    String source = String.join(":", java.util.Objects.toString(binding.getAdapterCode(), ""),
                            java.util.Objects.toString(binding.getDevtype(), ""), java.util.Objects.toString(binding.getDevid(), ""),
@@ -207,8 +217,9 @@
        for (CollectorRealtimeMeasurement measurement : realtime.measurements())
        {
            if (startedAt != null && measurement.deviceTime().isBefore(startedAt.toInstant())) continue;
            // 继续接收只接受恢复动作之后查询到的快照；暂停期间采集器自行产生的历史不回补。
            if (test.getUpdateTime() != null && measurement.serverReceivedAt().isBefore(test.getUpdateTime().toInstant())) continue;
            // 继续接收只接受恢复动作之后服务器收到的采样；暂停期间采集器历史不回补。
            if (test.getReceptionBoundaryAtUtc() != null
                    && measurement.serverReceivedAt().isBefore(test.getReceptionBoundaryAtUtc().toInstant())) continue;
            boolean mapped = battery.getCollectorChannels().stream().anyMatch(mapping ->
                    java.util.Objects.equals(mapping.getChannelCode(), measurement.channelCode())
                    && metricMatches(mapping.getMeasurementCode(), measurement.metricCode()));