Merge pull request #33 from MichaelBlume/emit-zeros

Ensure zeros get reported.

src/jvm/storm/starter/tools/SlidingWindowCounter.java

@@ -85,6 +85,7 @@ public void incrementCount(T obj) {
      */
     public Map<T, Long> getCountsThenAdvanceWindow() {
         Map<T, Long> counts = objCounter.getCounts();
+        objCounter.wipeZeros();
         objCounter.wipeSlot(tailSlot);
         advanceHead();
         return counts;

src/jvm/storm/starter/tools/SlotBasedCounter.java

@@ -68,21 +68,11 @@ private long computeTotalCount(T obj) {
     /**
      * Reset the slot count of any tracked objects to zero for the given slot.
      * 
-     * <em>Implementation detail:</em> As an optimization this method will also remove any object from the counter whose
-     * total count is zero after the wipe of the slot (to free up memory).
-     * 
      * @param slot
      */
     public void wipeSlot(int slot) {
-        Set<T> objToBeRemoved = new HashSet<T>();
         for (T obj : objToCounts.keySet()) {
             resetSlotCountToZero(obj, slot);
-            if (shouldBeRemovedFromCounter(obj)) {
-                objToBeRemoved.add(obj);
-            }
-        }
-        for (T obj : objToBeRemoved) {
-            objToCounts.remove(obj);
         }
     }
 
@@ -95,4 +85,19 @@ private boolean shouldBeRemovedFromCounter(T obj) {
         return computeTotalCount(obj) == 0;
     }
 
+    /**
+     * Remove any object from the counter whose total count is zero (to free up memory).
+     */
+    public void wipeZeros() {
+        Set<T> objToBeRemoved = new HashSet<T>();
+        for (T obj : objToCounts.keySet()) {
+            if (shouldBeRemovedFromCounter(obj)) {
+                objToBeRemoved.add(obj);
+            }
+        }
+        for (T obj : objToBeRemoved) {
+            objToCounts.remove(obj);
+        }
+    }
+
 }