@Test public void shouldSkipRerouteRequestsForFlowWithoutAffectedPathSegment() { PathNode islSide = new PathNode(SWITCH_A.getSwitchId(), 1, 0); FlowPathRepository pathRepository = mock(FlowPathRepository.class); when(pathRepository.findBySegmentEndpoint(eq(islSide.getSwitchId()), eq(islSide.getPortNo()))) .thenReturn(Arrays.asList(regularFlow.getForwardPath(), regularFlow.getReversePath())); FlowRepository flowRepository = mock(FlowRepository.class); PathSegmentRepository pathSegmentRepository = mock(PathSegmentRepository.class); doThrow(new EntityNotFoundException("Not found")) .when(pathSegmentRepository).updateFailedStatus(any(), any(), anyBoolean()); RepositoryFactory repositoryFactory = mock(RepositoryFactory.class); when(repositoryFactory.createPathSegmentRepository()) .thenReturn(pathSegmentRepository); when(repositoryFactory.createFlowPathRepository()) .thenReturn(pathRepository); when(repositoryFactory.createFlowRepository()) .thenReturn(flowRepository); PersistenceManager persistenceManager = mock(PersistenceManager.class); when(persistenceManager.getRepositoryFactory()).thenReturn(repositoryFactory); when(persistenceManager.getTransactionManager()).thenReturn(transactionManager); RerouteService rerouteService = new RerouteService(persistenceManager); RerouteAffectedFlows request = new RerouteAffectedFlows(islSide, "dummy-reason - unittest"); rerouteService.rerouteAffectedFlows(carrier, CORRELATION_ID, request); verifyZeroInteractions(carrier); }
public void rerouteAffectedFlows(MessageSender sender, String correlationId, RerouteAffectedFlows command) { PathNode pathNode = command.getPathNode(); int port = pathNode.getPortNo(); SwitchId switchId = pathNode.getSwitchId(); final IslEndpoint affectedIsl = new IslEndpoint(switchId, port); transactionManager.doInTransaction(() -> { Collection<FlowPath> affectedFlowPaths = getAffectedFlowPaths(pathNode.getSwitchId(), pathNode.getPortNo()); List<FlowPath> pathsForSwapping = getPathsForSwapping(affectedFlowPaths); for (FlowPath path : pathsForSwapping) { sender.emitPathSwapCommand(correlationId, path, command.getReason()); } for (FlowWithAffectedPaths entry : groupPathsForRerouting(affectedFlowPaths)) { Flow flow = entry.getFlow(); boolean flowPathFound = updateFlowPathsStateForFlow(switchId, port, entry.getAffectedPaths()); FlowStatus flowStatus = flow.computeFlowStatus(); String flowStatusInfo = null; if (!FlowStatus.UP.equals(flowStatus)) { flowStatusInfo = command.getReason(); } flowRepository.updateStatusSafe(flow, flowStatus, flowStatusInfo); if (flowPathFound) { FlowThrottlingData flowThrottlingData = getFlowThrottlingDataBuilder(flow) .correlationId(correlationId) .affectedIsl(Collections.singleton(affectedIsl)) .force(false) .effectivelyDown(true) .reason(command.getReason()) .build(); sender.emitRerouteCommand(flow.getFlowId(), flowThrottlingData); } } Set<Flow> affectedPinnedFlows = groupAffectedPinnedFlows(affectedFlowPaths); for (Flow flow : affectedPinnedFlows) { List<FlowPath> flowPaths = new ArrayList<>(flow.getPaths()); updateFlowPathsStateForFlow(switchId, port, flowPaths); if (flow.getStatus() != FlowStatus.DOWN) { flowDashboardLogger.onFlowStatusUpdate(flow.getFlowId(), FlowStatus.DOWN); flowRepository.updateStatusSafe(flow, FlowStatus.DOWN, command.getReason()); } } }); }
RerouteService { public void rerouteAffectedFlows(MessageSender sender, String correlationId, RerouteAffectedFlows command) { PathNode pathNode = command.getPathNode(); int port = pathNode.getPortNo(); SwitchId switchId = pathNode.getSwitchId(); final IslEndpoint affectedIsl = new IslEndpoint(switchId, port); transactionManager.doInTransaction(() -> { Collection<FlowPath> affectedFlowPaths = getAffectedFlowPaths(pathNode.getSwitchId(), pathNode.getPortNo()); List<FlowPath> pathsForSwapping = getPathsForSwapping(affectedFlowPaths); for (FlowPath path : pathsForSwapping) { sender.emitPathSwapCommand(correlationId, path, command.getReason()); } for (FlowWithAffectedPaths entry : groupPathsForRerouting(affectedFlowPaths)) { Flow flow = entry.getFlow(); boolean flowPathFound = updateFlowPathsStateForFlow(switchId, port, entry.getAffectedPaths()); FlowStatus flowStatus = flow.computeFlowStatus(); String flowStatusInfo = null; if (!FlowStatus.UP.equals(flowStatus)) { flowStatusInfo = command.getReason(); } flowRepository.updateStatusSafe(flow, flowStatus, flowStatusInfo); if (flowPathFound) { FlowThrottlingData flowThrottlingData = getFlowThrottlingDataBuilder(flow) .correlationId(correlationId) .affectedIsl(Collections.singleton(affectedIsl)) .force(false) .effectivelyDown(true) .reason(command.getReason()) .build(); sender.emitRerouteCommand(flow.getFlowId(), flowThrottlingData); } } Set<Flow> affectedPinnedFlows = groupAffectedPinnedFlows(affectedFlowPaths); for (Flow flow : affectedPinnedFlows) { List<FlowPath> flowPaths = new ArrayList<>(flow.getPaths()); updateFlowPathsStateForFlow(switchId, port, flowPaths); if (flow.getStatus() != FlowStatus.DOWN) { flowDashboardLogger.onFlowStatusUpdate(flow.getFlowId(), FlowStatus.DOWN); flowRepository.updateStatusSafe(flow, FlowStatus.DOWN, command.getReason()); } } }); } }
RerouteService { public void rerouteAffectedFlows(MessageSender sender, String correlationId, RerouteAffectedFlows command) { PathNode pathNode = command.getPathNode(); int port = pathNode.getPortNo(); SwitchId switchId = pathNode.getSwitchId(); final IslEndpoint affectedIsl = new IslEndpoint(switchId, port); transactionManager.doInTransaction(() -> { Collection<FlowPath> affectedFlowPaths = getAffectedFlowPaths(pathNode.getSwitchId(), pathNode.getPortNo()); List<FlowPath> pathsForSwapping = getPathsForSwapping(affectedFlowPaths); for (FlowPath path : pathsForSwapping) { sender.emitPathSwapCommand(correlationId, path, command.getReason()); } for (FlowWithAffectedPaths entry : groupPathsForRerouting(affectedFlowPaths)) { Flow flow = entry.getFlow(); boolean flowPathFound = updateFlowPathsStateForFlow(switchId, port, entry.getAffectedPaths()); FlowStatus flowStatus = flow.computeFlowStatus(); String flowStatusInfo = null; if (!FlowStatus.UP.equals(flowStatus)) { flowStatusInfo = command.getReason(); } flowRepository.updateStatusSafe(flow, flowStatus, flowStatusInfo); if (flowPathFound) { FlowThrottlingData flowThrottlingData = getFlowThrottlingDataBuilder(flow) .correlationId(correlationId) .affectedIsl(Collections.singleton(affectedIsl)) .force(false) .effectivelyDown(true) .reason(command.getReason()) .build(); sender.emitRerouteCommand(flow.getFlowId(), flowThrottlingData); } } Set<Flow> affectedPinnedFlows = groupAffectedPinnedFlows(affectedFlowPaths); for (Flow flow : affectedPinnedFlows) { List<FlowPath> flowPaths = new ArrayList<>(flow.getPaths()); updateFlowPathsStateForFlow(switchId, port, flowPaths); if (flow.getStatus() != FlowStatus.DOWN) { flowDashboardLogger.onFlowStatusUpdate(flow.getFlowId(), FlowStatus.DOWN); flowRepository.updateStatusSafe(flow, FlowStatus.DOWN, command.getReason()); } } }); } RerouteService(PersistenceManager persistenceManager); }
RerouteService { public void rerouteAffectedFlows(MessageSender sender, String correlationId, RerouteAffectedFlows command) { PathNode pathNode = command.getPathNode(); int port = pathNode.getPortNo(); SwitchId switchId = pathNode.getSwitchId(); final IslEndpoint affectedIsl = new IslEndpoint(switchId, port); transactionManager.doInTransaction(() -> { Collection<FlowPath> affectedFlowPaths = getAffectedFlowPaths(pathNode.getSwitchId(), pathNode.getPortNo()); List<FlowPath> pathsForSwapping = getPathsForSwapping(affectedFlowPaths); for (FlowPath path : pathsForSwapping) { sender.emitPathSwapCommand(correlationId, path, command.getReason()); } for (FlowWithAffectedPaths entry : groupPathsForRerouting(affectedFlowPaths)) { Flow flow = entry.getFlow(); boolean flowPathFound = updateFlowPathsStateForFlow(switchId, port, entry.getAffectedPaths()); FlowStatus flowStatus = flow.computeFlowStatus(); String flowStatusInfo = null; if (!FlowStatus.UP.equals(flowStatus)) { flowStatusInfo = command.getReason(); } flowRepository.updateStatusSafe(flow, flowStatus, flowStatusInfo); if (flowPathFound) { FlowThrottlingData flowThrottlingData = getFlowThrottlingDataBuilder(flow) .correlationId(correlationId) .affectedIsl(Collections.singleton(affectedIsl)) .force(false) .effectivelyDown(true) .reason(command.getReason()) .build(); sender.emitRerouteCommand(flow.getFlowId(), flowThrottlingData); } } Set<Flow> affectedPinnedFlows = groupAffectedPinnedFlows(affectedFlowPaths); for (Flow flow : affectedPinnedFlows) { List<FlowPath> flowPaths = new ArrayList<>(flow.getPaths()); updateFlowPathsStateForFlow(switchId, port, flowPaths); if (flow.getStatus() != FlowStatus.DOWN) { flowDashboardLogger.onFlowStatusUpdate(flow.getFlowId(), FlowStatus.DOWN); flowRepository.updateStatusSafe(flow, FlowStatus.DOWN, command.getReason()); } } }); } RerouteService(PersistenceManager persistenceManager); void rerouteAffectedFlows(MessageSender sender, String correlationId, RerouteAffectedFlows command); void rerouteInactiveAffectedFlows(MessageSender sender, String correlationId,
                                             SwitchId switchId); void rerouteInactiveFlows(MessageSender sender, String correlationId, RerouteInactiveFlows command); Collection<FlowPath> getAffectedFlowPaths(SwitchId switchId, int port); List<FlowPath> getPathsForSwapping(Collection<FlowPath> paths); List<FlowWithAffectedPaths> groupPathsForRerouting(Collection<FlowPath> paths); Set<Flow> groupAffectedPinnedFlows(Collection<FlowPath> paths); Map<Flow, Set<PathId>> getInactiveFlowsForRerouting(); Set<Flow> getAffectedInactiveFlowsForRerouting(SwitchId switchId); void processManualRerouteRequest(MessageSender sender, String correlationId, FlowRerouteRequest request); void processSingleSwitchFlowStatusUpdate(SwitchStateChanged request); }
RerouteService { public void rerouteAffectedFlows(MessageSender sender, String correlationId, RerouteAffectedFlows command) { PathNode pathNode = command.getPathNode(); int port = pathNode.getPortNo(); SwitchId switchId = pathNode.getSwitchId(); final IslEndpoint affectedIsl = new IslEndpoint(switchId, port); transactionManager.doInTransaction(() -> { Collection<FlowPath> affectedFlowPaths = getAffectedFlowPaths(pathNode.getSwitchId(), pathNode.getPortNo()); List<FlowPath> pathsForSwapping = getPathsForSwapping(affectedFlowPaths); for (FlowPath path : pathsForSwapping) { sender.emitPathSwapCommand(correlationId, path, command.getReason()); } for (FlowWithAffectedPaths entry : groupPathsForRerouting(affectedFlowPaths)) { Flow flow = entry.getFlow(); boolean flowPathFound = updateFlowPathsStateForFlow(switchId, port, entry.getAffectedPaths()); FlowStatus flowStatus = flow.computeFlowStatus(); String flowStatusInfo = null; if (!FlowStatus.UP.equals(flowStatus)) { flowStatusInfo = command.getReason(); } flowRepository.updateStatusSafe(flow, flowStatus, flowStatusInfo); if (flowPathFound) { FlowThrottlingData flowThrottlingData = getFlowThrottlingDataBuilder(flow) .correlationId(correlationId) .affectedIsl(Collections.singleton(affectedIsl)) .force(false) .effectivelyDown(true) .reason(command.getReason()) .build(); sender.emitRerouteCommand(flow.getFlowId(), flowThrottlingData); } } Set<Flow> affectedPinnedFlows = groupAffectedPinnedFlows(affectedFlowPaths); for (Flow flow : affectedPinnedFlows) { List<FlowPath> flowPaths = new ArrayList<>(flow.getPaths()); updateFlowPathsStateForFlow(switchId, port, flowPaths); if (flow.getStatus() != FlowStatus.DOWN) { flowDashboardLogger.onFlowStatusUpdate(flow.getFlowId(), FlowStatus.DOWN); flowRepository.updateStatusSafe(flow, FlowStatus.DOWN, command.getReason()); } } }); } RerouteService(PersistenceManager persistenceManager); void rerouteAffectedFlows(MessageSender sender, String correlationId, RerouteAffectedFlows command); void rerouteInactiveAffectedFlows(MessageSender sender, String correlationId,
                                             SwitchId switchId); void rerouteInactiveFlows(MessageSender sender, String correlationId, RerouteInactiveFlows command); Collection<FlowPath> getAffectedFlowPaths(SwitchId switchId, int port); List<FlowPath> getPathsForSwapping(Collection<FlowPath> paths); List<FlowWithAffectedPaths> groupPathsForRerouting(Collection<FlowPath> paths); Set<Flow> groupAffectedPinnedFlows(Collection<FlowPath> paths); Map<Flow, Set<PathId>> getInactiveFlowsForRerouting(); Set<Flow> getAffectedInactiveFlowsForRerouting(SwitchId switchId); void processManualRerouteRequest(MessageSender sender, String correlationId, FlowRerouteRequest request); void processSingleSwitchFlowStatusUpdate(SwitchStateChanged request);  }