@Test public void shouldNotProduceOutputBeforeFirstWindowCompletes () { configure(); addInputTuple( 5, 20, 1 ); addInputTuple( 10, 25, 2 ); operator.invoke( invocationCtx ); assertThat( output.getTupleCount( 0 ), equalTo( 0 ) ); assertWindow( 2, new double[] { 5, 10, 0 }, new double[] { 20, 25, 0 }, 15, 45 ); }
@Override public void invoke ( final InvocationCtx ctx ) { final KVStore kvStore = ctx.getKVStore(); final Tuple currentWindow = kvStore.getOrDefault( WINDOW_KEY, this::createWindowTuple ); final double[] vwapValues = currentWindow.get( VWAPS_FIELD ); final double[] volumeValues = currentWindow.get( VOLUMES_FIELD ); double vwapSum = currentWindow.get( SINGLE_VWAP_FIELD ); double volumeSum = currentWindow.get( SINGLE_VOLUME_FIELD ); int tupleCount = currentWindow.get( TUPLE_COUNT_FIELD ); for ( Tuple input : ctx.getInputTuplesByDefaultPort() ) { final double vwap = input.getDoubleValueOrDefault( TUPLE_INPUT_VWAP_FIELD, 0d ); final double volume = input.getDoubleValueOrDefault( TUPLE_VOLUME_FIELD, 0d ); final int i = tupleCount++ % windowSize; vwapSum -= vwapValues[ i ]; volumeSum -= volumeValues[ i ]; vwapValues[ i ] = vwap; volumeValues[ i ] = volume; vwapSum += vwap; volumeSum += volume; if ( endOfWindow( tupleCount ) || endOfSlide( tupleCount ) ) { final Tuple result = createOutputTuple( input.getString( TICKER_SYMBOL_FIELD ), input.getLong( TIMESTAMP_FIELD ), vwapSum, volumeSum ); result.attachTo( input ); ctx.output( result ); } } currentWindow.set( SINGLE_VWAP_FIELD, vwapSum ).set( SINGLE_VOLUME_FIELD, volumeSum ).set( TUPLE_COUNT_FIELD, tupleCount ); kvStore.set( WINDOW_KEY, currentWindow ); }
VWAPAggregatorOperator implements Operator { @Override public void invoke ( final InvocationCtx ctx ) { final KVStore kvStore = ctx.getKVStore(); final Tuple currentWindow = kvStore.getOrDefault( WINDOW_KEY, this::createWindowTuple ); final double[] vwapValues = currentWindow.get( VWAPS_FIELD ); final double[] volumeValues = currentWindow.get( VOLUMES_FIELD ); double vwapSum = currentWindow.get( SINGLE_VWAP_FIELD ); double volumeSum = currentWindow.get( SINGLE_VOLUME_FIELD ); int tupleCount = currentWindow.get( TUPLE_COUNT_FIELD ); for ( Tuple input : ctx.getInputTuplesByDefaultPort() ) { final double vwap = input.getDoubleValueOrDefault( TUPLE_INPUT_VWAP_FIELD, 0d ); final double volume = input.getDoubleValueOrDefault( TUPLE_VOLUME_FIELD, 0d ); final int i = tupleCount++ % windowSize; vwapSum -= vwapValues[ i ]; volumeSum -= volumeValues[ i ]; vwapValues[ i ] = vwap; volumeValues[ i ] = volume; vwapSum += vwap; volumeSum += volume; if ( endOfWindow( tupleCount ) || endOfSlide( tupleCount ) ) { final Tuple result = createOutputTuple( input.getString( TICKER_SYMBOL_FIELD ), input.getLong( TIMESTAMP_FIELD ), vwapSum, volumeSum ); result.attachTo( input ); ctx.output( result ); } } currentWindow.set( SINGLE_VWAP_FIELD, vwapSum ).set( SINGLE_VOLUME_FIELD, volumeSum ).set( TUPLE_COUNT_FIELD, tupleCount ); kvStore.set( WINDOW_KEY, currentWindow ); } }
VWAPAggregatorOperator implements Operator { @Override public void invoke ( final InvocationCtx ctx ) { final KVStore kvStore = ctx.getKVStore(); final Tuple currentWindow = kvStore.getOrDefault( WINDOW_KEY, this::createWindowTuple ); final double[] vwapValues = currentWindow.get( VWAPS_FIELD ); final double[] volumeValues = currentWindow.get( VOLUMES_FIELD ); double vwapSum = currentWindow.get( SINGLE_VWAP_FIELD ); double volumeSum = currentWindow.get( SINGLE_VOLUME_FIELD ); int tupleCount = currentWindow.get( TUPLE_COUNT_FIELD ); for ( Tuple input : ctx.getInputTuplesByDefaultPort() ) { final double vwap = input.getDoubleValueOrDefault( TUPLE_INPUT_VWAP_FIELD, 0d ); final double volume = input.getDoubleValueOrDefault( TUPLE_VOLUME_FIELD, 0d ); final int i = tupleCount++ % windowSize; vwapSum -= vwapValues[ i ]; volumeSum -= volumeValues[ i ]; vwapValues[ i ] = vwap; volumeValues[ i ] = volume; vwapSum += vwap; volumeSum += volume; if ( endOfWindow( tupleCount ) || endOfSlide( tupleCount ) ) { final Tuple result = createOutputTuple( input.getString( TICKER_SYMBOL_FIELD ), input.getLong( TIMESTAMP_FIELD ), vwapSum, volumeSum ); result.attachTo( input ); ctx.output( result ); } } currentWindow.set( SINGLE_VWAP_FIELD, vwapSum ).set( SINGLE_VOLUME_FIELD, volumeSum ).set( TUPLE_COUNT_FIELD, tupleCount ); kvStore.set( WINDOW_KEY, currentWindow ); }  }
VWAPAggregatorOperator implements Operator { @Override public void invoke ( final InvocationCtx ctx ) { final KVStore kvStore = ctx.getKVStore(); final Tuple currentWindow = kvStore.getOrDefault( WINDOW_KEY, this::createWindowTuple ); final double[] vwapValues = currentWindow.get( VWAPS_FIELD ); final double[] volumeValues = currentWindow.get( VOLUMES_FIELD ); double vwapSum = currentWindow.get( SINGLE_VWAP_FIELD ); double volumeSum = currentWindow.get( SINGLE_VOLUME_FIELD ); int tupleCount = currentWindow.get( TUPLE_COUNT_FIELD ); for ( Tuple input : ctx.getInputTuplesByDefaultPort() ) { final double vwap = input.getDoubleValueOrDefault( TUPLE_INPUT_VWAP_FIELD, 0d ); final double volume = input.getDoubleValueOrDefault( TUPLE_VOLUME_FIELD, 0d ); final int i = tupleCount++ % windowSize; vwapSum -= vwapValues[ i ]; volumeSum -= volumeValues[ i ]; vwapValues[ i ] = vwap; volumeValues[ i ] = volume; vwapSum += vwap; volumeSum += volume; if ( endOfWindow( tupleCount ) || endOfSlide( tupleCount ) ) { final Tuple result = createOutputTuple( input.getString( TICKER_SYMBOL_FIELD ), input.getLong( TIMESTAMP_FIELD ), vwapSum, volumeSum ); result.attachTo( input ); ctx.output( result ); } } currentWindow.set( SINGLE_VWAP_FIELD, vwapSum ).set( SINGLE_VOLUME_FIELD, volumeSum ).set( TUPLE_COUNT_FIELD, tupleCount ); kvStore.set( WINDOW_KEY, currentWindow ); }  @Override SchedulingStrategy init( final InitCtx ctx ); @Override void invoke( final InvocationCtx ctx ); }
VWAPAggregatorOperator implements Operator { @Override public void invoke ( final InvocationCtx ctx ) { final KVStore kvStore = ctx.getKVStore(); final Tuple currentWindow = kvStore.getOrDefault( WINDOW_KEY, this::createWindowTuple ); final double[] vwapValues = currentWindow.get( VWAPS_FIELD ); final double[] volumeValues = currentWindow.get( VOLUMES_FIELD ); double vwapSum = currentWindow.get( SINGLE_VWAP_FIELD ); double volumeSum = currentWindow.get( SINGLE_VOLUME_FIELD ); int tupleCount = currentWindow.get( TUPLE_COUNT_FIELD ); for ( Tuple input : ctx.getInputTuplesByDefaultPort() ) { final double vwap = input.getDoubleValueOrDefault( TUPLE_INPUT_VWAP_FIELD, 0d ); final double volume = input.getDoubleValueOrDefault( TUPLE_VOLUME_FIELD, 0d ); final int i = tupleCount++ % windowSize; vwapSum -= vwapValues[ i ]; volumeSum -= volumeValues[ i ]; vwapValues[ i ] = vwap; volumeValues[ i ] = volume; vwapSum += vwap; volumeSum += volume; if ( endOfWindow( tupleCount ) || endOfSlide( tupleCount ) ) { final Tuple result = createOutputTuple( input.getString( TICKER_SYMBOL_FIELD ), input.getLong( TIMESTAMP_FIELD ), vwapSum, volumeSum ); result.attachTo( input ); ctx.output( result ); } } currentWindow.set( SINGLE_VWAP_FIELD, vwapSum ).set( SINGLE_VOLUME_FIELD, volumeSum ).set( TUPLE_COUNT_FIELD, tupleCount ); kvStore.set( WINDOW_KEY, currentWindow ); }  @Override SchedulingStrategy init( final InitCtx ctx ); @Override void invoke( final InvocationCtx ctx ); static final String WINDOW_SIZE_CONfIG_PARAMETER; static final String SLIDE_FACTOR_CONfIG_PARAMETER; static final String TICKER_SYMBOL_FIELD; static final String SINGLE_VWAP_FIELD; static final String SINGLE_VOLUME_FIELD; static final String TUPLE_INPUT_VWAP_FIELD; static final String TUPLE_VOLUME_FIELD; static final String TIMESTAMP_FIELD; static final String TUPLE_COUNT_FIELD; static final String WINDOW_KEY; static final String VWAPS_FIELD; static final String VOLUMES_FIELD; }