BarrierOperator implements Operator { @Override public SchedulingStrategy init ( final InitCtx ctx ) { final OperatorConfig config = ctx.getConfig(); this.inputPortCount = ctx.getInputPortCount(); this.mergePolicy = config.getOrFail( MERGE_POLICY_CONfIG_PARAMETER ); this.entryMerger = new EntryMerger( mergePolicy ); this.outputSchema = ctx.getOutputPortSchema( 0 ); final int[] inputPorts = IntStream.range( 0, inputPortCount ).toArray(); return scheduleWhenTuplesAvailableOnAll( AT_LEAST_BUT_SAME_ON_ALL_PORTS, ctx.getInputPortCount(), 1, inputPorts ); }  @Override SchedulingStrategy init( final InitCtx ctx ); @Override void invoke( final InvocationCtx ctx ); static final String MERGE_POLICY_CONfIG_PARAMETER; }
@Test( expected = IllegalArgumentException.class ) public void shouldFailWithNoMergePolicy () { operator.init( initCtx ); }
@Test public void shouldScheduleOnGivenInputPorts () { config.set( MERGE_POLICY_CONfIG_PARAMETER, KEEP_EXISTING_VALUE ); final SchedulingStrategy initialStrategy = operator.init( initCtx ); assertSchedulingStrategy( initialStrategy ); }