fix issues with doOnNextCount() operator, add tests

Author: thomasnield

build.gradle

@@ -18,7 +18,7 @@ sourceCompatibility = JavaVersion.VERSION_1_8
 targetCompatibility = JavaVersion.VERSION_1_8
 
 dependencies {
-    compile 'io.reactivex.rxjava2:rxjava:2.1.3'
+    compile 'io.reactivex.rxjava2:rxjava:2.1.6'
     testCompile 'junit:junit-dep:4.+'
     testCompile 'org.mockito:mockito-core:1.8.5'
 }

gradle.properties

@@ -1 +1 @@
-version=2.2.0
+version=2.2.2

gradle/wrapper/gradle-wrapper.jar

@@ -1,6 +1,6 @@
-#Wed Feb 05 12:05:54 CET 2014
+#Sun Nov 26 18:15:25 CST 2017
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
-distributionUrl=http\://services.gradle.org/distributions/gradle-2.8-all.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-2.8-all.zip

gradle/wrapper/gradle-wrapper.properties

@@ -42,11 +42,6 @@ case "`uname`" in
     ;;
 esac
 
-# For Cygwin, ensure paths are in UNIX format before anything is touched.
-if $cygwin ; then
-    [ -n "$JAVA_HOME" ] && JAVA_HOME=`cygpath --unix "$JAVA_HOME"`
-fi
-
 # Attempt to set APP_HOME
 # Resolve links: $0 may be a link
 PRG="$0"
@@ -61,9 +56,9 @@ while [ -h "$PRG" ] ; do
     fi
 done
 SAVED="`pwd`"
-cd "`dirname \"$PRG\"`/" >&-
+cd "`dirname \"$PRG\"`/" >/dev/null
 APP_HOME="`pwd -P`"
-cd "$SAVED" >&-
+cd "$SAVED" >/dev/null
 
 CLASSPATH=$APP_HOME/gradle/wrapper/gradle-wrapper.jar
 
@@ -114,6 +109,7 @@ fi
 if $cygwin ; then
     APP_HOME=`cygpath --path --mixed "$APP_HOME"`
     CLASSPATH=`cygpath --path --mixed "$CLASSPATH"`
+    JAVACMD=`cygpath --unix "$JAVACMD"`
 
     # We build the pattern for arguments to be converted via cygpath
     ROOTDIRSRAW=`find -L / -maxdepth 1 -mindepth 1 -type d 2>/dev/null`

gradlew

@@ -217,8 +217,9 @@ public void onNext(T t) {
                     if (done)
                         return;
                     try {
+                        ++count;
                         if (ctObserver.doOnNextCountAction != null)
-                            ctObserver.doOnNextCountAction.accept(++count);
+                            ctObserver.doOnNextCountAction.accept(count);
                     } catch(Exception e) {
                         Exceptions.throwIfFatal(e);
                         onError(e);

src/main/java/io/reactivex/rxjavafx/transformers/FxFlowableTransformers.java

@@ -210,8 +210,9 @@ public void onNext(T t) {
                     if (done)
                         return;
                     try {
+                        ++count;
                         if (ctObserver.doOnNextCountAction != null)
-                            ctObserver.doOnNextCountAction.accept(++count);
+                            ctObserver.doOnNextCountAction.accept(count);
                     } catch(Exception e) {
                         Exceptions.throwIfFatal(e);
                         onError(e);

src/main/java/io/reactivex/rxjavafx/transformers/FxObservableTransformers.java

@@ -0,0 +1,92 @@
+package io.reactivex.rxjavafx.operators;
+
+import io.reactivex.Flowable;
+import io.reactivex.Observable;
+import io.reactivex.rxjavafx.transformers.FxFlowableTransformers;
+import io.reactivex.rxjavafx.transformers.FxObservableTransformers;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public final class OperatorsTest {
+
+    @Test
+    public void testDoOnNextCountObservable() {
+
+        final List<Integer> onNextCounts = new ArrayList<>();
+
+        Observable.just("Alpha", "Beta", "Gamma")
+                .compose(FxObservableTransformers.doOnNextCount(onNextCounts::add))
+                .subscribe();
+
+        Assert.assertTrue(onNextCounts.containsAll(Arrays.asList(1, 2, 3)));
+    }
+
+    @Test
+    public void testDoOnCompleteCountObservable() {
+        AtomicInteger onCompleteCount = new AtomicInteger();
+
+
+        Observable.just("Alpha", "Beta", "Gamma")
+                .compose(FxObservableTransformers.doOnCompleteCount(onCompleteCount::set))
+                .subscribe();
+
+        Assert.assertTrue(onCompleteCount.get() == 3);
+    }
+
+    @Test
+    public void testDoOnErrorCountObservable() {
+        AtomicInteger onErrorCount = new AtomicInteger();
+
+
+        Observable.just(5, 10, 15, 0, 20)
+                .map(i -> 5 / i)
+                .compose(FxObservableTransformers.doOnErrorCount(onErrorCount::set))
+                .subscribe();
+
+        Assert.assertTrue(onErrorCount.get() == 3);
+    }
+
+
+    @Test
+    public void testDoOnNextCountFlowable() {
+
+        final List<Integer> onNextCounts = new ArrayList<>();
+
+
+        Flowable.just("Alpha", "Beta", "Gamma")
+                .compose(FxFlowableTransformers.doOnNextCount(onNextCounts::add))
+                .subscribe();
+
+        Assert.assertTrue(onNextCounts.containsAll(Arrays.asList(1, 2, 3)));
+    }
+
+    @Test
+    public void testDoOnCompleteCountFlowable() {
+        AtomicInteger onCompleteCount = new AtomicInteger();
+
+
+        Flowable.just("Alpha", "Beta", "Gamma")
+                .compose(FxFlowableTransformers.doOnCompleteCount(onCompleteCount::set))
+                .subscribe();
+
+        Assert.assertTrue(onCompleteCount.get() == 3);
+    }
+
+    @Test
+    public void testDoOnErrorCountFlowable() {
+        AtomicInteger onErrorCount = new AtomicInteger();
+
+
+        Flowable.just(5, 10, 15, 0, 20)
+                .map(i -> 5 / i)
+                .compose(FxFlowableTransformers.doOnErrorCount(onErrorCount::set))
+                .subscribe();
+
+        Assert.assertTrue(onErrorCount.get() == 3);
+    }
+}