package com.ruoyi.system.service.impl;
|
|
import java.io.IOException;
|
import java.time.Instant;
|
import java.util.ArrayList;
|
import java.util.LinkedHashMap;
|
import java.util.List;
|
import java.util.Map;
|
import java.util.Objects;
|
import java.util.concurrent.Executors;
|
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;
|
import com.ruoyi.device.bts.BtsDevelopmentSimulator;
|
import com.ruoyi.device.bts.BtsDevelopmentSimulator.CollectorSnapshot;
|
import com.ruoyi.device.discovery.DeviceDiscoveryCandidateStore;
|
import com.ruoyi.device.discovery.DeviceDiscoveryCandidateStore.CabinetCandidate;
|
import com.ruoyi.device.discovery.DeviceDiscoveryCandidateStore.ChannelCandidate;
|
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;
|
|
/** 将设备最新快照转换为测试视角，并托管 SSE 连接生命周期。 */
|
@Service
|
public class TTestLiveServiceImpl implements ITestLiveService
|
{
|
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, TTestMapper testMapper)
|
{ this.tests = tests; this.candidates = candidates; this.simulator = simulator; this.testMapper = testMapper; }
|
|
@Override public LiveState current(Long testId)
|
{
|
TTest test = tests.selectTestById(testId);
|
List<CabinetCandidate> cabinets = candidates.snapshot(Instant.now()).cabinets();
|
List<LiveSource> sources = new ArrayList<>();
|
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), freezeSnapshots));
|
if (battery.getCollectorId() != null)
|
sources.add(snapshotFor(testId, collectorSource(battery), freezeSnapshots));
|
}
|
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());
|
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)
|
{
|
String key = testId + ":" + fresh.sourceType() + ":" + fresh.sourceId() + ":" + fresh.batteryNo();
|
if (freeze) return lastSources.getOrDefault(key, fresh);
|
lastSources.put(key, fresh);
|
return fresh;
|
}
|
|
@Override public SseEmitter stream(Long testId)
|
{
|
SseEmitter emitter = createEmitter();
|
try { sendAll(emitter, testId); }
|
catch (IOException exception)
|
{
|
// 客户端刷新或离开页面会断开连接，不能再交给全局异常处理器写 SSE JSON。
|
emitter.complete();
|
return emitter;
|
}
|
catch (RuntimeException exception) { emitter.completeWithError(exception); return emitter; }
|
ScheduledFuture<?> future = scheduler.scheduleAtFixedRate(() -> {
|
try { sendAll(emitter, testId); }
|
catch (IOException exception)
|
{
|
// 写出失败表示 SSE 客户端已断开，静默结束连接。
|
emitter.complete();
|
}
|
catch (RuntimeException exception) { emitter.completeWithError(exception); }
|
}, 3, 3, TimeUnit.SECONDS);
|
emitter.onCompletion(() -> future.cancel(false));
|
emitter.onTimeout(() -> future.cancel(false));
|
emitter.onError(error -> future.cancel(false));
|
return emitter;
|
}
|
|
/** 创建实时流发射器，供断连生命周期测试替换。 */
|
protected SseEmitter createEmitter()
|
{
|
return new SseEmitter(0L);
|
}
|
|
private void sendAll(SseEmitter emitter, Long testId) throws IOException
|
{
|
LiveState state = current(testId);
|
emitter.send(SseEmitter.event().name("snapshot").data(state));
|
emitter.send(SseEmitter.event().name("device-status").data(state.sources()));
|
emitter.send(SseEmitter.event().name("reception-status").data(Map.of("testId", testId, "receptionStatus", state.receptionStatus())));
|
emitter.send(SseEmitter.event().name("test-status").data(Map.of("testId", testId, "testStatus", state.testStatus())));
|
emitter.send(SseEmitter.event().name("exception").data(state.exceptions()));
|
emitter.send(SseEmitter.event().name("heartbeat").data(Map.of("at", state.updatedAt())));
|
}
|
|
private LiveSource cabinetSource(TTestBattery battery, List<CabinetCandidate> cabinets)
|
{
|
var binding = battery.getBinding();
|
for (CabinetCandidate cabinet : cabinets) for (ChannelCandidate channel : cabinet.channels())
|
if (Objects.equals(channel.identity().devtype(), binding.getDevtype())
|
&& Objects.equals(channel.identity().devid(), binding.getDevid())
|
&& 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));
|
return new LiveSource("CABINET_CHANNEL", binding.getChannelId(), battery.getBatteryNo(), "UNKNOWN", Map.of());
|
}
|
|
private Map<String, Object> snapshot(ChannelCandidate channel)
|
{
|
Map<String, Object> value = new LinkedHashMap<>();
|
if (channel.latestSnapshot() == null) return value;
|
value.put("source", "检测柜");
|
value.put("observedAt", channel.latestSnapshot().receivedAt());
|
value.put("measurements", channel.latestSnapshot().measurements().entrySet().stream().map(entry -> {
|
Measurement measurement = entry.getValue();
|
return Map.of("metricCode", entry.getKey(), "value", measurement.value(), "unit", measurement.unit());
|
}).toList());
|
value.put("units", channel.latestSnapshot().measurements().values().stream().map(Measurement::unit).distinct().toList());
|
return value;
|
}
|
|
private LiveSource collectorSource(TTestBattery battery)
|
{
|
for (CollectorSnapshot collector : simulator.snapshot().collectors()) if (collector.collectorId() == battery.getCollectorId())
|
{
|
Map<String, Object> value = new LinkedHashMap<>();
|
value.put("source", "采集器"); value.put("sampleIntervalMs", collector.sampleIntervalMs());
|
value.put("measurements", collector.latestMeasurements());
|
value.put("units", collector.latestMeasurements().stream().map(item -> item.unit()).distinct().toList());
|
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);
|
}
|
return new LiveSource("COLLECTOR", battery.getCollectorId(), battery.getBatteryNo(), "UNKNOWN", Map.of());
|
}
|
|
@PreDestroy void shutdown() { scheduler.shutdownNow(); }
|
}
|
