From c2eefadb8239df552ef43d7ba55d5735bb603ed1 Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期日, 30 八月 2026 14:50:28 +0800
Subject: [PATCH] 实现测试暂停继续接收区间
---
ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TTestServiceImpl.java | 163 ++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 163 insertions(+), 0 deletions(-)
diff --git a/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TTestServiceImpl.java b/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TTestServiceImpl.java
index e2fedad..a60fe2f 100644
--- a/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TTestServiceImpl.java
+++ b/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)
{
--
Gitblit v1.7.1