Fix issue ReactiveX#1451

zsxwing · Jul 17, 2014 · 53b4659 · 53b4659 · akarnokd · Jul 17, 2014
1 parent cb75b97
commit 53b4659
Show file tree

Hide file tree

Showing 2 changed files with 70 additions and 0 deletions.
diff --git a/rxjava-core/src/main/java/rx/internal/operators/OperatorDoOnEach.java b/rxjava-core/src/main/java/rx/internal/operators/OperatorDoOnEach.java
@@ -33,30 +33,45 @@ public OperatorDoOnEach(Observer<? super T> doOnEachObserver) {
     @Override
     public Subscriber<? super T> call(final Subscriber<? super T> observer) {
         return new Subscriber<T>(observer) {
+
+            private boolean done = false;
+
             @Override
             public void onCompleted() {
+                if (done) {
+                    return;
+                }
                 try {
                     doOnEachObserver.onCompleted();
                 } catch (Throwable e) {
                     onError(e);
                     return;
                 }
                 observer.onCompleted();
+                // Set `done` here so that the error in `doOnEachObserver.onCompleted()` can be noticed by observer
+                done = true;
             }
 
             @Override
             public void onError(Throwable e) {
+                if (done) {
+                    return;
+                }
                 try {
                     doOnEachObserver.onError(e);
                 } catch (Throwable e2) {
                     observer.onError(e2);
                     return;
                 }
                 observer.onError(e);
+                done = true;
             }
 
             @Override
             public void onNext(T value) {
+                if (done) {
+                    return;
+                }
                 try {
                     doOnEachObserver.onNext(value);
                 } catch (Throwable e) {

diff --git a/rxjava-core/src/test/java/rx/internal/operators/OperatorDoOnEachTest.java b/rxjava-core/src/test/java/rx/internal/operators/OperatorDoOnEachTest.java
@@ -15,6 +15,7 @@
  */
 package rx.internal.operators;
 
+import static org.junit.Assert.assertEquals;
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
@@ -30,6 +31,9 @@
 import rx.functions.Action1;
 import rx.functions.Func1;
 
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
 public class OperatorDoOnEachTest {
 
     @Mock
@@ -114,4 +118,55 @@ public void call(String s) {
 
     }
 
+    @Test
+    public void testIssue1451Case1() {
+        // https://github.com/Netflix/RxJava/issues/1451
+        int[] nums = {1, 2, 3};
+        final AtomicInteger count = new AtomicInteger();
+        for (final int n : nums) {
+            Observable
+                    .from(Boolean.TRUE, Boolean.FALSE)
+                    .takeWhile(new Func1<Boolean, Boolean>() {
+                        @Override
+                        public Boolean call(Boolean value) {
+                            return value;
+                        }
+                    })
+                    .toList()
+                    .doOnNext(new Action1<List<Boolean>>() {
+                        @Override
+                        public void call(List<Boolean> booleans) {
+                            count.incrementAndGet();
+                        }
+                    })
+                    .subscribe();
+        }
+        assertEquals(nums.length, count.get());
+    }
+
+    @Test
+    public void testIssue1451Case2() {
+        // https://github.com/Netflix/RxJava/issues/1451
+        int[] nums = {1, 2, 3};
+        final AtomicInteger count = new AtomicInteger();
+        for (final int n : nums) {
+            Observable
+                    .from(Boolean.TRUE, Boolean.FALSE, Boolean.FALSE)
+                    .takeWhile(new Func1<Boolean, Boolean>() {
+                        @Override
+                        public Boolean call(Boolean value) {
+                            return value;
+                        }
+                    })
+                    .toList()
+                    .doOnNext(new Action1<List<Boolean>>() {
+                        @Override
+                        public void call(List<Boolean> booleans) {
+                            count.incrementAndGet();
+                        }
+                    })
+                    .subscribe();
+        }
+        assertEquals(nums.length, count.get());
+    }
 }