refactor: unify station reroute execution flow

Junjie

3 小时以前 6296120dadc9088e7c2318358abf4b8bdef01ad2

refactor: unify station reroute execution flow

 src/main/java/com/zy/core/utils/StationOperateProcessUtils.java

@@ -483,169 +483,25 @@
                        }
                        redisUtil.set(RedisKeyType.CHECK_STATION_RUN_BLOCK_LIMIT_.key + stationProtocol.getTaskNo(), "lock", 15);

                        if (wrkMast.getIoType() == WrkIoType.IN.id && runBlockReassignLocStationList.contains(stationProtocol.getStationId())) {
                            //站点处于重新分配库位区域
                            int currentTaskBufferCommandCount = countCurrentTaskBufferCommands(
                                    stationProtocol.getTaskBufferItems(),
                                    stationProtocol.getTaskNo()
                            );
                            if (currentTaskBufferCommandCount > 0) {
                                News.info("输送站点运行堵塞重分配已跳过，缓存区仍存在当前任务命令。站点号={}，工作号={}，当前任务命令数={}",
                                        stationProtocol.getStationId(),
                                        stationProtocol.getTaskNo(),
                                        currentTaskBufferCommandCount);
                                continue;
                            }
                            if (stationMoveCoordinator != null) {
                                stationMoveCoordinator.cancelSession(wrkMast.getWrkNo());
                            }
                            //运行堵塞，重新申请任务
                            String response = wmsOperateUtils.applyReassignTaskLocNo(wrkMast.getWrkNo(), stationProtocol.getStationId());
                            if (Cools.isEmpty(response)) {
                                News.taskError(wrkMast.getWrkNo(), "请求WMS重新分配库位接口失败，接口未响应！！！response：{}", response);
                                continue;
                            }
                            JSONObject jsonObject = JSON.parseObject(response);
                            if (jsonObject.getInteger("code").equals(200)) {
                                StartupDto dto = jsonObject.getObject("data", StartupDto.class);

                                String sourceLocNo = wrkMast.getLocNo();
                                String locNo = dto.getLocNo();

                                LocMast sourceLocMast = locMastService.queryByLoc(sourceLocNo);
                                if (sourceLocMast == null) {
                                    News.taskInfo(wrkMast.getWrkNo(), "库位号:{} 源库位信息不存在", sourceLocNo);
                                    continue;
                                }

                                if (!sourceLocMast.getLocSts().equals("S")) {
                                    News.taskInfo(wrkMast.getWrkNo(), "库位号:{} 源库位状态不处于入库预约", sourceLocNo);
                                    continue;
                                }

                                LocMast locMast = locMastService.queryByLoc(locNo);
                                if (locMast == null) {
                                    News.taskInfo(wrkMast.getWrkNo(), "库位号:{} 目标库位信息不存在", locNo);
                                    continue;
                                }

                                if (!locMast.getLocSts().equals("O")) {
                                    News.taskInfo(wrkMast.getWrkNo(), "库位号:{} 目标库位状态不处于空库位", locNo);
                                    continue;
                                }

                                FindCrnNoResult findCrnNoResult = commonService.findCrnNoByLocNo(locNo);
                                if (findCrnNoResult == null) {
                                    News.taskInfo(wrkMast.getWrkNo(), "{}工作,未匹配到堆垛机", wrkMast.getWrkNo());
                                    continue;
                                }
                                Integer crnNo = findCrnNoResult.getCrnNo();

                                Integer targetStationId = commonService.findInStationId(findCrnNoResult, stationProtocol.getStationId());
                                if (targetStationId == null) {
                                    News.taskInfo(wrkMast.getWrkNo(), "{}站点,搜索入库站点失败", stationProtocol.getStationId());
                                    continue;
                                }

                                StationCommand command = stationThread.getCommand(StationCommandType.MOVE, wrkMast.getWrkNo(), stationProtocol.getStationId(), targetStationId, 0);
                                if (command == null) {
                                    News.taskInfo(wrkMast.getWrkNo(), "{}工作,获取输送线命令失败", wrkMast.getWrkNo());
                                    continue;
                                }

                                //更新源库位
                                sourceLocMast.setLocSts("O");
                                sourceLocMast.setModiTime(new Date());
                                locMastService.updateById(sourceLocMast);

                                //更新目标库位
                                locMast.setLocSts("S");
                                locMast.setModiTime(new Date());
                                locMastService.updateById(locMast);

                                //更新工作档数据
                                wrkMast.setLocNo(locNo);
                                wrkMast.setStaNo(targetStationId);

                                if (findCrnNoResult.getCrnType().equals(SlaveType.Crn)) {
                                    wrkMast.setCrnNo(crnNo);
                                } else if (findCrnNoResult.getCrnType().equals(SlaveType.DualCrn)) {
                                    wrkMast.setDualCrnNo(crnNo);
                                } else {
                                    throw new CoolException("未知设备类型");
                                }

                                if (wrkMastService.updateById(wrkMast)) {
                                    boolean offered = offerDevpCommandWithDedup(basDevp.getDevpNo(), command, "checkStationRunBlock_direct");
                                    if (!offered) {
                                        continue;
                                    }
                                    if (stationMoveCoordinator != null) {
                                        stationMoveCoordinator.recordDispatch(
                                                wrkMast.getWrkNo(),
                                                stationProtocol.getStationId(),
                                                "checkStationRunBlock_direct",
                                                command,
                                                false
                                        );
                                    }
                                }
                            } else {
                                News.error("请求WMS接口失败！！！response：{}", response);
                            }
                        } else {
                            //运行堵塞，重新计算路线
                            Double pathLenFactor = resolveOutboundPathLenFactor(wrkMast);
                            OutOrderDispatchDecision dispatchDecision = resolveOutboundDispatchDecision(
                                    stationProtocol.getStationId(),
                                    wrkMast,
                                    outOrderStationIds,
                                    pathLenFactor
                            );
                            Integer moveStaNo = dispatchDecision == null ? null : dispatchDecision.getTargetStationId();
                            if (moveStaNo == null || Objects.equals(moveStaNo, stationProtocol.getStationId())) {
                                continue;
                            }

                            if (countCurrentTaskBufferCommands(stationProtocol.getTaskBufferItems(), stationProtocol.getTaskNo()) > 0) {
                                continue;
                            }

                            StationCommand command = stationThread.getRunBlockRerouteCommand(
                                    wrkMast.getWrkNo(),
                                    stationProtocol.getStationId(),
                                    moveStaNo,
                                    0,
                                    pathLenFactor
                            );
                            if (command == null) {
                                News.taskInfo(wrkMast.getWrkNo(),
                                        "输送站点堵塞重规划未找到可下发路线，当前站点={}，目标站点={}",
                                        stationProtocol.getStationId(),
                                        moveStaNo);
                                continue;
                            }

                            if (stationMoveCoordinator != null) {
                                stationMoveCoordinator.cancelSession(wrkMast.getWrkNo());
                            }
                            resetSegmentMoveCommandsBeforeReroute(wrkMast.getWrkNo());
                            boolean offered = offerDevpCommandWithDedup(basDevp.getDevpNo(), command, "checkStationRunBlock_reroute");
                            if (!offered) {
                                continue;
                            }
                            syncOutOrderWatchState(wrkMast, stationProtocol.getStationId(), outOrderStationIds, dispatchDecision, command);
                            if (stationMoveCoordinator != null) {
                                stationMoveCoordinator.recordDispatch(
                                        wrkMast.getWrkNo(),
                                        stationProtocol.getStationId(),
                                        "checkStationRunBlock_reroute",
                                        command,
                                        dispatchDecision != null && dispatchDecision.isCircle()
                                );
                            }
                            News.info("输送站点堵塞后重新计算路径命令下发成功，站点号={}，工作号={}，命令数据={}", stationProtocol.getStationId(), wrkMast.getWrkNo(), JSON.toJSONString(command));
                        if (shouldUseRunBlockDirectReassign(wrkMast, stationProtocol.getStationId(), runBlockReassignLocStationList)) {
                            executeRunBlockDirectReassign(basDevp, stationThread, stationProtocol, wrkMast);
                            continue;
                        }

                        Double pathLenFactor = resolveOutboundPathLenFactor(wrkMast);
                        RerouteContext context = RerouteContext.create(
                                RerouteSceneType.RUN_BLOCK_REROUTE,
                                basDevp,
                                stationThread,
                                stationProtocol,
                                wrkMast,
                                outOrderStationIds,
                                pathLenFactor,
                                "checkStationRunBlock_reroute"
                        ).withRunBlockCommand()
                                .withCancelSessionBeforeDispatch()
                                .withResetSegmentCommandsBeforeDispatch();
                        executeSharedReroute(context);
                    }
                }
            }
@@ -878,6 +734,19 @@
                context.pathLenFactor()
        );
        if (command == null) {
            if (context.sceneType() == RerouteSceneType.RUN_BLOCK_REROUTE) {
                News.taskInfo(context.wrkMast().getWrkNo(),
                        "输送站点堵塞重规划未找到可下发路线，当前站点={}，目标站点={}",
                        currentStationId,
                        targetStationId);
            } else if (context.sceneType() == RerouteSceneType.IDLE_RECOVER) {
                News.taskInfo(context.wrkMast().getWrkNo(),
                        "站点任务停留超时后重算路径失败，当前站点={}，目标站点={}",
                        currentStationId,
                        targetStationId);
            } else {
                News.taskInfo(context.wrkMast().getWrkNo(), "获取输送线命令失败");
            }
            return RerouteCommandPlan.skip("missing-command");
        }
        return RerouteCommandPlan.dispatch(command, decision, context.dispatchScene());
@@ -908,6 +777,12 @@
            return RerouteExecutionResult.skip("recent-dispatch");
        }
        if (countCurrentTaskBufferCommands(stationProtocol.getTaskBufferItems(), taskNo) > 0) {
            if (context.sceneType() == RerouteSceneType.IDLE_RECOVER) {
                News.info("输送站点任务停留超时，但缓存区仍存在当前任务命令，已跳过重算。站点号={}，工作号={}，当前任务命令数={}",
                        stationId,
                        taskNo,
                        countCurrentTaskBufferCommands(stationProtocol.getTaskBufferItems(), taskNo));
            }
            return RerouteExecutionResult.skip("buffer-has-current-task");
        }
        if (context.checkSuppressDispatch()
@@ -1571,67 +1446,143 @@
        if (lock != null) {
            return;
        }
        Double pathLenFactor = resolveOutboundPathLenFactor(wrkMast);
        RerouteContext context = RerouteContext.create(
                RerouteSceneType.IDLE_RECOVER,
                basDevp,
                stationThread,
                stationProtocol,
                wrkMast,
                outOrderList,
                pathLenFactor,
                "checkStationIdleRecover"
        ).withCancelSessionBeforeDispatch()
                .withExecutionLock(RedisKeyType.CHECK_STATION_IDLE_RECOVER_LIMIT_.key + stationProtocol.getTaskNo(), STATION_IDLE_RECOVER_LIMIT_SECONDS)
                .withResetSegmentCommandsBeforeDispatch()
                .clearIdleIssuedCommands(idleTrack);
        executeSharedReroute(context);
    }

    boolean shouldUseRunBlockDirectReassign(WrkMast wrkMast,
                                            Integer stationId,
                                            List<Integer> runBlockReassignLocStationList) {
        return wrkMast != null
                && Objects.equals(wrkMast.getIoType(), WrkIoType.IN.id)
                && stationId != null
                && runBlockReassignLocStationList != null
                && runBlockReassignLocStationList.contains(stationId);
    }

    private void executeRunBlockDirectReassign(BasDevp basDevp,
                                               StationThread stationThread,
                                               StationProtocol stationProtocol,
                                               WrkMast wrkMast) {
        if (basDevp == null || stationThread == null || stationProtocol == null || wrkMast == null) {
            return;
        }
        int currentTaskBufferCommandCount = countCurrentTaskBufferCommands(
                stationProtocol.getTaskBufferItems(),
                stationProtocol.getTaskNo()
        );
        if (currentTaskBufferCommandCount > 0) {
            News.info("输送站点任务停留超时，但缓存区仍存在当前任务命令，已跳过重算。站点号={}，工作号={}，当前任务命令数={}",
            News.info("输送站点运行堵塞重分配已跳过，缓存区仍存在当前任务命令。站点号={}，工作号={}，当前任务命令数={}",
                    stationProtocol.getStationId(),
                    stationProtocol.getTaskNo(),
                    currentTaskBufferCommandCount);
            return;
        }

        Double pathLenFactor = resolveOutboundPathLenFactor(wrkMast);
        OutOrderDispatchDecision dispatchDecision = null;
        Integer moveStaNo;
        if (Objects.equals(wrkMast.getWrkSts(), WrkStsType.STATION_RUN.sts)) {
            dispatchDecision = resolveOutboundDispatchDecision(stationProtocol.getStationId(), wrkMast, outOrderList, pathLenFactor);
            moveStaNo = dispatchDecision == null ? null : dispatchDecision.getTargetStationId();
        } else {
            moveStaNo = wrkMast.getStaNo();
        }
        if (moveStaNo == null || Objects.equals(moveStaNo, stationProtocol.getStationId())) {
            return;
        }

        if (stationMoveCoordinator != null) {
            stationMoveCoordinator.cancelSession(stationProtocol.getTaskNo());
            stationMoveCoordinator.cancelSession(wrkMast.getWrkNo());
        }
        redisUtil.set(RedisKeyType.CHECK_STATION_IDLE_RECOVER_LIMIT_.key + stationProtocol.getTaskNo(), "lock", STATION_IDLE_RECOVER_LIMIT_SECONDS);
        resetSegmentMoveCommandsBeforeReroute(stationProtocol.getTaskNo());
        int clearedCommandCount = clearIssuedMoveCommandsDuringIdleStay(idleTrack, stationProtocol.getTaskNo(), stationProtocol.getStationId());

        StationCommand command = buildOutboundMoveCommand(
                stationThread,
                wrkMast,
                stationProtocol.getStationId(),
                moveStaNo,
                pathLenFactor
        );
        if (command == null) {
            News.taskInfo(wrkMast.getWrkNo(), "站点任务停留超时后重算路径失败，当前站点={}，目标站点={}", stationProtocol.getStationId(), moveStaNo);
        String response = wmsOperateUtils.applyReassignTaskLocNo(wrkMast.getWrkNo(), stationProtocol.getStationId());
        if (Cools.isEmpty(response)) {
            News.taskError(wrkMast.getWrkNo(), "请求WMS重新分配库位接口失败，接口未响应！！！response：{}", response);
            return;
        }
        JSONObject jsonObject = JSON.parseObject(response);
        if (!jsonObject.getInteger("code").equals(200)) {
            News.error("请求WMS接口失败！！！response：{}", response);
            return;
        }

        boolean offered = offerDevpCommandWithDedup(basDevp.getDevpNo(), command, "checkStationIdleRecover");
        StartupDto dto = jsonObject.getObject("data", StartupDto.class);
        String sourceLocNo = wrkMast.getLocNo();
        String locNo = dto.getLocNo();

        LocMast sourceLocMast = locMastService.queryByLoc(sourceLocNo);
        if (sourceLocMast == null) {
            News.taskInfo(wrkMast.getWrkNo(), "库位号:{} 源库位信息不存在", sourceLocNo);
            return;
        }
        if (!sourceLocMast.getLocSts().equals("S")) {
            News.taskInfo(wrkMast.getWrkNo(), "库位号:{} 源库位状态不处于入库预约", sourceLocNo);
            return;
        }

        LocMast locMast = locMastService.queryByLoc(locNo);
        if (locMast == null) {
            News.taskInfo(wrkMast.getWrkNo(), "库位号:{} 目标库位信息不存在", locNo);
            return;
        }
        if (!locMast.getLocSts().equals("O")) {
            News.taskInfo(wrkMast.getWrkNo(), "库位号:{} 目标库位状态不处于空库位", locNo);
            return;
        }

        FindCrnNoResult findCrnNoResult = commonService.findCrnNoByLocNo(locNo);
        if (findCrnNoResult == null) {
            News.taskInfo(wrkMast.getWrkNo(), "{}工作,未匹配到堆垛机", wrkMast.getWrkNo());
            return;
        }
        Integer crnNo = findCrnNoResult.getCrnNo();

        Integer targetStationId = commonService.findInStationId(findCrnNoResult, stationProtocol.getStationId());
        if (targetStationId == null) {
            News.taskInfo(wrkMast.getWrkNo(), "{}站点,搜索入库站点失败", stationProtocol.getStationId());
            return;
        }

        StationCommand command = stationThread.getCommand(StationCommandType.MOVE, wrkMast.getWrkNo(), stationProtocol.getStationId(), targetStationId, 0);
        if (command == null) {
            News.taskInfo(wrkMast.getWrkNo(), "{}工作,获取输送线命令失败", wrkMast.getWrkNo());
            return;
        }

        sourceLocMast.setLocSts("O");
        sourceLocMast.setModiTime(new Date());
        locMastService.updateById(sourceLocMast);

        locMast.setLocSts("S");
        locMast.setModiTime(new Date());
        locMastService.updateById(locMast);

        wrkMast.setLocNo(locNo);
        wrkMast.setStaNo(targetStationId);

        if (findCrnNoResult.getCrnType().equals(SlaveType.Crn)) {
            wrkMast.setCrnNo(crnNo);
        } else if (findCrnNoResult.getCrnType().equals(SlaveType.DualCrn)) {
            wrkMast.setDualCrnNo(crnNo);
        } else {
            throw new CoolException("未知设备类型");
        }

        if (!wrkMastService.updateById(wrkMast)) {
            return;
        }
        boolean offered = offerDevpCommandWithDedup(basDevp.getDevpNo(), command, "checkStationRunBlock_direct");
        if (!offered) {
            return;
        }
        syncOutOrderWatchState(wrkMast, stationProtocol.getStationId(), outOrderList, dispatchDecision, command);
        if (stationMoveCoordinator != null) {
            stationMoveCoordinator.recordDispatch(
                    wrkMast.getWrkNo(),
                    stationProtocol.getStationId(),
                    "checkStationIdleRecover",
                    "checkStationRunBlock_direct",
                    command,
                    dispatchDecision != null && dispatchDecision.isCircle()
                    false
            );
        }
        saveStationTaskIdleTrack(new StationTaskIdleTrack(wrkMast.getWrkNo(), stationProtocol.getStationId(), System.currentTimeMillis()));
        News.info("输送站点任务停留{}秒未运行，已重新计算路径并重启运行，站点号={}，目标站={}，工作号={}，清理旧分段命令数={}，命令数据={}",
                STATION_IDLE_RECOVER_SECONDS, stationProtocol.getStationId(), moveStaNo, wrkMast.getWrkNo(), clearedCommandCount, JSON.toJSONString(command));
    }

    private boolean canRecoverIdleStationTask(WrkMast wrkMast, Integer currentStationId) {

 src/test/java/com/zy/core/utils/StationOperateProcessUtilsReroutePipelineTest.java

@@ -2,11 +2,16 @@

import com.zy.asrs.entity.BasDevp;
import com.zy.asrs.entity.WrkMast;
import com.zy.core.enums.WrkIoType;
import com.zy.core.move.StationMoveCoordinator;
import com.zy.core.move.StationMoveSession;
import com.zy.core.model.command.StationCommand;
import com.zy.core.model.protocol.StationProtocol;
import com.zy.core.model.protocol.StationTaskBufferItem;
import com.zy.core.thread.StationThread;
import com.zy.common.utils.RedisUtil;
import org.junit.jupiter.api.Test;
import org.springframework.test.util.ReflectionTestUtils;

import java.util.Collections;
import java.util.List;
@@ -15,6 +20,7 @@
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

@@ -137,6 +143,61 @@
        assertEquals(20, watchCircleDecision.targetStationId());
    }

    @Test
    void runBlockReroute_keepsDirectReassignAndNormalRerouteSeparate() {
        StationOperateProcessUtils utils = new StationOperateProcessUtils();
        WrkMast inboundWrkMast = buildWrkMast(100, 20);
        inboundWrkMast.setIoType(WrkIoType.IN.id);

        assertTrue(utils.shouldUseRunBlockDirectReassign(inboundWrkMast, 10, List.of(10)));
        assertTrue(!utils.shouldUseRunBlockDirectReassign(inboundWrkMast, 11, List.of(10)));
    }

    @Test
    void idleRecover_skipsWhenLastDispatchIsTooRecent() {
        StationOperateProcessUtils utils = new StationOperateProcessUtils();
        StationMoveCoordinator coordinator = mock(StationMoveCoordinator.class);
        RedisUtil redisUtil = mock(RedisUtil.class);
        ReflectionTestUtils.setField(utils, "stationMoveCoordinator", coordinator);
        ReflectionTestUtils.setField(utils, "redisUtil", redisUtil);

        StationMoveSession session = new StationMoveSession();
        session.setStatus(StationMoveSession.STATUS_RUNNING);
        session.setCurrentStationId(10);
        session.setDispatchStationId(10);
        session.setLastIssuedAt(System.currentTimeMillis());
        when(coordinator.loadSession(100)).thenReturn(session);

        StationCommand command = new StationCommand();
        command.setTaskNo(100);
        command.setStationId(10);
        command.setTargetStaNo(20);

        StationOperateProcessUtils.RerouteContext context = StationOperateProcessUtils.RerouteContext.create(
                StationOperateProcessUtils.RerouteSceneType.IDLE_RECOVER,
                buildBasDevp(1),
                mock(StationThread.class),
                buildStationProtocol(10, 100, 10),
                buildWrkMast(100, 20),
                Collections.emptyList(),
                0.0d,
                "checkStationIdleRecover"
        ).withRecentDispatchGuard();

        StationOperateProcessUtils.RerouteExecutionResult result = utils.executeReroutePlan(
                context,
                StationOperateProcessUtils.RerouteCommandPlan.dispatch(
                        command,
                        StationOperateProcessUtils.RerouteDecision.proceed(20),
                        "checkStationIdleRecover"
                )
        );

        assertTrue(result.skipped());
        assertEquals("recent-dispatch", result.skipReason());
        verify(coordinator, never()).cancelSession(100);
    }

    private static BasDevp buildBasDevp(int devpNo) {
        BasDevp basDevp = new BasDevp();
        basDevp.setDevpNo(devpNo);