Merge pull request #958 from akarnokd/OperatorSkipWhile

OperatorSkipWhile

Author: benjchristensen

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

@@ -91,7 +91,7 @@
 import rx.operators.OperationSkip;
 import rx.operators.OperationSkipLast;
 import rx.operators.OperationSkipUntil;
-import rx.operators.OperationSkipWhile;
+import rx.operators.OperatorSkipWhile;
 import rx.operators.OperationSum;
 import rx.operators.OperationSwitch;
 import rx.operators.OperationSynchronize;
@@ -6427,7 +6427,7 @@ public final <U> Observable<T> skipUntil(Observable<U> other) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229685.aspx">MSDN: Observable.SkipWhile</a>
      */
     public final Observable<T> skipWhile(Func1<? super T, Boolean> predicate) {
-        return create(OperationSkipWhile.skipWhile(this, predicate));
+        return lift(new OperatorSkipWhile<T>(OperatorSkipWhile.toPredicate2(predicate)));
     }
 
     /**
@@ -6445,7 +6445,7 @@ public final Observable<T> skipWhile(Func1<? super T, Boolean> predicate) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh211631.aspx">MSDN: Observable.SkipWhile</a>
      */
     public final Observable<T> skipWhileWithIndex(Func2<? super T, Integer, Boolean> predicate) {
-        return create(OperationSkipWhile.skipWhileWithIndex(this, predicate));
+        return lift(new OperatorSkipWhile<T>(predicate));
     }
 
     /**

rxjava-core/src/main/java/rx/operators/OperationSkipWhile.java

@@ -0,0 +1,78 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.operators;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import rx.Observable;
+import rx.Observable.OnSubscribeFunc;
+import rx.Observable.Operator;
+import rx.Observer;
+import rx.Subscriber;
+import rx.Subscription;
+import rx.functions.Func1;
+import rx.functions.Func2;
+
+/**
+ * Skips any emitted source items as long as the specified condition holds true. Emits all further source items
+ * as soon as the condition becomes false.
+ */
+public final class OperatorSkipWhile<T> implements Operator<T, T> {
+    private final Func2<? super T, Integer, Boolean> predicate;
+
+    public OperatorSkipWhile(Func2<? super T, Integer, Boolean> predicate) {
+        this.predicate = predicate;
+    }
+    @Override
+    public Subscriber<? super T> call(final Subscriber<? super T> child) {
+        return new Subscriber<T>(child) {
+            boolean skipping = true;
+            int index;
+            @Override
+            public void onNext(T t) {
+                if (!skipping) {
+                    child.onNext(t);
+                } else {
+                    if (!predicate.call(t, index++)) {
+                        skipping = false;
+                        child.onNext(t);
+                    }
+                }
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                child.onError(e);
+            }
+
+            @Override
+            public void onCompleted() {
+                child.onCompleted();
+            }
+        };
+    }
+    /** Convert to Func2 type predicate. */
+    public static <T> Func2<T, Integer, Boolean> toPredicate2(final Func1<? super T, Boolean> predicate) {
+        return new Func2<T, Integer, Boolean>() {
+
+            @Override
+            public Boolean call(T t1, Integer t2) {
+                return predicate.call(t1);
+            }
+        };
+    }
+}

rxjava-core/src/main/java/rx/operators/OperatorSkipWhile.java

@@ -15,9 +15,7 @@
  */
 package rx.operators;
 
-import static org.mockito.Matchers.*;
 import static org.mockito.Mockito.*;
-import static rx.operators.OperationSkipWhile.*;
 
 import org.junit.Test;
 import org.mockito.InOrder;
@@ -51,7 +49,7 @@ public Boolean call(Integer value, Integer index) {
     @Test
     public void testSkipWithIndex() {
         Observable<Integer> src = Observable.from(1, 2, 3, 4, 5);
-        Observable.create(skipWhileWithIndex(src, INDEX_LESS_THAN_THREE)).subscribe(w);
+        src.skipWhileWithIndex(INDEX_LESS_THAN_THREE).subscribe(w);
 
         InOrder inOrder = inOrder(w);
         inOrder.verify(w, times(1)).onNext(4);
@@ -63,7 +61,7 @@ public void testSkipWithIndex() {
     @Test
     public void testSkipEmpty() {
         Observable<Integer> src = Observable.empty();
-        Observable.create(skipWhile(src, LESS_THAN_FIVE)).subscribe(w);
+        src.skipWhile(LESS_THAN_FIVE).subscribe(w);
         verify(w, never()).onNext(anyInt());
         verify(w, never()).onError(any(Throwable.class));
         verify(w, times(1)).onCompleted();
@@ -72,7 +70,7 @@ public void testSkipEmpty() {
     @Test
     public void testSkipEverything() {
         Observable<Integer> src = Observable.from(1, 2, 3, 4, 3, 2, 1);
-        Observable.create(skipWhile(src, LESS_THAN_FIVE)).subscribe(w);
+        src.skipWhile(LESS_THAN_FIVE).subscribe(w);
         verify(w, never()).onNext(anyInt());
         verify(w, never()).onError(any(Throwable.class));
         verify(w, times(1)).onCompleted();
@@ -81,7 +79,7 @@ public void testSkipEverything() {
     @Test
     public void testSkipNothing() {
         Observable<Integer> src = Observable.from(5, 3, 1);
-        Observable.create(skipWhile(src, LESS_THAN_FIVE)).subscribe(w);
+        src.skipWhile(LESS_THAN_FIVE).subscribe(w);
 
         InOrder inOrder = inOrder(w);
         inOrder.verify(w, times(1)).onNext(5);
@@ -94,7 +92,7 @@ public void testSkipNothing() {
     @Test
     public void testSkipSome() {
         Observable<Integer> src = Observable.from(1, 2, 3, 4, 5, 3, 1, 5);
-        Observable.create(skipWhile(src, LESS_THAN_FIVE)).subscribe(w);
+        src.skipWhile(LESS_THAN_FIVE).subscribe(w);
 
         InOrder inOrder = inOrder(w);
         inOrder.verify(w, times(1)).onNext(5);
@@ -108,11 +106,30 @@ public void testSkipSome() {
     @Test
     public void testSkipError() {
         Observable<Integer> src = Observable.from(1, 2, 42, 5, 3, 1);
-        Observable.create(skipWhile(src, LESS_THAN_FIVE)).subscribe(w);
+        src.skipWhile(LESS_THAN_FIVE).subscribe(w);
 
         InOrder inOrder = inOrder(w);
         inOrder.verify(w, never()).onNext(anyInt());
         inOrder.verify(w, never()).onCompleted();
         inOrder.verify(w, times(1)).onError(any(RuntimeException.class));
     }
+    
+    @Test
+    public void testSkipManySubscribers() {
+        Observable<Integer> src = Observable.range(1, 10).skipWhile(LESS_THAN_FIVE);
+        int n = 5;
+        for (int i = 0; i < n; i++) {
+            @SuppressWarnings("unchecked")
+            Observer<Object> o = mock(Observer.class);
+            InOrder inOrder = inOrder(o);
+            
+            src.subscribe(o);
+            
+            for (int j = 5; j < 10; j++) {
+                inOrder.verify(o).onNext(j);
+            } 
+            inOrder.verify(o).onCompleted();
+            verify(o, never()).onError(any(Throwable.class));
+        }
+    }
 }