Merge pull request #1656 from zsxwing/fix-takeUntil

Make TakeUntil obey Rx contract
ReactiveX · Sep 5, 2014 · a8ff607 · a8ff607
2 parents 24ce690 + 5224d7d
commit a8ff607
Showing 1 changed file with 2 additions and 1 deletion.
diff --git a/rxjava-core/src/main/java/rx/internal/operators/OperatorTakeUntil.java b/rxjava-core/src/main/java/rx/internal/operators/OperatorTakeUntil.java
@@ -18,6 +18,7 @@
 import rx.Observable;
 import rx.Observable.Operator;
 import rx.Subscriber;
+import rx.observers.SerializedSubscriber;
 
 /**
  * Returns an Observable that emits the items from the source Observable until another Observable
@@ -35,7 +36,7 @@ public OperatorTakeUntil(final Observable<? extends E> other) {
 
     @Override
     public Subscriber<? super T> call(final Subscriber<? super T> child) {
-        final Subscriber<T> parent = new Subscriber<T>(child) {
+        final Subscriber<T> parent = new SerializedSubscriber<T>(child) {
 
             @Override
             public void onCompleted() {