Repeat Operator

- merge and slight modification of #518

Author: benjchristensen

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

@@ -73,6 +73,7 @@
 import rx.operators.OperationOnExceptionResumeNextViaObservable;
 import rx.operators.OperationParallel;
 import rx.operators.OperationParallelMerge;
+import rx.operators.OperationRepeat;
 import rx.operators.OperationReplay;
 import rx.operators.OperationRetry;
 import rx.operators.OperationSample;
@@ -1097,6 +1098,28 @@ public static Observable<Integer> range(int start, int count, Scheduler schedule
         return from(Range.createWithCount(start, count), scheduler);
     }
 
+    /**
+     * Repeats the observable sequence indefinitely.
+     * <p>
+     *
+     * @return The observable sequence producing the elements of the given sequence repeatedly and sequentially.
+     * @see <a href="http://msdn.microsoft.com/en-us/library/hh229428(v=vs.103).aspx">MSDN: Observable.Repeat</a>
+     */
+    public Observable<T> repeat() {
+        return this.repeat(Schedulers.currentThread());
+    }
+
+    /**
+     * Repeats the observable sequence indefinitely.
+     * <p>
+     * @param scheduler the scheduler to send the values on.
+     * @return The observable sequence producing the elements of the given sequence repeatedly and sequentially.
+     * @see <a href="http://msdn.microsoft.com/en-us/library/hh229428(v=vs.103).aspx">MSDN: Observable.Repeat</a>
+     */
+    public Observable<T> repeat(Scheduler scheduler) {
+        return create(OperationRepeat.repeat(this, scheduler));
+    }
+
     /**
      * Returns an Observable that calls an Observable factory to create its
      * Observable for each new Observer that subscribes. That is, for each

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

@@ -0,0 +1,71 @@
+/**
+ * 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 rx.Observable;
+import rx.Observer;
+import rx.Scheduler;
+import rx.Subscription;
+import rx.subscriptions.CompositeSubscription;
+import rx.subscriptions.MultipleAssignmentSubscription;
+import rx.util.functions.Action0;
+import rx.util.functions.Action1;
+
+public class OperationRepeat<T> implements Observable.OnSubscribeFunc<T> {
+
+    private final Observable<T> source;
+    private final Scheduler scheduler;
+
+    public static <T> Observable.OnSubscribeFunc<T> repeat(Observable<T> source, Scheduler scheduler) {
+        return new OperationRepeat<T>(source, scheduler);
+    }
+
+    private OperationRepeat(Observable<T> source, Scheduler scheduler) {
+        this.source = source;
+        this.scheduler = scheduler;
+    }
+
+    @Override
+    public Subscription onSubscribe(final Observer<? super T> observer) {
+        final CompositeSubscription compositeSubscription = new CompositeSubscription();
+        final MultipleAssignmentSubscription innerSubscription = new MultipleAssignmentSubscription();
+        compositeSubscription.add(innerSubscription);
+        compositeSubscription.add(scheduler.schedule(new Action1<Action0>() {
+            @Override
+            public void call(final Action0 self) {
+                innerSubscription.set(source.subscribe(new Observer<T>() {
+
+                    @Override
+                    public void onCompleted() {
+                        self.call();
+                    }
+
+                    @Override
+                    public void onError(Throwable error) {
+                        observer.onError(error);
+                    }
+
+                    @Override
+                    public void onNext(T value) {
+                        observer.onNext(value);
+                    }
+                }));
+            }
+        }));
+        return compositeSubscription;
+    }
+}

rxjava-core/src/test/java/rx/ObservableTests.java

@@ -37,11 +37,11 @@
 import org.mockito.stubbing.Answer;
 
 import rx.Observable.OnSubscribeFunc;
-import rx.schedulers.TestScheduler;
 import rx.observables.ConnectableObservable;
+import rx.schedulers.Schedulers;
+import rx.schedulers.TestScheduler;
 import rx.subscriptions.BooleanSubscription;
 import rx.subscriptions.Subscriptions;
-import rx.util.functions.Action0;
 import rx.util.functions.Action1;
 import rx.util.functions.Func0;
 import rx.util.functions.Func1;

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

@@ -0,0 +1,62 @@
+/**
+ * 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 static org.junit.Assert.*;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Test;
+
+import rx.Observable;
+import rx.Observable.OnSubscribeFunc;
+import rx.Observer;
+import rx.Subscription;
+import rx.schedulers.Schedulers;
+import rx.subscriptions.Subscriptions;
+
+public class OperationRepeatTest {
+
+    @Test
+    public void testRepetition() {
+        int NUM = 10;
+        final AtomicInteger count = new AtomicInteger();
+        int value = Observable.create(new OnSubscribeFunc<Integer>() {
+
+            @Override
+            public Subscription onSubscribe(Observer<? super Integer> o) {
+                o.onNext(count.incrementAndGet());
+                o.onCompleted();
+                return Subscriptions.empty();
+            }
+        }).repeat(Schedulers.threadPoolForComputation()).take(NUM).toBlockingObservable().last();
+
+        assertEquals(NUM, value);
+    }
+
+    @Test
+    public void testRepeatTake() {
+        Observable<Integer> xs = Observable.from(1, 2);
+        Object[] ys = xs.repeat(Schedulers.newThread()).take(4).toList().toBlockingObservable().last().toArray();
+        assertArrayEquals(new Object[] { 1, 2, 1, 2 }, ys);
+    }
+
+    @Test
+    public void testNoStackOverFlow() {
+        Observable.from(1).repeat(Schedulers.newThread()).take(100000).toBlockingObservable().last();
+    }
+
+}