检测室
blame | 历史 | 补丁 | 提交 | 提交对比 | show whitespace
yeg
2026-09-05 92f32acd847ea1700590bc669750f23ef499840e 
 battery-business/src/main/java/com/battery/business/service/impl/TTestLiveServiceImpl.java


File was renamed from ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TTestLiveServiceImpl.java


@@ -1,34 +1,27 @@


package com.ruoyi.system.service.impl;


package com.battery.business.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 jakarta.annotation.PreDestroy;


import org.springframework.stereotype.Service;


import org.springframework.beans.factory.annotation.Autowired;


import org.springframework.lang.Nullable;


import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;


import com.battery.device.simulator.BtsDevelopmentSimulator;


import com.battery.device.simulator.BtsDevelopmentSimulator.CollectorSnapshot;


import com.battery.device.discovery.DeviceDiscoveryCandidateStore;


import com.battery.device.discovery.DeviceDiscoveryCandidateStore.CabinetCandidate;


import com.battery.device.discovery.DeviceDiscoveryCandidateStore.ChannelCandidate;


import com.battery.device.model.Measurement;


import com.ruoyi.system.domain.TTest;


import com.ruoyi.system.domain.TTestBattery;


import com.ruoyi.system.service.ITestLiveService;


import com.ruoyi.system.service.ITestLiveService.TestPlanRow;


import com.ruoyi.system.service.ITestLiveService.TestPlanSnapshot;


import com.ruoyi.system.service.ITTestService;


import com.ruoyi.system.mapper.TTestMapper;


import com.battery.device.api.TestDataObservationAdapter;


import com.battery.device.api.model.ChannelIdentity;


import com.battery.device.api.discovery.DeviceDiscoveryCandidateStore;


import com.battery.device.api.discovery.DeviceDiscoveryCandidateStore.CabinetCandidate;


import com.battery.device.api.discovery.DeviceDiscoveryCandidateStore.ChannelCandidate;


import com.battery.device.api.model.Measurement;


import com.battery.business.domain.TTest;


import com.battery.business.domain.TTestBattery;


import com.battery.business.service.ITestLiveService;


import com.battery.business.service.ITestLiveService.TestPlanRow;


import com.battery.business.service.ITestLiveService.TestPlanSnapshot;


import com.battery.business.service.ITTestService;


import com.battery.business.mapper.TTestMapper;




/** 将设备最新快照转换为测试视角，并托管 SSE 连接生命周期。 */


@Service


@@ -36,18 +29,17 @@


{


    private final ITTestService tests;


    private final DeviceDiscoveryCandidateStore candidates;


    private final BtsDevelopmentSimulator simulator;


    private final TestDataObservationAdapter adapter;


    private final TTestMapper mapper;


    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);




    @Autowired


    public TTestLiveServiceImpl(ITTestService tests, DeviceDiscoveryCandidateStore candidates,


            @Nullable BtsDevelopmentSimulator simulator, TTestMapper mapper)


    { this.tests = tests; this.candidates = candidates; this.simulator = simulator; this.mapper = mapper; }


            @Nullable TestDataObservationAdapter adapter, TTestMapper mapper)


    { this.tests = tests; this.candidates = candidates; this.adapter = adapter; this.mapper = mapper; }




    public TTestLiveServiceImpl(ITTestService tests, DeviceDiscoveryCandidateStore candidates,


            BtsDevelopmentSimulator simulator)


    { this(tests, candidates, simulator, null); }


            TestDataObservationAdapter adapter)


    { this(tests, candidates, adapter, null); }




    @Override public LiveState current(Long testId)


    {


@@ -83,7 +75,7 @@




    private TestPlanSnapshot detectorTestPlan(TTest test)


    {


        if (simulator == null) return null;


        if (adapter == null) return null;


        if (test.getBatteries() == null) return null;


        for (TTestBattery battery : test.getBatteries())


        {


@@ -91,9 +83,10 @@


                    || battery.getBinding().getChlid() == null) continue;


            try


            {


                var source = "NOT_STARTED".equals(test.getReceptionStatus())


                        ? simulator.btsTest(battery.getBinding().getDevid(), battery.getBinding().getChlid(), 1)


                        : simulator.btsTest(battery.getBinding().getDevid(), battery.getBinding().getChlid());


                var binding = battery.getBinding();


                var source = adapter.testPlan(new ChannelIdentity(binding.getAdapterCode(), binding.getDevtype(),


                        binding.getDevid(), binding.getSubdevid(), binding.getChlid()));


                if (source == null) continue;


                List<TestPlanRow> rows = source.rows().stream()


                        .map(row -> new TestPlanRow(row.rowType(), row.stepNo(), row.cyclePosition(), row.stepName(), row.durationSeconds(),


                                row.voltageV(), row.executionStatus(), row.cycleCount(), row.recordCount()))


@@ -101,7 +94,7 @@


                return new TestPlanSnapshot(source.currentCycleNo(), source.currentStepNo(),


                        source.plannedCycleCount(), rows);


            }


            catch (IllegalArgumentException ignored) { /* 登记设备尚未被开发模拟器提供时保持无方案事实。 */ }


            catch (IllegalArgumentException ignored) { /* 来源未提供方案时保持为空。 */ }


        }


        return null;


    }


@@ -119,54 +112,12 @@


        return new LiveSource(source.sourceType(), source.sourceId(), source.batteryNo(), source.deviceStatus(), Map.of());


    }




    @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())


            if (Objects.equals(channel.identity().adapterCode(), binding.getAdapterCode())


                    && 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()))


@@ -208,14 +159,16 @@




    private LiveSource collectorSource(TTestBattery battery)


    {


        if (simulator == null) return new LiveSource("COLLECTOR", battery.getCollectorId(), battery.getBatteryNo(),


        if (adapter == null || mapper == null) return new LiveSource("COLLECTOR", battery.getCollectorId(), battery.getBatteryNo(),


                "OFFLINE", Map.of());


        for (CollectorSnapshot collector : simulator.snapshot().collectors()) if (collector.collectorId() == battery.getCollectorId())


        var collector = adapter.latestCollector(mapper.selectCollectorCode(battery.getCollectorId()));


        if (collector != null)


        {


            Map<String, Object> value = new LinkedHashMap<>();


            value.put("source", "采集器"); value.put("sampleIntervalMs", collector.sampleIntervalMs());


            // 在业务响应边界转换毫秒时间，设备时间与服务器接收时间分别保留。


            value.put("measurements", collector.latestMeasurements().stream().map(item -> {


            value.put("measurements", collector.measurements().stream().filter(item -> battery.getCollectorChannels() != null


                    && battery.getCollectorChannels().stream().anyMatch(binding -> binding.getChannelCode().equals(item.channelCode()))).map(item -> {


                Map<String, Object> measurement = new LinkedHashMap<>();


                measurement.put("metricCode", item.metricCode());


                measurement.put("channelCode", item.channelCode());


@@ -225,14 +178,13 @@


                measurement.put("serverReceivedAt", item.serverReceivedAt().toEpochMilli());


                return measurement;


            }).toList());


            value.put("units", collector.latestMeasurements().stream().map(item -> item.unit()).distinct().toList());


            collector.latestMeasurements().stream().map(item -> item.serverReceivedAt()).max(Instant::compareTo)


            value.put("units", collector.measurements().stream().map(item -> item.unit()).distinct().toList());


            collector.measurements().stream().map(item -> item.serverReceivedAt()).max(Instant::compareTo)


                    .ifPresent(observedAt -> value.put("observedAt", observedAt.toEpochMilli()));


            return new LiveSource("COLLECTOR", (long) collector.collectorId(), battery.getBatteryNo(),


            return new LiveSource("COLLECTOR", battery.getCollectorId(), battery.getBatteryNo(),


                    collector.online() ? "ONLINE" : "OFFLINE", collector.online() ? value : Map.of());


        }


        return new LiveSource("COLLECTOR", battery.getCollectorId(), battery.getBatteryNo(), "OFFLINE", Map.of());


    }




    @PreDestroy void shutdown() { scheduler.shutdownNow(); }


}