@Test public void shouldInitializeWithMapper () { final BiConsumer<Tuple, Tuple> mapper = ( input, output ) -> input.sinkTo( output::set ); config.set( MAPPER_CONFIG_PARAMETER, mapper ); final SchedulingStrategy strategy = operator.init( initCtx ); assertScheduleWhenTuplesAvailableStrategy( strategy, 1 ); }
@Override public SchedulingStrategy init ( final InitCtx ctx ) { final OperatorConfig config = ctx.getConfig(); this.mapper = config.getOrFail( MAPPER_CONFIG_PARAMETER ); this.outputSchema = ctx.getOutputPortSchema( 0 ); return scheduleWhenTuplesAvailableOnDefaultPort( DEFAULT_TUPLE_COUNT_CONFIG_VALUE ); }
MapperOperator implements Operator { @Override public SchedulingStrategy init ( final InitCtx ctx ) { final OperatorConfig config = ctx.getConfig(); this.mapper = config.getOrFail( MAPPER_CONFIG_PARAMETER ); this.outputSchema = ctx.getOutputPortSchema( 0 ); return scheduleWhenTuplesAvailableOnDefaultPort( DEFAULT_TUPLE_COUNT_CONFIG_VALUE ); } }
MapperOperator implements Operator { @Override public SchedulingStrategy init ( final InitCtx ctx ) { final OperatorConfig config = ctx.getConfig(); this.mapper = config.getOrFail( MAPPER_CONFIG_PARAMETER ); this.outputSchema = ctx.getOutputPortSchema( 0 ); return scheduleWhenTuplesAvailableOnDefaultPort( DEFAULT_TUPLE_COUNT_CONFIG_VALUE ); }  }
MapperOperator implements Operator { @Override public SchedulingStrategy init ( final InitCtx ctx ) { final OperatorConfig config = ctx.getConfig(); this.mapper = config.getOrFail( MAPPER_CONFIG_PARAMETER ); this.outputSchema = ctx.getOutputPortSchema( 0 ); return scheduleWhenTuplesAvailableOnDefaultPort( DEFAULT_TUPLE_COUNT_CONFIG_VALUE ); }  @Override SchedulingStrategy init( final InitCtx ctx ); @Override void invoke( final InvocationCtx ctx ); }
MapperOperator implements Operator { @Override public SchedulingStrategy init ( final InitCtx ctx ) { final OperatorConfig config = ctx.getConfig(); this.mapper = config.getOrFail( MAPPER_CONFIG_PARAMETER ); this.outputSchema = ctx.getOutputPortSchema( 0 ); return scheduleWhenTuplesAvailableOnDefaultPort( DEFAULT_TUPLE_COUNT_CONFIG_VALUE ); }  @Override SchedulingStrategy init( final InitCtx ctx ); @Override void invoke( final InvocationCtx ctx ); static final String MAPPER_CONFIG_PARAMETER; }