@Test public void shouldMergeStatelessAndStatefulRegions () { final OperatorDef stateless0 = OperatorDefBuilder.newInstance( "stateless0", StatelessOperator0.class ).build(); final OperatorDef stateless1 = OperatorDefBuilder.newInstance( "stateless1", StatelessOperator1.class ).build(); final OperatorDef stateful = OperatorDefBuilder.newInstance( "stateful", StatefulOperator2.class ).build(); final FlowDef flow = new FlowDefBuilder().add( stateless0 ) .add( stateless1 ) .add( stateful ) .connect( "stateless0", "stateless1" ) .connect( "stateless1", "stateful" ) .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( stateless1, stateful ), 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 );  }