清理并隔离离线实时快照

yeg

2026-09-02 c86e87f774e0b6cebbf73017d06bfbe739959d54

清理并隔离离线实时快照

移除无效快照缓存，离线来源不再返回旧采集值。

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

@@ -11,8 +11,6 @@
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import jakarta.annotation.PreDestroy;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
@@ -37,7 +35,6 @@
    private final DeviceDiscoveryCandidateStore candidates;
    private final BtsDevelopmentSimulator simulator;
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
    private final ConcurrentMap<String, LiveSource> lastSources = new ConcurrentHashMap<>();

    public TTestLiveServiceImpl(ITTestService tests, DeviceDiscoveryCandidateStore candidates,
            BtsDevelopmentSimulator simulator)
@@ -49,14 +46,12 @@
        List<CabinetCandidate> cabinets = candidates.snapshot(Instant.now()).cabinets();
        List<LiveSource> sources = new ArrayList<>();
        boolean receiving = "RECEIVING".equals(test.getReceptionStatus());
        boolean freezeSnapshots = "PAUSED".equals(test.getReceptionStatus())
                || "ENDED".equals(test.getReceptionStatus());
        for (TTestBattery battery : test.getBatteries() == null ? List.<TTestBattery>of() : test.getBatteries())
        {
            if (battery.getBinding() != null)
                sources.add(snapshotFor(testId, cabinetSource(battery, cabinets), receiving, freezeSnapshots));
                sources.add(snapshotFor(cabinetSource(battery, cabinets), receiving));
            if (battery.getCollectorId() != null)
                sources.add(snapshotFor(testId, collectorSource(battery), receiving, freezeSnapshots));
                sources.add(snapshotFor(collectorSource(battery), receiving));
        }
        List<Map<String, Object>> exceptions = new ArrayList<>();
        exceptions.addAll(sources.stream().filter(source -> !"COLLECTOR".equals(source.sourceType())
@@ -94,13 +89,11 @@
        return null;
    }

    private LiveSource snapshotFor(Long testId, LiveSource fresh, boolean receiving, boolean freeze)
    private LiveSource snapshotFor(LiveSource fresh, boolean receiving)
    {
        String key = testId + ":" + fresh.sourceType() + ":" + fresh.sourceId() + ":" + fresh.batteryNo();
        // 暂停或停止只停止数据请求；设备状态仍以本次查询的最新事实为准，采集快照不继续展示。
        if (!receiving)
            return withoutSnapshot(fresh);
        lastSources.put(key, fresh);
        return fresh;
    }

@@ -161,7 +154,8 @@
                    && Objects.equals(channel.identity().subdevid(), binding.getSubdevid())
                    && Objects.equals(channel.identity().chlid(), binding.getChlid()))
                return new LiveSource("CABINET_CHANNEL", binding.getChannelId(), battery.getBatteryNo(),
                        channel.online() ? channel.workStatus().name() : "OFFLINE", snapshot(channel));
                        channel.online() ? channel.workStatus().name() : "OFFLINE",
                        channel.online() ? snapshot(channel) : Map.of());
        return new LiveSource("CABINET_CHANNEL", binding.getChannelId(), battery.getBatteryNo(), "UNKNOWN", Map.of());
    }

@@ -206,7 +200,7 @@
            collector.latestMeasurements().stream().map(item -> item.serverReceivedAt()).max(Instant::compareTo)
                    .ifPresent(observedAt -> value.put("observedAt", observedAt));
            return new LiveSource("COLLECTOR", (long) collector.collectorId(), battery.getBatteryNo(),
                    collector.online() ? "ONLINE" : "OFFLINE", value);
                    collector.online() ? "ONLINE" : "OFFLINE", collector.online() ? value : Map.of());
        }
        return new LiveSource("COLLECTOR", battery.getCollectorId(), battery.getBatteryNo(), "UNKNOWN", Map.of());
    }