MultiWindowRangeKeyVal extends RangeKeyVal<K, V> { @SuppressWarnings({ "unchecked", "rawtypes" }) @Override public void endWindow() { boolean emit = (++windowCount) % windowSize == 0; if (!emit) { return; } for (Map.Entry<K, V> e: high.entrySet()) { HighLow<V> hl = new HighLow<V>(); hl.setHigh(getValue(e.getValue().doubleValue())); hl.setLow(getValue(low.get(e.getKey()).doubleValue())); range.emit(new KeyValPair(e.getKey(), hl)); } clearCache(); }  void setWindowSize(int windowSize); @SuppressWarnings({ "unchecked", "rawtypes" }) @Override void endWindow();  }
@SuppressWarnings({ "rawtypes", "unchecked" }) @Test public void testNodeProcessing() throws InterruptedException { MultiWindowRangeKeyVal<String, Integer> oper = new MultiWindowRangeKeyVal<String, Integer>(); CollectorTestSink swinSink = new CollectorTestSink(); oper.range.setSink(swinSink); oper.beginWindow(0); KeyValPair<String, Integer> low = new KeyValPair<String, Integer>("a", 3); oper.data.process(low); KeyValPair<String, Integer> high = new KeyValPair<String, Integer>("a", 11); oper.data.process(high); oper.endWindow(); oper.beginWindow(1); low = new KeyValPair<String, Integer>("a", 1); oper.data.process(low); high = new KeyValPair<String, Integer>("a", 9); oper.data.process(high); oper.endWindow(); Assert.assertEquals("number emitted tuples", 1, swinSink.collectedTuples.size()); for (Object o : swinSink.collectedTuples) { log.debug(o.toString()); } }