OperatorTakeUntil

Author: zsxwing

rxjava-core/src/main/java/rx/Observable.java

@@ -6555,7 +6555,7 @@ public final Observable<List<T>> takeLastBuffer(long time, TimeUnit unit, Schedu
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Conditional-and-Boolean-Operators#wiki-takeuntil">RxJava Wiki: takeUntil()</a>
      */
     public final <E> Observable<T> takeUntil(Observable<? extends E> other) {
-        return OperationTakeUntil.takeUntil(this, other);
+        return OperatorTakeUntil.takeUntil(this, other);
     }
 
     /**

rxjava-core/src/main/java/rx/operators/OperationTakeUntil.java renamed to rxjava-core/src/main/java/rx/operators/OperatorTakeUntil.java

@@ -16,19 +16,18 @@
 package rx.operators;
 
 import rx.Observable;
-import rx.Observable.OnSubscribeFunc;
-import rx.Observer;
 import rx.Subscriber;
-import rx.Subscription;
 import rx.functions.Func1;
 
+import static rx.Observable.Operator;
+
 /**
  * Returns an Observable that emits the items from the source Observable until another Observable
  * emits an item.
  * <p>
  * <img width="640" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/takeUntil.png">
  */
-public class OperationTakeUntil {
+public final class OperatorTakeUntil {
 
     /**
      * Returns the values from the source observable sequence until the other observable sequence produces a value.
@@ -44,8 +43,8 @@ public class OperationTakeUntil {
      * @return An observable sequence containing the elements of the source sequence up to the point the other sequence interrupted further propagation.
      */
     public static <T, E> Observable<T> takeUntil(final Observable<? extends T> source, final Observable<? extends E> other) {
-        Observable<Notification<T>> s = Observable.create(new SourceObservable<T>(source));
-        Observable<Notification<T>> o = Observable.create(new OtherObservable<T, E>(other));
+        Observable<Notification<T>> s = source.lift(new SourceObservable<T>());
+        Observable<Notification<T>> o = other.lift(new OtherObservable<T, E>());
 
         Observable<Notification<T>> result = Observable.merge(s, o);
 
@@ -81,19 +80,14 @@ private Notification(boolean halt, T value) {
 
     }
 
-    private static class SourceObservable<T> implements OnSubscribeFunc<Notification<T>> {
-        private final Observable<? extends T> sequence;
-
-        private SourceObservable(Observable<? extends T> sequence) {
-            this.sequence = sequence;
-        }
+    private static class SourceObservable<T> implements Operator<Notification<T>, T> {
 
         @Override
-        public Subscription onSubscribe(final Observer<? super Notification<T>> notificationObserver) {
-            return sequence.unsafeSubscribe(new Subscriber<T>() {
+        public Subscriber<? super T> call(final Subscriber<? super Notification<T>> notificationObserver) {
+            return new Subscriber<T>(notificationObserver) {
                 @Override
                 public void onCompleted() {
-                    notificationObserver.onNext(Notification.<T> halt());
+                    notificationObserver.onNext(Notification.<T>halt());
                 }
 
                 @Override
@@ -105,23 +99,18 @@ public void onError(Throwable e) {
                 public void onNext(T args) {
                     notificationObserver.onNext(Notification.value(args));
                 }
-            });
+            };
         }
     }
 
-    private static class OtherObservable<T, E> implements OnSubscribeFunc<Notification<T>> {
-        private final Observable<? extends E> sequence;
-
-        private OtherObservable(Observable<? extends E> sequence) {
-            this.sequence = sequence;
-        }
+    private static class OtherObservable<T, E> implements Operator<Notification<T>, E> {
 
         @Override
-        public Subscription onSubscribe(final Observer<? super Notification<T>> notificationObserver) {
-            return sequence.unsafeSubscribe(new Subscriber<E>() {
+        public Subscriber<? super E> call(final Subscriber<? super Notification<T>> notificationObserver) {
+            return new Subscriber<E>(notificationObserver) {
                 @Override
                 public void onCompleted() {
-                    notificationObserver.onNext(Notification.<T> halt());
+                    notificationObserver.onNext(Notification.<T>halt());
                 }
 
                 @Override
@@ -131,9 +120,9 @@ public void onError(Throwable e) {
 
                 @Override
                 public void onNext(E args) {
-                    notificationObserver.onNext(Notification.<T> halt());
+                    notificationObserver.onNext(Notification.<T>halt());
                 }
-            });
+            };
         }
     }
 }

rxjava-core/src/test/java/rx/operators/OperationTakeUntilTest.java renamed to rxjava-core/src/test/java/rx/operators/OperatorTakeUntilTest.java

@@ -18,15 +18,15 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
-import static rx.operators.OperationTakeUntil.takeUntil;
+import static rx.operators.OperatorTakeUntil.takeUntil;
 
 import org.junit.Test;
 
 import rx.Observable;
 import rx.Observer;
 import rx.Subscription;
 
-public class OperationTakeUntilTest {
+public class OperatorTakeUntilTest {
 
     @Test
     @SuppressWarnings("unchecked")