@Test public void shouldReturnNoOutputWithQuoteButNoVWAP () { input.add( 1, newQuoteTuple( 0, 5d, 100 ) ); operator.invoke( invocationCtx ); assertThat( output.getTupleCount( 0 ), equalTo( 0 ) ); }
@Override public void invoke ( final InvocationCtx ctx ) { final KVStore kvStore = ctx.getKVStore(); final Iterator<Tuple> it = new MergedTupleListsIterator( ctx.getInputTuples( 0 ), ctx.getInputTuples( 1 ), comparing( t -> t.getLong( TIMESTAMP_FIELD ) ) ); while ( it.hasNext() ) { final Tuple tuple = it.next(); final String tickerSymbol = tuple.get( TICKER_SYMBOL_FIELD ); if ( tuple.contains( CVWAP_FIELD ) ) { kvStore.set( tickerSymbol, tuple.get( CVWAP_FIELD ) ); } else { final Double cvwap = kvStore.get( tickerSymbol ); if ( cvwap != null ) { final Tuple bargainIndex = createBargainIndexTuple( cvwap, tuple ); if ( bargainIndex != null ) { ctx.output( bargainIndex ); } } } } }
BargainIndexOperator implements Operator { @Override public void invoke ( final InvocationCtx ctx ) { final KVStore kvStore = ctx.getKVStore(); final Iterator<Tuple> it = new MergedTupleListsIterator( ctx.getInputTuples( 0 ), ctx.getInputTuples( 1 ), comparing( t -> t.getLong( TIMESTAMP_FIELD ) ) ); while ( it.hasNext() ) { final Tuple tuple = it.next(); final String tickerSymbol = tuple.get( TICKER_SYMBOL_FIELD ); if ( tuple.contains( CVWAP_FIELD ) ) { kvStore.set( tickerSymbol, tuple.get( CVWAP_FIELD ) ); } else { final Double cvwap = kvStore.get( tickerSymbol ); if ( cvwap != null ) { final Tuple bargainIndex = createBargainIndexTuple( cvwap, tuple ); if ( bargainIndex != null ) { ctx.output( bargainIndex ); } } } } } }
BargainIndexOperator implements Operator { @Override public void invoke ( final InvocationCtx ctx ) { final KVStore kvStore = ctx.getKVStore(); final Iterator<Tuple> it = new MergedTupleListsIterator( ctx.getInputTuples( 0 ), ctx.getInputTuples( 1 ), comparing( t -> t.getLong( TIMESTAMP_FIELD ) ) ); while ( it.hasNext() ) { final Tuple tuple = it.next(); final String tickerSymbol = tuple.get( TICKER_SYMBOL_FIELD ); if ( tuple.contains( CVWAP_FIELD ) ) { kvStore.set( tickerSymbol, tuple.get( CVWAP_FIELD ) ); } else { final Double cvwap = kvStore.get( tickerSymbol ); if ( cvwap != null ) { final Tuple bargainIndex = createBargainIndexTuple( cvwap, tuple ); if ( bargainIndex != null ) { ctx.output( bargainIndex ); } } } } }  }
BargainIndexOperator implements Operator { @Override public void invoke ( final InvocationCtx ctx ) { final KVStore kvStore = ctx.getKVStore(); final Iterator<Tuple> it = new MergedTupleListsIterator( ctx.getInputTuples( 0 ), ctx.getInputTuples( 1 ), comparing( t -> t.getLong( TIMESTAMP_FIELD ) ) ); while ( it.hasNext() ) { final Tuple tuple = it.next(); final String tickerSymbol = tuple.get( TICKER_SYMBOL_FIELD ); if ( tuple.contains( CVWAP_FIELD ) ) { kvStore.set( tickerSymbol, tuple.get( CVWAP_FIELD ) ); } else { final Double cvwap = kvStore.get( tickerSymbol ); if ( cvwap != null ) { final Tuple bargainIndex = createBargainIndexTuple( cvwap, tuple ); if ( bargainIndex != null ) { ctx.output( bargainIndex ); } } } } }  @Override SchedulingStrategy init( final InitCtx ctx ); @Override void invoke( final InvocationCtx ctx ); }
BargainIndexOperator implements Operator { @Override public void invoke ( final InvocationCtx ctx ) { final KVStore kvStore = ctx.getKVStore(); final Iterator<Tuple> it = new MergedTupleListsIterator( ctx.getInputTuples( 0 ), ctx.getInputTuples( 1 ), comparing( t -> t.getLong( TIMESTAMP_FIELD ) ) ); while ( it.hasNext() ) { final Tuple tuple = it.next(); final String tickerSymbol = tuple.get( TICKER_SYMBOL_FIELD ); if ( tuple.contains( CVWAP_FIELD ) ) { kvStore.set( tickerSymbol, tuple.get( CVWAP_FIELD ) ); } else { final Double cvwap = kvStore.get( tickerSymbol ); if ( cvwap != null ) { final Tuple bargainIndex = createBargainIndexTuple( cvwap, tuple ); if ( bargainIndex != null ) { ctx.output( bargainIndex ); } } } } }  @Override SchedulingStrategy init( final InitCtx ctx ); @Override void invoke( final InvocationCtx ctx ); static final String ASKED_TICKER_SYMBOL_PRICE_FIELD; static final String ASKED_SIZE_FIELD; static final String BARGAIN_INDEX_FIELD; }