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

yeg

2026-08-30 c2eefadb8239df552ef43d7ba55d5735bb603ed1

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

新增暂停、继续接收事务与实际采样边界校验，记录多段接收区间和追加审计。补充区间查询接口、权限迁移及服务契约测试。

 ruoyi-admin/src/main/java/com/ruoyi/web/controller/system/TTestController.java

@@ -20,6 +20,7 @@
import com.ruoyi.common.enums.BusinessType;
import com.ruoyi.system.domain.TTest;
import com.ruoyi.system.domain.TTestResult;
import com.ruoyi.system.domain.TTestReceptionInterval;
import com.ruoyi.system.service.ITTestService;

/** 测试管理 Controller。 */
@@ -95,6 +96,35 @@
        return success(testService.selectTestById(testId));
    }

    /** 独立暂停接收动作；只关闭系统接收区间，不向设备发送控制命令。 */
    @PreAuthorize("@ss.hasPermi('system:test:pause')")
    @Log(title = "测试数据接收", businessType = BusinessType.UPDATE)
    @PostMapping("/{testId}/reception/pause")
    public AjaxResult pauseReception(@PathVariable Long testId)
    {
        testService.pauseReception(testId, getUserId(), getUsername());
        return success(testService.selectTestById(testId));
    }

    /** 独立继续接收动作；从新的成功观测边界打开区间，不向设备发送控制命令。 */
    @PreAuthorize("@ss.hasPermi('system:test:resume')")
    @Log(title = "测试数据接收", businessType = BusinessType.UPDATE)
    @PostMapping("/{testId}/reception/resume")
    public AjaxResult resumeReception(@PathVariable Long testId)
    {
        testService.resumeReception(testId, getUserId(), getUsername());
        return success(testService.selectTestById(testId));
    }

    /** 返回测试各来源的接收区间、动作时间和实际采样边界。 */
    @PreAuthorize("@ss.hasPermi('system:test:query')")
    @GetMapping("/{testId}/reception/intervals")
    public AjaxResult receptionIntervals(@PathVariable Long testId)
    {
        List<TTestReceptionInterval> intervals = testService.selectReceptionIntervals(testId);
        return success(intervals);
    }

    @PreAuthorize("@ss.hasPermi('system:test:remove')")
    @Log(title = "测试管理", businessType = BusinessType.DELETE)
    @DeleteMapping("/{testIds}")

 ruoyi-admin/src/main/resources/db/migration/V15__pause_resume_reception_actions.sql

New file
@@ -0,0 +1,27 @@
-- Ticket10：暂停/继续接收权限与区间查询索引；不改变已应用迁移。
ALTER TABLE t_test_reception_interval
    ADD KEY idx_reception_interval_status (test_id, interval_status, interval_id);

SET @pause_menu_id := (SELECT COALESCE(MAX(menu_id), 0) + 1 FROM sys_menu);
INSERT INTO sys_menu
    (menu_id, menu_name, parent_id, order_num, path, component, query, route_name,
     is_frame, is_cache, menu_type, visible, status, perms, icon,
     create_by, create_time, update_by, update_time, remark)
SELECT @pause_menu_id, '暂停接收', menu_id, 7, '', NULL, NULL, '', 1, 0, 'F', '0', '0',
       'system:test:pause', '#', 'admin', NOW(), '', NULL, '测试暂停接收权限'
FROM sys_menu
WHERE perms = 'system:test:list'
  AND NOT EXISTS (SELECT 1 FROM sys_menu existing WHERE existing.perms = 'system:test:pause')
LIMIT 1;

SET @resume_menu_id := (SELECT COALESCE(MAX(menu_id), 0) + 1 FROM sys_menu);
INSERT INTO sys_menu
    (menu_id, menu_name, parent_id, order_num, path, component, query, route_name,
     is_frame, is_cache, menu_type, visible, status, perms, icon,
     create_by, create_time, update_by, update_time, remark)
SELECT @resume_menu_id, '继续接收', menu_id, 8, '', NULL, NULL, '', 1, 0, 'F', '0', '0',
       'system:test:resume', '#', 'admin', NOW(), '', NULL, '测试继续接收权限'
FROM sys_menu
WHERE perms = 'system:test:list'
  AND NOT EXISTS (SELECT 1 FROM sys_menu existing WHERE existing.perms = 'system:test:resume')
LIMIT 1;

 ruoyi-admin/src/test/java/com/ruoyi/system/service/impl/TTestServiceImplTest.java

@@ -15,12 +15,16 @@
import java.math.BigDecimal;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.time.Instant;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import com.ruoyi.device.bts.BtsDevelopmentSimulator;
import com.ruoyi.device.simulator.DevelopmentCollectorSimulation.CollectorRealtimeSnapshot;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorRealtimeMeasurement;
import org.springframework.dao.DuplicateKeyException;
import com.ruoyi.common.exception.ServiceException;
import com.ruoyi.system.domain.TTest;
@@ -31,6 +35,7 @@
import com.ruoyi.system.domain.TTestPlanStep;
import com.ruoyi.system.domain.TTestWarningThreshold;
import com.ruoyi.system.domain.TDetectionChannel;
import com.ruoyi.system.domain.TTestReceptionInterval;
import com.ruoyi.system.mapper.TTestMapper;
import com.ruoyi.system.service.ISharedFieldSnapshotService;

@@ -43,6 +48,9 @@

    @Mock
    private ISharedFieldSnapshotService sharedFieldSnapshotService;

    @Mock
    private BtsDevelopmentSimulator collectorSimulator;

    @InjectMocks
    private TTestServiceImpl testService;
@@ -488,6 +496,91 @@
        verify(testMapper, never()).selectTestByIdForUpdate(anyLong());
    }

    @Test
    void pauseReceptionShouldCloseAllOpenIntervalsAndAppendAudit()
    {
        TTest current = validTest();
        current.setTestId(10L);
        current.setReceptionStatus("RECEIVING");
        when(testMapper.selectTestByIdForUpdate(10L)).thenReturn(current);
        TTestReceptionInterval cabinet = receptionInterval(1L, "CABINET_CHANNEL", 100L, null);
        TTestReceptionInterval collector = receptionInterval(2L, "COLLECTOR_CHANNEL", 8L, "S-01");
        when(testMapper.selectOpenReceptionIntervalsForUpdate(10L)).thenReturn(List.of(cabinet, collector));
        TDetectionChannel channel = new TDetectionChannel();
        channel.setChannelId(100L); channel.setOnlineStatus("ONLINE"); channel.setLastSeenAtUtc(Instant.parse("2026-08-29T12:00:00Z"));
        when(testMapper.selectReceptionChannelForUpdate(100L)).thenReturn(channel);
        when(testMapper.collectorOnline(8L)).thenReturn(true);
        mockCollectorBoundary();
        when(testMapper.closeReceptionInterval(anyLong(), any(), any())).thenReturn(1);
        when(testMapper.updateReceptionPaused(eq(10L), eq("operator"), any())).thenReturn(1);

        assertEquals(1, testService.pauseReception(10L, 7L, "operator"));
        verify(testMapper, times(2)).closeReceptionInterval(anyLong(), any(), any());
        verify(testMapper).updateReceptionPaused(eq(10L), eq("operator"), any());
        verify(testMapper).insertTestAudit(eq(10L), eq("RECEPTION_PAUSED"), eq(7L), eq("operator"), isNull(), any());
    }

    @Test
    void pauseReceptionShouldFailBeforeWritesWhenBoundaryUnavailable()
    {
        TTest current = validTest();
        current.setTestId(10L); current.setReceptionStatus("RECEIVING");
        when(testMapper.selectTestByIdForUpdate(10L)).thenReturn(current);
        TTestReceptionInterval interval = receptionInterval(1L, "CABINET_CHANNEL", 100L, null);
        when(testMapper.selectOpenReceptionIntervalsForUpdate(10L)).thenReturn(List.of(interval));
        when(testMapper.selectReceptionChannelForUpdate(100L)).thenReturn(null);

        ServiceException exception = assertThrows(ServiceException.class,
                () -> testService.pauseReception(10L, 7L, "operator"));
        assertEquals("检测柜通道 100 当前无法确认暂停接收边界", exception.getMessage());
        verify(testMapper, never()).closeReceptionInterval(anyLong(), any(), any());
        verify(testMapper, never()).updateReceptionPaused(anyLong(), any(), any());
    }

    @Test
    void resumeReceptionShouldOpenNewIntervalsAndAppendAudit()
    {
        TTest current = validTest();
        current.setTestId(10L); current.setReceptionStatus("PAUSED");
        current.getBatteries().get(0).getBinding().setBindingId(30L);
        when(testMapper.selectTestByIdForUpdate(10L)).thenReturn(current);
        TDetectionChannel channel = new TDetectionChannel();
        channel.setChannelId(100L); channel.setOnlineStatus("ONLINE"); channel.setLastSeenAtUtc(Instant.parse("2026-08-29T12:00:00Z"));
        when(testMapper.selectReceptionChannelForUpdate(100L)).thenReturn(channel);
        when(testMapper.countCollectorEnabled(8L)).thenReturn(1);
        when(testMapper.collectorOnline(8L)).thenReturn(true);
        when(testMapper.selectCollectorChannelCount(8L)).thenReturn(4);
        mockCollectorBoundary();
        when(testMapper.updateReceptionResumed(eq(10L), eq("operator"), any())).thenReturn(1);

        assertEquals(1, testService.resumeReception(10L, 7L, "operator"));
        verify(testMapper, times(4)).insertReceptionInterval(any());
        verify(testMapper).updateReceptionResumed(eq(10L), eq("operator"), any());
        verify(testMapper).insertTestAudit(eq(10L), eq("RECEPTION_RESUMED"), eq(7L), eq("operator"), isNull(), any());
    }

    private TTestReceptionInterval receptionInterval(Long id, String sourceType, Long sourceId, String channelCode)
    {
        TTestReceptionInterval interval = new TTestReceptionInterval();
        interval.setIntervalId(id); interval.setTestId(10L); interval.setSourceType(sourceType);
        interval.setSourceId(sourceId); interval.setSourceChannelCode(channelCode); interval.setIntervalStatus("OPEN");
        return interval;
    }

    private void mockCollectorBoundary()
    {
        Instant deviceTime = Instant.parse("2026-08-29T12:00:00Z");
        CollectorRealtimeMeasurement measurement = new CollectorRealtimeMeasurement("dev-sim", "collector-8", "S-01",
                1L, deviceTime, "2026-08-29 20:00:00", deviceTime, "PRIMARY_TEMPERATURE", new BigDecimal("25"), "℃");
        CollectorRealtimeMeasurement positive = new CollectorRealtimeMeasurement("dev-sim", "collector-8", "S-02",
                1L, deviceTime, "2026-08-29 20:00:00", deviceTime, "PRIMARY_POSITIVE_POTENTIAL", new BigDecimal("1"), "V");
        CollectorRealtimeMeasurement negative = new CollectorRealtimeMeasurement("dev-sim", "collector-8", "S-03",
                1L, deviceTime, "2026-08-29 20:00:00", deviceTime, "PRIMARY_NEGATIVE_POTENTIAL", new BigDecimal("1"), "V");
        when(collectorSimulator.collectorRealtime(8)).thenReturn(
                new CollectorRealtimeSnapshot("collector-8", true, 1000,
                        List.of(measurement, positive, negative)));
    }

    private void mockEnabledOptions()
    {
        mockEnabledPeopleAndType();

 ruoyi-admin/src/test/java/com/ruoyi/web/controller/system/TTestControllerContractTest.java

@@ -4,6 +4,7 @@
import java.lang.reflect.Method;
import org.junit.jupiter.api.Test;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.security.access.prepost.PreAuthorize;

class TTestControllerContractTest
@@ -18,6 +19,14 @@
        assertPermission("startReception", "@ss.hasPermi('system:test:start')", Long.class);
        Method startReception = TTestController.class.getMethod("startReception", Long.class);
        assertEquals("/{testId}/reception/start", startReception.getAnnotation(PostMapping.class).value()[0]);
        assertPermission("pauseReception", "@ss.hasPermi('system:test:pause')", Long.class);
        assertEquals("/{testId}/reception/pause", TTestController.class.getMethod("pauseReception", Long.class)
                .getAnnotation(PostMapping.class).value()[0]);
        assertPermission("resumeReception", "@ss.hasPermi('system:test:resume')", Long.class);
        assertEquals("/{testId}/reception/resume", TTestController.class.getMethod("resumeReception", Long.class)
                .getAnnotation(PostMapping.class).value()[0]);
        assertEquals("/{testId}/reception/intervals", TTestController.class.getMethod("receptionIntervals", Long.class)
                .getAnnotation(GetMapping.class).value()[0]);
    }

    private void assertPermission(String methodName, String permission, Class<?>... parameterTypes) throws Exception

 ruoyi-system/src/main/java/com/ruoyi/system/mapper/TTestMapper.java

@@ -78,4 +78,15 @@
    int updateReceptionStarted(@Param("testId") Long testId, @Param("startedAtUtc") Date startedAtUtc,
            @Param("updateBy") String updateBy);
    int insertReceptionInterval(TTestReceptionInterval interval);
    List<TTestReceptionInterval> selectOpenReceptionIntervalsForUpdate(Long testId);
    List<TTestReceptionInterval> selectReceptionIntervals(Long testId);
    TTestReceptionInterval selectLatestReceptionIntervalForUpdate(@Param("testId") Long testId,
            @Param("sourceType") String sourceType, @Param("sourceId") Long sourceId,
            @Param("sourceChannelCode") String sourceChannelCode);
    int closeReceptionInterval(@Param("intervalId") Long intervalId, @Param("endBoundaryAtUtc") Date endBoundaryAtUtc,
            @Param("boundaryNote") String boundaryNote);
    int updateReceptionPaused(@Param("testId") Long testId, @Param("updateBy") String updateBy,
            @Param("updatedAtUtc") Date updatedAtUtc);
    int updateReceptionResumed(@Param("testId") Long testId, @Param("updateBy") String updateBy,
            @Param("updatedAtUtc") Date updatedAtUtc);
}

 ruoyi-system/src/main/java/com/ruoyi/system/service/ITTestService.java

@@ -29,6 +29,15 @@
    /** 原子开始接收：校验全部来源并从实际观测边界打开区间。 */
    int startReception(Long testId, Long operatorId, String updateBy);

    /** 原子暂停接收：关闭全部开放区间并保留设备实时观察。 */
    int pauseReception(Long testId, Long operatorId, String updateBy);

    /** 原子继续接收：从新的成功观测边界打开全部来源区间。 */
    int resumeReception(Long testId, Long operatorId, String updateBy);

    /** 查询测试各数据来源的接收区间和实际边界。 */
    List<com.ruoyi.system.domain.TTestReceptionInterval> selectReceptionIntervals(Long testId);

    /** 仅删除未开始接收的测试并追加审计记录。 */
    int deleteTestByIds(Long[] testIds, Long operatorId, String updateBy);


 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)
    {

 ruoyi-system/src/main/resources/mapper/system/TTestMapper.xml

@@ -399,4 +399,52 @@
        values(#{testId}, #{bindingId}, #{sourceType}, #{sourceId}, #{sourceChannelCode}, #{actionAtUtc},
               #{startBoundaryAtUtc}, #{boundaryNote}, #{intervalStatus})
    </insert>
    <select id="selectOpenReceptionIntervalsForUpdate" parameterType="Long"
            resultType="com.ruoyi.system.domain.TTestReceptionInterval">
        select interval_id intervalId, test_id testId, binding_id bindingId, source_type sourceType,
               source_id sourceId, source_channel_code sourceChannelCode, action_at_utc actionAtUtc,
               start_boundary_at_utc startBoundaryAtUtc, end_boundary_at_utc endBoundaryAtUtc,
               boundary_note boundaryNote, interval_status intervalStatus
        from t_test_reception_interval
        where test_id = #{testId} and interval_status = 'OPEN'
        order by interval_id
        for update
    </select>
    <select id="selectReceptionIntervals" parameterType="Long"
            resultType="com.ruoyi.system.domain.TTestReceptionInterval">
        select interval_id intervalId, test_id testId, binding_id bindingId, source_type sourceType,
               source_id sourceId, source_channel_code sourceChannelCode, action_at_utc actionAtUtc,
               start_boundary_at_utc startBoundaryAtUtc, end_boundary_at_utc endBoundaryAtUtc,
               boundary_note boundaryNote, interval_status intervalStatus
        from t_test_reception_interval
        where test_id = #{testId}
        order by interval_id
    </select>
    <select id="selectLatestReceptionIntervalForUpdate" resultType="com.ruoyi.system.domain.TTestReceptionInterval">
        select interval_id intervalId, test_id testId, binding_id bindingId, source_type sourceType,
               source_id sourceId, source_channel_code sourceChannelCode, action_at_utc actionAtUtc,
               start_boundary_at_utc startBoundaryAtUtc, end_boundary_at_utc endBoundaryAtUtc,
               boundary_note boundaryNote, interval_status intervalStatus
        from t_test_reception_interval
        where test_id = #{testId} and source_type = #{sourceType} and source_id = #{sourceId}
          and (source_channel_code = #{sourceChannelCode} or (source_channel_code is null and #{sourceChannelCode} is null))
        order by interval_id desc
        limit 1
        for update
    </select>
    <update id="closeReceptionInterval">
        update t_test_reception_interval
        set end_boundary_at_utc = #{endBoundaryAtUtc}, boundary_note = #{boundaryNote}, interval_status = 'CLOSED'
        where interval_id = #{intervalId} and interval_status = 'OPEN'
    </update>
    <update id="updateReceptionPaused">
        update t_test
        set test_status = 'PAUSED', reception_status = 'PAUSED', update_by = #{updateBy}, update_time = #{updatedAtUtc}
        where test_id = #{testId} and reception_status = 'RECEIVING' and del_flag = 0
    </update>
    <update id="updateReceptionResumed">
        update t_test
        set test_status = 'RUNNING', reception_status = 'RECEIVING', update_by = #{updateBy}, update_time = #{updatedAtUtc}
        where test_id = #{testId} and reception_status = 'PAUSED' and del_flag = 0
    </update>
</mapper>