Emit reset-stream event after updating counts

Author: cboehme

metafacture-flowcontrol/src/main/java/org/metafacture/flowcontrol/ObjectBatchResetter.java

@@ -91,9 +91,9 @@ public void process(final T obj) {
 
         objectCount += 1;
         if (objectCount >= batchSize) {
-            getReceiver().resetStream();
             batchCount += 1;
             objectCount = 0;
+            getReceiver().resetStream();
         }
     }
 

metafacture-flowcontrol/src/test/java/org/metafacture/flowcontrol/ObjectBatchResetterTest.java

@@ -23,6 +23,7 @@
 import org.junit.Rule;
 import org.junit.Test;
 import org.metafacture.framework.ObjectReceiver;
+import org.metafacture.framework.helpers.DefaultObjectReceiver;
 import org.mockito.InOrder;
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnit;
@@ -113,6 +114,22 @@ public void shouldResetCountsOnResetStream() {
                 .isZero();
     }
 
+    @Test
+    public void shouldEmitResetStreamEventAfterUpdatingCounts() {
+
+        systemUnderTest.setBatchSize(2);
+        systemUnderTest.setReceiver(new DefaultObjectReceiver<String>() {
+            @Override
+            public void resetStream() {
+                assertThat(systemUnderTest.getObjectCount()).isZero();
+                assertThat(systemUnderTest.getBatchCount()).isEqualTo(1);
+            }
+        });
+
+        systemUnderTest.process("1");
+        systemUnderTest.process("2");
+    }
+
     @Rule
     public MockitoRule mockitoRule = MockitoJUnit.rule();