Operation SkipUntil

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

@@ -76,6 +76,7 @@
 import rx.operators.OperationScan;
 import rx.operators.OperationSkip;
 import rx.operators.OperationSkipLast;
+import rx.operators.OperationSkipUntil;
 import rx.operators.OperationSkipWhile;
 import rx.operators.OperationSubscribeOn;
 import rx.operators.OperationSum;
@@ -6119,4 +6120,18 @@ public <K, V> Observable<Map<K, Collection<V>>> toMultimap(Func1<? super T, ? ex
     public <K, V> Observable<Map<K, Collection<V>>> toMultimap(Func1<? super T, ? extends K> keySelector, Func1<? super T, ? extends V> valueSelector, Func0<? extends Map<K, Collection<V>>> mapFactory, Func1<? super K, ? extends Collection<V>> collectionFactory) {
         return create(OperationToMultimap.toMultimap(this, keySelector, valueSelector, mapFactory, collectionFactory));
     } 
+
+    /**
+     * Return an Observable that skips elements from the source Observable until the secondary
+     * observable emits an element.
+     * 
+     * @param other the other Observable that has to emit an element before this
+     *              Observable's elements are relayed
+     * @return an Observable that skips elements from the source Observable until the secondary
+     *         observable emits an element.
+     * @see <a href='http://msdn.microsoft.com/en-us/library/hh229358.aspx'>MSDN: Observable.SkipUntil</a>
+     */
+    public <U> Observable<T> skipUntil(Observable<U> other) {
+        return create(new OperationSkipUntil<T, U>(this, other));
+    }
 }

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

@@ -0,0 +1,121 @@
+/**
+ * Copyright 2013 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 rx.Observable;
+import rx.Observable.OnSubscribeFunc;
+import rx.Observer;
+import rx.Subscription;
+import rx.subscriptions.CompositeSubscription;
+import rx.subscriptions.SerialSubscription;
+
+/**
+ * Skip elements from the source Observable until the secondary
+ * observable fires an element.
+ * 
+ * @see <a href='http://msdn.microsoft.com/en-us/library/hh229358.aspx'>MSDN: Observable.SkipUntil</a>
+ */
+public class OperationSkipUntil<T, U> implements OnSubscribeFunc<T> {
+    protected final Observable<T> source;
+    protected final Observable<U> other;
+    public OperationSkipUntil(Observable<T> source, Observable<U> other) {
+        this.source = source;
+        this.other = other;
+    }
+
+    @Override
+    public Subscription onSubscribe(Observer<? super T> t1) {
+        return new ResultManager(t1).init();
+    }
+    /** Manage the source and other observers. */
+    private class ResultManager implements Subscription, Observer<T> {
+        final Observer<? super T> observer;
+        final CompositeSubscription cancel;
+        final Object guard = new Object();
+        final AtomicBoolean running = new AtomicBoolean();
+        public ResultManager(Observer<? super T> observer) {
+            this.observer = observer;
+            this.cancel = new CompositeSubscription();
+        }
+        public ResultManager init() {
+
+            SerialSubscription toSource = new SerialSubscription();
+            SerialSubscription toOther = new SerialSubscription();
+
+            cancel.add(toSource);
+            cancel.add(toOther);
+
+            toSource.setSubscription(source.subscribe(this));
+            toOther.setSubscription(other.subscribe(new OtherObserver(toOther)));
+
+            return this;
+        }
+
+        @Override
+        public void unsubscribe() {
+            cancel.unsubscribe();
+        }
+
+        @Override
+        public void onNext(T args) {
+            if (running.get()) {
+                observer.onNext(args);
+            }
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            synchronized (guard) {
+                observer.onError(e);
+                unsubscribe();
+            }
+        }
+
+        @Override
+        public void onCompleted() {
+            synchronized (guard) {
+                observer.onCompleted();
+                unsubscribe();
+            }
+        }
+
+        /** Observe the other stream. */
+        private class OtherObserver implements Observer<U> {
+            final Subscription self;
+            public OtherObserver(Subscription self) {
+                this.self = self;
+            }
+
+            @Override
+            public void onNext(U args) {
+                running.set(true);
+                self.unsubscribe();
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                ResultManager.this.onError(e);
+            }
+
+            @Override
+            public void onCompleted() {
+                self.unsubscribe();
+            }
+
+        }
+    }
+}

rxjava-core/src/test/java/rx/operators/OperationSkipUntilTest.java

@@ -0,0 +1,157 @@
+/**
+ * Copyright 2013 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 org.junit.Before;
+import org.junit.Test;
+import static org.mockito.Matchers.any;
+import org.mockito.Mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import org.mockito.MockitoAnnotations;
+import rx.Observable;
+import rx.Observer;
+import rx.subjects.PublishSubject;
+
+public class OperationSkipUntilTest {
+    @Mock
+    Observer<Object> observer;
+
+    @Before
+    public void before() {
+        MockitoAnnotations.initMocks(this);
+    }
+
+    @Test
+    public void normal1() {
+        PublishSubject<Integer> source = PublishSubject.create();
+        PublishSubject<Integer> other = PublishSubject.create();
+
+        Observable<Integer> m = source.skipUntil(other);
+        m.subscribe(observer);
+
+        source.onNext(0);
+        source.onNext(1);
+
+        other.onNext(100);
+
+        source.onNext(2);
+        source.onNext(3);
+        source.onNext(4);
+        source.onCompleted();
+
+        verify(observer, never()).onError(any(Throwable.class));
+        verify(observer, times(1)).onNext(2);
+        verify(observer, times(1)).onNext(3);
+        verify(observer, times(1)).onNext(4);
+        verify(observer, times(1)).onCompleted();
+    }
+    @Test
+    public void otherNeverFires() {
+        PublishSubject<Integer> source = PublishSubject.create();
+
+        Observable<Integer> m = source.skipUntil(Observable.never());
+
+        m.subscribe(observer);
+
+        source.onNext(0);
+        source.onNext(1);
+        source.onNext(2);
+        source.onNext(3);
+        source.onNext(4);
+        source.onCompleted();
+
+        verify(observer, never()).onError(any(Throwable.class));
+        verify(observer, never()).onNext(any());
+        verify(observer, times(1)).onCompleted();
+    }
+    @Test
+    public void otherEmpty() {
+        PublishSubject<Integer> source = PublishSubject.create();
+
+        Observable<Integer> m = source.skipUntil(Observable.empty());
+
+        m.subscribe(observer);
+
+        verify(observer, never()).onError(any(Throwable.class));
+        verify(observer, never()).onNext(any());
+        verify(observer, never()).onCompleted();
+    }
+    @Test
+    public void otherFiresAndCompletes() {
+        PublishSubject<Integer> source = PublishSubject.create();
+        PublishSubject<Integer> other = PublishSubject.create();
+
+        Observable<Integer> m = source.skipUntil(other);
+        m.subscribe(observer);
+
+        source.onNext(0);
+        source.onNext(1);
+
+        other.onNext(100);
+        other.onCompleted();
+
+        source.onNext(2);
+        source.onNext(3);
+        source.onNext(4);
+        source.onCompleted();
+
+        verify(observer, never()).onError(any(Throwable.class));
+        verify(observer, times(1)).onNext(2);
+        verify(observer, times(1)).onNext(3);
+        verify(observer, times(1)).onNext(4);
+        verify(observer, times(1)).onCompleted();
+    }
+    @Test
+    public void sourceThrows() {
+        PublishSubject<Integer> source = PublishSubject.create();
+        PublishSubject<Integer> other = PublishSubject.create();
+
+        Observable<Integer> m = source.skipUntil(other);
+        m.subscribe(observer);
+
+        source.onNext(0);
+        source.onNext(1);
+
+        other.onNext(100);
+        other.onCompleted();
+
+        source.onNext(2);
+        source.onError(new RuntimeException("Forced failure"));
+
+        verify(observer, times(1)).onNext(2);
+        verify(observer, times(1)).onError(any(Throwable.class));
+        verify(observer, never()).onCompleted();
+    }
+    @Test
+    public void otherThrowsImmediately() {
+        PublishSubject<Integer> source = PublishSubject.create();
+        PublishSubject<Integer> other = PublishSubject.create();
+
+        Observable<Integer> m = source.skipUntil(other);
+        m.subscribe(observer);
+
+        source.onNext(0);
+        source.onNext(1);
+
+        other.onError(new RuntimeException("Forced failure"));
+
+        verify(observer, never()).onNext(any());
+        verify(observer, times(1)).onError(any(Throwable.class));
+        verify(observer, never()).onCompleted();
+    }
+}