Fix related classes for issue ReactiveX#1451

zsxwing · Jul 17, 2014 · 6bbd921 · 6bbd921
1 parent 53b4659
commit 6bbd921
Show file tree

Hide file tree

Showing 4 changed files with 18 additions and 6 deletions.
diff --git a/rxjava-core/src/main/java/rx/internal/operators/OnSubscribeFromIterable.java b/rxjava-core/src/main/java/rx/internal/operators/OnSubscribeFromIterable.java
@@ -70,7 +70,9 @@ public void request(long n) {
                     }
                     o.onNext(it.next());
                 }
-                o.onCompleted();
+                if (!o.isUnsubscribed()) {
+                    o.onCompleted();
+                }
             } else if(n > 0) {
                 // backpressure is requested
                 long _c = REQUESTED_UPDATER.getAndAdd(this, n);

diff --git a/rxjava-core/src/main/java/rx/internal/operators/OnSubscribeRange.java b/rxjava-core/src/main/java/rx/internal/operators/OnSubscribeRange.java
@@ -64,7 +64,9 @@ public void request(long n) {
                     }
                     o.onNext((int) i);
                 }
-                o.onCompleted();
+                if (!o.isUnsubscribed()) {
+                    o.onCompleted();
+                }
             } else if (n > 0) {
                 // backpressure is requested
                 long _c = REQUESTED_UPDATER.getAndAdd(this, n);

diff --git a/rxjava-core/src/main/java/rx/internal/operators/OperatorDoOnEach.java b/rxjava-core/src/main/java/rx/internal/operators/OperatorDoOnEach.java
@@ -47,24 +47,24 @@ public void onCompleted() {
                     onError(e);
                     return;
                 }
-                observer.onCompleted();
                 // Set `done` here so that the error in `doOnEachObserver.onCompleted()` can be noticed by observer
                 done = true;
+                observer.onCompleted();
             }
 
             @Override
             public void onError(Throwable e) {
                 if (done) {
                     return;
                 }
+                done = true;
                 try {
                     doOnEachObserver.onError(e);
                 } catch (Throwable e2) {
                     observer.onError(e2);
                     return;
                 }
                 observer.onError(e);
-                done = true;
             }
 
             @Override

diff --git a/rxjava-core/src/main/java/rx/internal/operators/OperatorTakeWhile.java b/rxjava-core/src/main/java/rx/internal/operators/OperatorTakeWhile.java
@@ -49,32 +49,40 @@ public Subscriber<? super T> call(final Subscriber<? super T> subscriber) {
 
             private int counter = 0;
 
+            private boolean done = false;
+
             @Override
             public void onNext(T args) {
                 boolean isSelected;
                 try {
                     isSelected = predicate.call(args, counter++);
                 } catch (Throwable e) {
+                    done = true;
                     subscriber.onError(e);
                     unsubscribe();
                     return;
                 }
                 if (isSelected) {
                     subscriber.onNext(args);
                 } else {
+                    done = true;
                     subscriber.onCompleted();
                     unsubscribe();
                 }
             }
 
             @Override
             public void onCompleted() {
-                subscriber.onCompleted();
+                if (!done) {
+                    subscriber.onCompleted();
+                }
             }
 
             @Override
             public void onError(Throwable e) {
-                subscriber.onError(e);
+                if (!done) {
+                    subscriber.onError(e);
+                }
             }
 
         };