UniqueCounter extends BaseUniqueKeyCounter<K> { @Override public void endWindow() { HashMap<K, Integer> tuple = new HashMap<>(); for (K key: inputSet) { tuple.put(key, map.get(key).toInteger()); } if (!tuple.isEmpty()) { count.emit(tuple); } if (!cumulative) { map.clear(); } inputSet.clear(); }  @Override void endWindow(); boolean isCumulative(); void setCumulative(boolean cumulative); final transient DefaultInputPort<K> data; final transient DefaultOutputPort<HashMap<K, Integer>> count; }
@SuppressWarnings({ "rawtypes", "unchecked" }) @Test public void testNodeProcessing() throws Exception { UniqueCounter<String> oper = new UniqueCounter<String>(); CountAndLastTupleTestSink sink = new CountAndLastTupleTestSink<HashMap<String, Integer>>(); oper.count.setSink(sink); String atuple = "a"; String btuple = "b"; String ctuple = "c"; String dtuple = "d"; int numTuples = 10000; oper.beginWindow(0); for (int i = 0; i < numTuples; i++) { oper.data.process(atuple); if (i % 2 == 0) { oper.data.process(btuple); } if (i % 5 == 0) { oper.data.process(ctuple); } } oper.endWindow(); oper.beginWindow(1); for (int i = 0; i < numTuples; i++) { oper.data.process(atuple); } oper.endWindow(); HashMap<String, Integer> tuple = (HashMap<String, Integer>)sink.tuple; int acount = tuple.get("a"); Assert.assertEquals("number emitted tuples", numTuples, acount); oper.beginWindow(2); for (int i = 0; i < numTuples; i++) { if (i % 2 == 0) { oper.data.process(btuple); } oper.data.process(btuple); if (i % 10 == 0) { oper.data.process(dtuple); } } oper.endWindow(); tuple = (HashMap<String, Integer>)sink.tuple; int bcount = tuple.get("b"); int dcount = tuple.get("d"); Assert.assertEquals("number emitted tuples", 3, sink.count); Assert.assertEquals("number emitted tuples", numTuples + (numTuples / 2), bcount); Assert.assertEquals("number emitted tuples", numTuples / 10, dcount); }