@Test( expected = IllegalArgumentException.class ) public void shouldFailToInitWithoutAccumulatorInitializer () { config.set( TUPLE_COUNT_CONFIG_PARAMETER, tupleCount ).set( REDUCER_CONFIG_PARAMETER, adder ); operator.init( initCtx ); }
@Override public SchedulingStrategy init ( final InitCtx ctx ) { final OperatorConfig config = ctx.getConfig(); this.reducer = config.getOrFail( REDUCER_CONFIG_PARAMETER ); this.tupleCount = config.getOrFail( TUPLE_COUNT_CONFIG_PARAMETER ); this.accumulatorInitializer = config.getOrFail( ACCUMULATOR_INITIALIZER_CONFIG_PARAMETER ); this.accumulatorSupplier = () -> { final Tuple accumulator = new Tuple( outputSchema ); accumulatorInitializer.accept( accumulator ); return accumulator; }; this.outputSchema = ctx.getOutputPortSchema( 0 ); this.windowSchema = new PortRuntimeSchemaBuilder( EXACT_FIELD_SET, asList( new RuntimeSchemaField( TUPLE_COUNT_FIELD, Integer.class ), new RuntimeSchemaField( WINDOW_FIELD, Integer.class ) ) ).build(); return scheduleWhenTuplesAvailableOnDefaultPort( 1 ); }
TupleCountBasedWindowReducerOperator implements Operator { @Override public SchedulingStrategy init ( final InitCtx ctx ) { final OperatorConfig config = ctx.getConfig(); this.reducer = config.getOrFail( REDUCER_CONFIG_PARAMETER ); this.tupleCount = config.getOrFail( TUPLE_COUNT_CONFIG_PARAMETER ); this.accumulatorInitializer = config.getOrFail( ACCUMULATOR_INITIALIZER_CONFIG_PARAMETER ); this.accumulatorSupplier = () -> { final Tuple accumulator = new Tuple( outputSchema ); accumulatorInitializer.accept( accumulator ); return accumulator; }; this.outputSchema = ctx.getOutputPortSchema( 0 ); this.windowSchema = new PortRuntimeSchemaBuilder( EXACT_FIELD_SET, asList( new RuntimeSchemaField( TUPLE_COUNT_FIELD, Integer.class ), new RuntimeSchemaField( WINDOW_FIELD, Integer.class ) ) ).build(); return scheduleWhenTuplesAvailableOnDefaultPort( 1 ); } }
TupleCountBasedWindowReducerOperator implements Operator { @Override public SchedulingStrategy init ( final InitCtx ctx ) { final OperatorConfig config = ctx.getConfig(); this.reducer = config.getOrFail( REDUCER_CONFIG_PARAMETER ); this.tupleCount = config.getOrFail( TUPLE_COUNT_CONFIG_PARAMETER ); this.accumulatorInitializer = config.getOrFail( ACCUMULATOR_INITIALIZER_CONFIG_PARAMETER ); this.accumulatorSupplier = () -> { final Tuple accumulator = new Tuple( outputSchema ); accumulatorInitializer.accept( accumulator ); return accumulator; }; this.outputSchema = ctx.getOutputPortSchema( 0 ); this.windowSchema = new PortRuntimeSchemaBuilder( EXACT_FIELD_SET, asList( new RuntimeSchemaField( TUPLE_COUNT_FIELD, Integer.class ), new RuntimeSchemaField( WINDOW_FIELD, Integer.class ) ) ).build(); return scheduleWhenTuplesAvailableOnDefaultPort( 1 ); }  }
TupleCountBasedWindowReducerOperator implements Operator { @Override public SchedulingStrategy init ( final InitCtx ctx ) { final OperatorConfig config = ctx.getConfig(); this.reducer = config.getOrFail( REDUCER_CONFIG_PARAMETER ); this.tupleCount = config.getOrFail( TUPLE_COUNT_CONFIG_PARAMETER ); this.accumulatorInitializer = config.getOrFail( ACCUMULATOR_INITIALIZER_CONFIG_PARAMETER ); this.accumulatorSupplier = () -> { final Tuple accumulator = new Tuple( outputSchema ); accumulatorInitializer.accept( accumulator ); return accumulator; }; this.outputSchema = ctx.getOutputPortSchema( 0 ); this.windowSchema = new PortRuntimeSchemaBuilder( EXACT_FIELD_SET, asList( new RuntimeSchemaField( TUPLE_COUNT_FIELD, Integer.class ), new RuntimeSchemaField( WINDOW_FIELD, Integer.class ) ) ).build(); return scheduleWhenTuplesAvailableOnDefaultPort( 1 ); }  @Override SchedulingStrategy init( final InitCtx ctx ); @Override void invoke( final InvocationCtx ctx ); }
TupleCountBasedWindowReducerOperator implements Operator { @Override public SchedulingStrategy init ( final InitCtx ctx ) { final OperatorConfig config = ctx.getConfig(); this.reducer = config.getOrFail( REDUCER_CONFIG_PARAMETER ); this.tupleCount = config.getOrFail( TUPLE_COUNT_CONFIG_PARAMETER ); this.accumulatorInitializer = config.getOrFail( ACCUMULATOR_INITIALIZER_CONFIG_PARAMETER ); this.accumulatorSupplier = () -> { final Tuple accumulator = new Tuple( outputSchema ); accumulatorInitializer.accept( accumulator ); return accumulator; }; this.outputSchema = ctx.getOutputPortSchema( 0 ); this.windowSchema = new PortRuntimeSchemaBuilder( EXACT_FIELD_SET, asList( new RuntimeSchemaField( TUPLE_COUNT_FIELD, Integer.class ), new RuntimeSchemaField( WINDOW_FIELD, Integer.class ) ) ).build(); return scheduleWhenTuplesAvailableOnDefaultPort( 1 ); }  @Override SchedulingStrategy init( final InitCtx ctx ); @Override void invoke( final InvocationCtx ctx ); static final String TUPLE_COUNT_CONFIG_PARAMETER; static final String REDUCER_CONFIG_PARAMETER; static final String ACCUMULATOR_INITIALIZER_CONFIG_PARAMETER; static final String WINDOW_FIELD; }