@Test public void testStatefulRegion5Downstream () { final List<RegionDef> downstream = getWholeDownstream( flow, regions, statefulRegion5 ); assertThat( downstream, hasSize( 3 ) ); assertThat( downstream, hasItem( statefulRegion4 ) ); assertThat( downstream, hasItem( statefulRegion6 ) ); assertThat( downstream, hasItem( statelessRegion ) ); }
public static List<RegionDef> getWholeDownstream ( final FlowDef flowDef, final List<RegionDef> regions, final RegionDef upstream ) { final Function<String, RegionDef> regionDefByFirstOperatorId = operatorId -> getRegionByFirstOperator( regions, operatorId ); final Set<RegionDef> current = new HashSet<>(); current.add( upstream ); final List<RegionDef> downstream = new ArrayList<>(); final Predicate<RegionDef> nonVisitedDownstreamRegion = regionDef -> !downstream.contains( regionDef ); final Consumer<RegionDef> addToDownstream = regionDef -> { downstream.add( regionDef ); current.add( regionDef ); }; while ( current.size() > 0 ) { final Iterator<RegionDef> it = current.iterator(); final RegionDef region = it.next(); it.remove(); flowDef.getOutboundConnections( getLastOperator( region ).getId() ) .values() .stream() .flatMap( Set::stream ) .map( Port::getOperatorId ) .map( regionDefByFirstOperatorId ) .filter( nonVisitedDownstreamRegion ) .forEach( addToDownstream ); } downstream.sort( comparing( RegionDef::getRegionId ) ); return unmodifiableList( downstream ); }
RegionUtil { public static List<RegionDef> getWholeDownstream ( final FlowDef flowDef, final List<RegionDef> regions, final RegionDef upstream ) { final Function<String, RegionDef> regionDefByFirstOperatorId = operatorId -> getRegionByFirstOperator( regions, operatorId ); final Set<RegionDef> current = new HashSet<>(); current.add( upstream ); final List<RegionDef> downstream = new ArrayList<>(); final Predicate<RegionDef> nonVisitedDownstreamRegion = regionDef -> !downstream.contains( regionDef ); final Consumer<RegionDef> addToDownstream = regionDef -> { downstream.add( regionDef ); current.add( regionDef ); }; while ( current.size() > 0 ) { final Iterator<RegionDef> it = current.iterator(); final RegionDef region = it.next(); it.remove(); flowDef.getOutboundConnections( getLastOperator( region ).getId() ) .values() .stream() .flatMap( Set::stream ) .map( Port::getOperatorId ) .map( regionDefByFirstOperatorId ) .filter( nonVisitedDownstreamRegion ) .forEach( addToDownstream ); } downstream.sort( comparing( RegionDef::getRegionId ) ); return unmodifiableList( downstream ); } }
RegionUtil { public static List<RegionDef> getWholeDownstream ( final FlowDef flowDef, final List<RegionDef> regions, final RegionDef upstream ) { final Function<String, RegionDef> regionDefByFirstOperatorId = operatorId -> getRegionByFirstOperator( regions, operatorId ); final Set<RegionDef> current = new HashSet<>(); current.add( upstream ); final List<RegionDef> downstream = new ArrayList<>(); final Predicate<RegionDef> nonVisitedDownstreamRegion = regionDef -> !downstream.contains( regionDef ); final Consumer<RegionDef> addToDownstream = regionDef -> { downstream.add( regionDef ); current.add( regionDef ); }; while ( current.size() > 0 ) { final Iterator<RegionDef> it = current.iterator(); final RegionDef region = it.next(); it.remove(); flowDef.getOutboundConnections( getLastOperator( region ).getId() ) .values() .stream() .flatMap( Set::stream ) .map( Port::getOperatorId ) .map( regionDefByFirstOperatorId ) .filter( nonVisitedDownstreamRegion ) .forEach( addToDownstream ); } downstream.sort( comparing( RegionDef::getRegionId ) ); return unmodifiableList( downstream ); }  }
RegionUtil { public static List<RegionDef> getWholeDownstream ( final FlowDef flowDef, final List<RegionDef> regions, final RegionDef upstream ) { final Function<String, RegionDef> regionDefByFirstOperatorId = operatorId -> getRegionByFirstOperator( regions, operatorId ); final Set<RegionDef> current = new HashSet<>(); current.add( upstream ); final List<RegionDef> downstream = new ArrayList<>(); final Predicate<RegionDef> nonVisitedDownstreamRegion = regionDef -> !downstream.contains( regionDef ); final Consumer<RegionDef> addToDownstream = regionDef -> { downstream.add( regionDef ); current.add( regionDef ); }; while ( current.size() > 0 ) { final Iterator<RegionDef> it = current.iterator(); final RegionDef region = it.next(); it.remove(); flowDef.getOutboundConnections( getLastOperator( region ).getId() ) .values() .stream() .flatMap( Set::stream ) .map( Port::getOperatorId ) .map( regionDefByFirstOperatorId ) .filter( nonVisitedDownstreamRegion ) .forEach( addToDownstream ); } downstream.sort( comparing( RegionDef::getRegionId ) ); return unmodifiableList( downstream ); }  static List<RegionDef> sortTopologically( final Map<String, OperatorDef> operators,
                                                      final Collection<Entry<Port, Port>> connections,
                                                      final List<RegionDef> regions ); static RegionDef getRegionByLastOperator( final List<RegionDef> regions, final OperatorDef operator ); static RegionDef getRegionByFirstOperator( final List<RegionDef> regions, final String operatorId ); static OperatorDef getFirstOperator( final RegionDef regionDef ); static OperatorDef getLastOperator( final RegionDef regionDef ); static List<RegionDef> getWholeDownstream( final FlowDef flowDef, final List<RegionDef> regions, final RegionDef upstream ); static List<RegionDef> getLeftMostRegions( final FlowDef flowDef,
                                                       final List<RegionDef> allRegions,
                                                       final List<RegionDef> regionsToFilter ); }
RegionUtil { public static List<RegionDef> getWholeDownstream ( final FlowDef flowDef, final List<RegionDef> regions, final RegionDef upstream ) { final Function<String, RegionDef> regionDefByFirstOperatorId = operatorId -> getRegionByFirstOperator( regions, operatorId ); final Set<RegionDef> current = new HashSet<>(); current.add( upstream ); final List<RegionDef> downstream = new ArrayList<>(); final Predicate<RegionDef> nonVisitedDownstreamRegion = regionDef -> !downstream.contains( regionDef ); final Consumer<RegionDef> addToDownstream = regionDef -> { downstream.add( regionDef ); current.add( regionDef ); }; while ( current.size() > 0 ) { final Iterator<RegionDef> it = current.iterator(); final RegionDef region = it.next(); it.remove(); flowDef.getOutboundConnections( getLastOperator( region ).getId() ) .values() .stream() .flatMap( Set::stream ) .map( Port::getOperatorId ) .map( regionDefByFirstOperatorId ) .filter( nonVisitedDownstreamRegion ) .forEach( addToDownstream ); } downstream.sort( comparing( RegionDef::getRegionId ) ); return unmodifiableList( downstream ); }  static List<RegionDef> sortTopologically( final Map<String, OperatorDef> operators,
                                                      final Collection<Entry<Port, Port>> connections,
                                                      final List<RegionDef> regions ); static RegionDef getRegionByLastOperator( final List<RegionDef> regions, final OperatorDef operator ); static RegionDef getRegionByFirstOperator( final List<RegionDef> regions, final String operatorId ); static OperatorDef getFirstOperator( final RegionDef regionDef ); static OperatorDef getLastOperator( final RegionDef regionDef ); static List<RegionDef> getWholeDownstream( final FlowDef flowDef, final List<RegionDef> regions, final RegionDef upstream ); static List<RegionDef> getLeftMostRegions( final FlowDef flowDef,
                                                       final List<RegionDef> allRegions,
                                                       final List<RegionDef> regionsToFilter );  }