@Test public void handleRerouteInactiveAffectedFlows() { FlowPathRepository pathRepository = mock(FlowPathRepository.class); when(pathRepository.findInactiveBySegmentSwitch(regularFlow.getSrcSwitchId())) .thenReturn(Arrays.asList(regularFlow.getForwardPath(), regularFlow.getReversePath())); RepositoryFactory repositoryFactory = mock(RepositoryFactory.class); when(repositoryFactory.createFlowPathRepository()) .thenReturn(pathRepository); PersistenceManager persistenceManager = mock(PersistenceManager.class); when(persistenceManager.getRepositoryFactory()).thenReturn(repositoryFactory); when(persistenceManager.getTransactionManager()).thenReturn(transactionManager); RerouteService rerouteService = new RerouteService(persistenceManager); regularFlow.setStatus(FlowStatus.DOWN); rerouteService.rerouteInactiveAffectedFlows(carrier, CORRELATION_ID, regularFlow.getSrcSwitchId()); FlowThrottlingData expected = FlowThrottlingData.builder() .correlationId(CORRELATION_ID) .priority(regularFlow.getPriority()) .timeCreate(regularFlow.getTimeCreate()) .affectedIsl(Collections.emptySet()) .force(false) .effectivelyDown(true) .reason(format("Switch '%s' online", regularFlow.getSrcSwitchId())) .build(); verify(carrier).emitRerouteCommand(eq(regularFlow.getFlowId()), eq(expected)); regularFlow.setStatus(FlowStatus.UP); }
public void rerouteInactiveAffectedFlows(MessageSender sender, String correlationId, SwitchId switchId) { Set<Flow> flowsForRerouting = getAffectedInactiveFlowsForRerouting(switchId); for (Flow flow : flowsForRerouting) { if (flow.isPinned()) { log.info("Skipping reroute command for pinned flow {}", flow.getFlowId()); } else { log.info("Produce reroute (attempt to restore inactive flow) request for {} (switch online {})", flow.getFlowId(), switchId); FlowThrottlingData flowThrottlingData = getFlowThrottlingDataBuilder(flow) .correlationId(correlationId) .affectedIsl(Collections.emptySet()) .force(false) .effectivelyDown(true) .reason(format("Switch '%s' online", switchId)) .build(); sender.emitRerouteCommand(flow.getFlowId(), flowThrottlingData); } } }
RerouteService { public void rerouteInactiveAffectedFlows(MessageSender sender, String correlationId, SwitchId switchId) { Set<Flow> flowsForRerouting = getAffectedInactiveFlowsForRerouting(switchId); for (Flow flow : flowsForRerouting) { if (flow.isPinned()) { log.info("Skipping reroute command for pinned flow {}", flow.getFlowId()); } else { log.info("Produce reroute (attempt to restore inactive flow) request for {} (switch online {})", flow.getFlowId(), switchId); FlowThrottlingData flowThrottlingData = getFlowThrottlingDataBuilder(flow) .correlationId(correlationId) .affectedIsl(Collections.emptySet()) .force(false) .effectivelyDown(true) .reason(format("Switch '%s' online", switchId)) .build(); sender.emitRerouteCommand(flow.getFlowId(), flowThrottlingData); } } } }
RerouteService { public void rerouteInactiveAffectedFlows(MessageSender sender, String correlationId, SwitchId switchId) { Set<Flow> flowsForRerouting = getAffectedInactiveFlowsForRerouting(switchId); for (Flow flow : flowsForRerouting) { if (flow.isPinned()) { log.info("Skipping reroute command for pinned flow {}", flow.getFlowId()); } else { log.info("Produce reroute (attempt to restore inactive flow) request for {} (switch online {})", flow.getFlowId(), switchId); FlowThrottlingData flowThrottlingData = getFlowThrottlingDataBuilder(flow) .correlationId(correlationId) .affectedIsl(Collections.emptySet()) .force(false) .effectivelyDown(true) .reason(format("Switch '%s' online", switchId)) .build(); sender.emitRerouteCommand(flow.getFlowId(), flowThrottlingData); } } } RerouteService(PersistenceManager persistenceManager); }
RerouteService { public void rerouteInactiveAffectedFlows(MessageSender sender, String correlationId, SwitchId switchId) { Set<Flow> flowsForRerouting = getAffectedInactiveFlowsForRerouting(switchId); for (Flow flow : flowsForRerouting) { if (flow.isPinned()) { log.info("Skipping reroute command for pinned flow {}", flow.getFlowId()); } else { log.info("Produce reroute (attempt to restore inactive flow) request for {} (switch online {})", flow.getFlowId(), switchId); FlowThrottlingData flowThrottlingData = getFlowThrottlingDataBuilder(flow) .correlationId(correlationId) .affectedIsl(Collections.emptySet()) .force(false) .effectivelyDown(true) .reason(format("Switch '%s' online", switchId)) .build(); sender.emitRerouteCommand(flow.getFlowId(), flowThrottlingData); } } } 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 rerouteInactiveAffectedFlows(MessageSender sender, String correlationId, SwitchId switchId) { Set<Flow> flowsForRerouting = getAffectedInactiveFlowsForRerouting(switchId); for (Flow flow : flowsForRerouting) { if (flow.isPinned()) { log.info("Skipping reroute command for pinned flow {}", flow.getFlowId()); } else { log.info("Produce reroute (attempt to restore inactive flow) request for {} (switch online {})", flow.getFlowId(), switchId); FlowThrottlingData flowThrottlingData = getFlowThrottlingDataBuilder(flow) .correlationId(correlationId) .affectedIsl(Collections.emptySet()) .force(false) .effectivelyDown(true) .reason(format("Switch '%s' online", switchId)) .build(); sender.emitRerouteCommand(flow.getFlowId(), flowThrottlingData); } } } 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);  }