MultiWindowSumKeyVal extends SumKeyVal<K, V> { @SuppressWarnings({ "unchecked", "rawtypes" }) @Override public void endWindow() { boolean emit = (++windowCount) % windowSize == 0; if (!emit) { return; } boolean dosum = sum.isConnected(); if (dosum) { for (Map.Entry<K, SumEntry> e : sums.entrySet()) { K key = e.getKey(); if (dosum) { sum.emit(new KeyValPair(key, getValue(e.getValue().sum.doubleValue()))); } } } sums.clear(); }  void setWindowSize(int windowSize); @SuppressWarnings({ "unchecked", "rawtypes" }) @Override void endWindow();  }
@SuppressWarnings({ "rawtypes", "unchecked" }) @Test public void testNodeProcessing() throws InterruptedException { MultiWindowSumKeyVal<String, Integer> oper = new MultiWindowSumKeyVal<String, Integer>(); CollectorTestSink swinSink = new CollectorTestSink(); oper.sum.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()); } }