实现测试暂停继续接收区间

yeg

2026-08-30 c2eefadb8239df552ef43d7ba55d5735bb603ed1

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

@@ -27,6 +27,8 @@
import com.ruoyi.system.domain.TDetectionChannel;
import com.ruoyi.system.domain.TTestReceptionInterval;
import com.ruoyi.system.domain.vo.TestConfigurationOptionsVo;
import com.ruoyi.device.bts.BtsDevelopmentSimulator;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorRealtimeMeasurement;
import com.ruoyi.system.mapper.TTestMapper;
import com.ruoyi.system.service.ITTestService;
import com.ruoyi.system.service.ISharedFieldSnapshotService;
@@ -49,6 +51,9 @@

    @Autowired
    private ISharedFieldSnapshotService sharedFieldSnapshotService;

    @Autowired(required = false)
    private BtsDevelopmentSimulator collectorSimulator;

    @Override
    public TTest selectTestById(Long testId)
@@ -229,6 +234,164 @@
        }
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public int pauseReception(Long testId, Long operatorId, String updateBy)
    {
        if (testId == null) throw new ServiceException("测试ID不能为空");
        if (operatorId == null) throw new ServiceException("操作人不能为空");
        TTest current = testMapper.selectTestByIdForUpdate(testId);
        if (current == null) throw new ServiceException("测试不存在");
        if (!"RECEIVING".equals(current.getReceptionStatus()))
            throw new ServiceException("当前测试不是接收中状态");

        List<TTestReceptionInterval> openIntervals = mutable(testMapper.selectOpenReceptionIntervalsForUpdate(testId));
        if (openIntervals.isEmpty()) throw new ServiceException("当前测试没有开放的接收区间");
        Date actionAt = new Date();
        Map<Long, Date> boundaries = new HashMap<>();
        // 先确认所有来源都能取得边界，再写入任何一条关闭记录，保证整项动作原子失败。
        for (TTestReceptionInterval open : openIntervals)
        {
            Date boundary = confirmBoundary(open, actionAt);
            boundaries.put(open.getIntervalId(), boundary);
            open.setEndBoundaryAtUtc(boundary);
            open.setIntervalStatus("CLOSED");
        }
        for (TTestReceptionInterval open : openIntervals)
        {
            if (testMapper.closeReceptionInterval(open.getIntervalId(), boundaries.get(open.getIntervalId()),
                    "暂停接收：采用来源最近成功观测边界") != 1)
                throw new ServiceException("接收区间已变更，请刷新后重试");
        }
        if (testMapper.updateReceptionPaused(testId, updateBy, actionAt) != 1)
            throw new ServiceException("测试状态已变更，请刷新后重试");
        testMapper.insertTestAudit(testId, "RECEPTION_PAUSED", operatorId, updateBy, null,
                JSON.toJSONString(Map.of("receptionStatus", "PAUSED", "intervalCount", openIntervals.size(),
                        "intervals", openIntervals)));
        return 1;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public int resumeReception(Long testId, Long operatorId, String updateBy)
    {
        if (testId == null) throw new ServiceException("测试ID不能为空");
        if (operatorId == null) throw new ServiceException("操作人不能为空");
        TTest current = testMapper.selectTestByIdForUpdate(testId);
        if (current == null) throw new ServiceException("测试不存在");
        if (!"PAUSED".equals(current.getReceptionStatus()))
            throw new ServiceException("当前测试不是暂停接收状态");
        loadConfiguration(current);
        if (current.getBatteries().isEmpty()) throw new ServiceException("测试未配置电池号");

        Date actionAt = new Date();
        List<TTestReceptionInterval> intervals = new ArrayList<>();
        Set<Long> channelIds = new HashSet<>();
        Set<String> collectorChannels = new HashSet<>();
        for (TTestBattery battery : current.getBatteries())
        {
            TTestChannelBinding binding = battery.getBinding();
            if (binding == null || binding.getChannelId() == null)
                throw new ServiceException("电池号 " + battery.getBatteryNo() + " 未绑定检测柜通道");
            if (!channelIds.add(binding.getChannelId())) throw new ServiceException("检测柜通道重复绑定，无法继续接收");
            TDetectionChannel channel = testMapper.selectReceptionChannelForUpdate(binding.getChannelId());
            if (channel == null || !isEnabled(channel.getStatus()) || !isOnline(channel.getOnlineStatus())
                    || channel.getLastSeenAtUtc() == null)
                throw new ServiceException("检测柜通道 " + binding.getChannelId() + " 当前无法确认继续接收边界");
            if (isEndedWorkStatus(channel.getWorkStatus()))
                throw new ServiceException("检测柜通道 " + binding.getChannelId() + " 已结束，不能继续接收");
            if (battery.getCollectorId() == null || testMapper.countCollectorEnabled(battery.getCollectorId()) == 0
                    || !testMapper.collectorOnline(battery.getCollectorId()))
                throw new ServiceException("采集器 " + battery.getCollectorId() + " 当前无法确认继续接收边界");
            Integer count = testMapper.selectCollectorChannelCount(battery.getCollectorId());
            validateCollectorChannels(battery.getCollectorId(), count == null ? 0 : count,
                    battery.getCollectorChannels(), collectorChannels);
            Date cabinetBoundary = Date.from(channel.getLastSeenAtUtc());
            ensureAfterPreviousBoundary(current.getTestId(), "CABINET_CHANNEL", binding.getChannelId(), null,
                    cabinetBoundary);
            intervals.add(interval(current, binding.getBindingId(), "CABINET_CHANNEL", binding.getChannelId(), null,
                    actionAt, cabinetBoundary, "继续接收：采用检测柜最近成功观测边界"));
            for (TTestCollectorChannel mapping : mutable(battery.getCollectorChannels()))
            {
                Date collectorBoundary = collectorBoundary(battery.getCollectorId(), mapping.getChannelCode());
                ensureAfterPreviousBoundary(current.getTestId(), "COLLECTOR_CHANNEL", battery.getCollectorId(),
                        mapping.getChannelCode(), collectorBoundary);
                intervals.add(interval(current, binding.getBindingId(), "COLLECTOR_CHANNEL", battery.getCollectorId(),
                        mapping.getChannelCode(), actionAt, collectorBoundary, "继续接收：采用采集器最近成功观测边界"));
            }
        }
        for (TTestReceptionInterval interval : intervals) testMapper.insertReceptionInterval(interval);
        if (testMapper.updateReceptionResumed(testId, updateBy, actionAt) != 1)
            throw new ServiceException("测试状态已变更，请刷新后重试");
        testMapper.insertTestAudit(testId, "RECEPTION_RESUMED", operatorId, updateBy, null,
                JSON.toJSONString(Map.of("receptionStatus", "RECEIVING", "intervalCount", intervals.size(),
                        "intervals", intervals)));
        return 1;
    }

    @Override
    public List<TTestReceptionInterval> selectReceptionIntervals(Long testId)
    {
        if (testId == null) throw new ServiceException("测试ID不能为空");
        selectTestById(testId);
        return mutable(testMapper.selectReceptionIntervals(testId));
    }

    private Date confirmBoundary(TTestReceptionInterval interval, Date actionAt)
    {
        if ("CABINET_CHANNEL".equals(interval.getSourceType()))
        {
            TDetectionChannel channel = testMapper.selectReceptionChannelForUpdate(interval.getSourceId());
            if (channel == null || !isEnabled(channel.getStatus()) || !isOnline(channel.getOnlineStatus())
                    || channel.getLastSeenAtUtc() == null)
                throw new ServiceException("检测柜通道 " + interval.getSourceId() + " 当前无法确认暂停接收边界");
            Date boundary = Date.from(channel.getLastSeenAtUtc());
            if (interval.getStartBoundaryAtUtc() != null && boundary.before(interval.getStartBoundaryAtUtc()))
                throw new ServiceException("检测柜通道 " + interval.getSourceId() + " 的暂停边界早于区间开始");
            return boundary;
        }
        if (!testMapper.collectorOnline(interval.getSourceId()))
            throw new ServiceException("采集器 " + interval.getSourceId() + " 当前无法确认暂停接收边界");
        Date boundary = collectorBoundary(interval.getSourceId(), interval.getSourceChannelCode());
        if (interval.getStartBoundaryAtUtc() != null && boundary.before(interval.getStartBoundaryAtUtc()))
            throw new ServiceException("采集器 " + interval.getSourceId() + " 的暂停边界早于区间开始");
        return boundary;
    }

    private Date collectorBoundary(Long collectorId, String channelCode)
    {
        if (collectorSimulator == null) throw new ServiceException("采集器 " + collectorId + " 当前无法取得成功观测边界");
        try
        {
            return collectorSimulator.collectorRealtime(collectorId.intValue()).measurements().stream()
                    .filter(item -> channelCode == null || channelCode.equals(item.channelCode()))
                    .map(CollectorRealtimeMeasurement::deviceTime).filter(java.util.Objects::nonNull)
                    .max(java.time.Instant::compareTo).map(Date::from)
                    .orElseThrow(() -> new ServiceException("采集器 " + collectorId + " 当前无法取得成功观测边界"));
        }
        catch (RuntimeException exception)
        {
            if (exception instanceof ServiceException serviceException) throw serviceException;
            throw new ServiceException("采集器 " + collectorId + " 当前无法取得成功观测边界");
        }
    }

    private void ensureAfterPreviousBoundary(Long testId, String sourceType, Long sourceId,
            String sourceChannelCode, Date boundary)
    {
        TTestReceptionInterval previous = testMapper.selectLatestReceptionIntervalForUpdate(testId, sourceType,
                sourceId, sourceChannelCode);
        if (previous != null && previous.getEndBoundaryAtUtc() != null
                && !boundary.after(previous.getEndBoundaryAtUtc()))
            throw new ServiceException("数据来源尚未取得新的接收边界，请稍后重试");
    }

    private boolean isEndedWorkStatus(String workStatus)
    {
        return "STOPPED".equalsIgnoreCase(workStatus) || "FINISHED".equalsIgnoreCase(workStatus)
                || "ENDED".equalsIgnoreCase(workStatus);
    }

    private TTestReceptionInterval interval(TTest test, Long bindingId, String sourceType, Long sourceId,
            String sourceChannelCode, Date actionAt, Date boundaryAt, String note)
    {