Range extends BaseNumberValueOperator<V> { @Override public void endWindow() { if ((low != null) && (high != null)) { HighLow tuple = new HighLow(getValue(high.doubleValue()), getValue(low.doubleValue())); range.emit(tuple); } high = null; low = null; }  @Override void endWindow(); final transient DefaultInputPort<V> data; final transient DefaultOutputPort<HighLow<V>> range; }
@SuppressWarnings("unchecked") @Test public void testNodeSchemaProcessing() { Range<Double> oper = new Range<Double>(); TestSink rangeSink = new TestSink(); oper.range.setSink(rangeSink); oper.beginWindow(0); int numTuples = 1000; for (int i = 0; i < numTuples; i++) { Double a = new Double(20.0); Double b = new Double(2.0); Double c = new Double(1000.0); oper.data.process(a); oper.data.process(b); oper.data.process(c); a = 1.0; oper.data.process(a); a = 10.0; oper.data.process(a); b = 5.0; oper.data.process(b); b = 12.0; oper.data.process(b); c = 22.0; oper.data.process(c); c = 14.0; oper.data.process(c); a = 46.0; oper.data.process(a); b = 2.0; oper.data.process(b); a = 23.0; oper.data.process(a); } oper.endWindow(); Assert.assertEquals("number emitted tuples", 1, rangeSink.collectedTuples.size()); for (Object o : rangeSink.collectedTuples) { HighLow<V> hl = (HighLow<V>)o; Assert.assertEquals("emitted high value was ", 1000.0, hl.getHigh()); Assert.assertEquals("emitted low value was ", 1.0, hl.getLow()); } }