@Test public void shouldMergeStatefulRegions () { final OperatorDef stateful0 = OperatorDefBuilder.newInstance( "stateful0", StatefulOperator0.class ).build(); final OperatorDef stateful1 = OperatorDefBuilder.newInstance( "stateful1", StatefulOperator2.class ).build(); final OperatorDef stateful2 = OperatorDefBuilder.newInstance( "stateful2", StatefulOperator2.class ).build(); final FlowDef flow = new FlowDefBuilder().add( stateful0 ) .add( stateful1 ) .add( stateful2 ) .connect( "stateful0", "stateful1" ) .connect( "stateful1", "stateful2" ) .build(); final List<RegionDef> regions = regionDefFormer.createRegions( flow ); flowOptimizer.mergeRegions( flow.getOperatorsMap(), flow.getConnections(), regions ); assertEquals( 2, regions.size() ); final RegionDef region = regions.get( 1 ); assertEquals( STATEFUL, region.getRegionType() ); assertEquals( asList( stateful1, stateful2 ), region.getOperators() ); }
void mergeRegions ( final Map<String, OperatorDef> operators, final Set<Entry<Port, Port>> connections, final List<RegionDef> regions ) { int before, after = regions.size(); do { before = after; attemptToMergeRegions( operators, connections, regions ); after = regions.size(); } while ( before != after ); }
FlowDefOptimizerImpl implements FlowDefOptimizer { void mergeRegions ( final Map<String, OperatorDef> operators, final Set<Entry<Port, Port>> connections, final List<RegionDef> regions ) { int before, after = regions.size(); do { before = after; attemptToMergeRegions( operators, connections, regions ); after = regions.size(); } while ( before != after ); } }
FlowDefOptimizerImpl implements FlowDefOptimizer { void mergeRegions ( final Map<String, OperatorDef> operators, final Set<Entry<Port, Port>> connections, final List<RegionDef> regions ) { int before, after = regions.size(); do { before = after; attemptToMergeRegions( operators, connections, regions ); after = regions.size(); } while ( before != after ); } @Inject  FlowDefOptimizerImpl( final JokerConfig jokerConfig, final IdGenerator idGenerator ); }
FlowDefOptimizerImpl implements FlowDefOptimizer { void mergeRegions ( final Map<String, OperatorDef> operators, final Set<Entry<Port, Port>> connections, final List<RegionDef> regions ) { int before, after = regions.size(); do { before = after; attemptToMergeRegions( operators, connections, regions ); after = regions.size(); } while ( before != after ); } @Inject  FlowDefOptimizerImpl( final JokerConfig jokerConfig, final IdGenerator idGenerator ); @Override Pair<FlowDef, List<RegionDef>> optimize( final FlowDef flow, final List<RegionDef> regions ); }
FlowDefOptimizerImpl implements FlowDefOptimizer { void mergeRegions ( final Map<String, OperatorDef> operators, final Set<Entry<Port, Port>> connections, final List<RegionDef> regions ) { int before, after = regions.size(); do { before = after; attemptToMergeRegions( operators, connections, regions ); after = regions.size(); } while ( before != after ); } @Inject  FlowDefOptimizerImpl( final JokerConfig jokerConfig, final IdGenerator idGenerator ); @Override Pair<FlowDef, List<RegionDef>> optimize( final FlowDef flow, final List<RegionDef> regions );  }