| | |
| | | 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.ITestLiveService.TestPlanRow; |
| | | import com.ruoyi.system.service.ITestLiveService.TestPlanSnapshot; |
| | |
| | | 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; } |
| | | BtsDevelopmentSimulator simulator) |
| | | { this.tests = tests; this.candidates = candidates; this.simulator = simulator; } |
| | | |
| | | @Override public LiveState current(Long testId) |
| | | { |
| | |
| | | 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()); |
| | | List<Map<String, Object>> exceptions = new ArrayList<>(); |
| | | 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(); |
| | | List<com.ruoyi.system.domain.TTestIntegrityIssue> integrity = new ArrayList<>(); |
| | | List<com.ruoyi.system.domain.TTestIntegrityIssue> persistedIntegrity = testMapper.selectIntegrityIssues(testId); |
| | | if (persistedIntegrity != null) integrity.addAll(persistedIntegrity); |
| | | List<com.ruoyi.system.domain.TTestIntegrityIssue> persistedExceptions = testMapper.selectTestExceptionFacts(testId); |
| | | if (persistedExceptions != null) integrity.addAll(persistedExceptions); |
| | | String testStatus = test.getTestStatus(); |
| | | TestPlanSnapshot testPlan = detectorTestPlan(test); |
| | | return new LiveState(testId, testStatus, test.getReceptionStatus(), Instant.now(), sources, exceptions, |
| | | integrity == null ? List.of() : integrity, testPlan); |
| | | return new LiveState(testId, testStatus, test.getReceptionStatus(), Instant.now(), sources, exceptions, testPlan); |
| | | } |
| | | |
| | | private TestPlanSnapshot detectorTestPlan(TTest test) |
| | |
| | | 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("integrity").data(state.integrityIssues())); |
| | | emitter.send(SseEmitter.event().name("heartbeat").data(Map.of("at", state.updatedAt()))); |
| | | } |
| | | |