OperatorRepeat

Author: benjchristensen

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

@@ -69,7 +69,7 @@
 import rx.operators.OperationOnErrorReturn;
 import rx.operators.OperationOnExceptionResumeNextViaObservable;
 import rx.operators.OperationParallelMerge;
-import rx.operators.OperationRepeat;
+import rx.operators.OperatorRepeat;
 import rx.operators.OperationReplay;
 import rx.operators.OperationRetry;
 import rx.operators.OperationSample;
@@ -1616,7 +1616,7 @@ public final static Observable<Long> interval(long interval, TimeUnit unit, Sche
     public final static <T> Observable<T> just(T value) {
         return from(Arrays.asList(value));
     }
-
+    
     /**
      * Returns an Observable that emits a single item and then completes, on a specified scheduler.
      * <p>
@@ -2355,6 +2355,15 @@ public final static <T extends Comparable<? super T>> Observable<T> min(Observab
         return OperationMinMax.min(source);
     }
 
+    /**
+     * Convert the current Observable<T> into an Observable<Observable<T>>.
+     * 
+     * @return
+     */
+    private final Observable<Observable<T>> nest() {
+        return from(this);
+    }
+    
     /**
      * Returns an Observable that never sends any items or notifications to an {@link Observer}.
      * <p>
@@ -5518,7 +5527,7 @@ public final <R> Observable<R> reduce(R initialValue, Func2<R, ? super T, R> acc
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229428.aspx">MSDN: Observable.Repeat</a>
      */
     public final Observable<T> repeat() {
-        return this.repeat(Schedulers.currentThread());
+        return nest().lift(new OperatorRepeat<T>());
     }
 
     /**
@@ -5535,7 +5544,7 @@ public final Observable<T> repeat() {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229428.aspx">MSDN: Observable.Repeat</a>
      */
     public final Observable<T> repeat(Scheduler scheduler) {
-        return create(OperationRepeat.repeat(this, scheduler));
+        return nest().lift(new OperatorRepeat<T>(scheduler));
     }
 
     /**

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

@@ -0,0 +1,87 @@
+/**
+ * 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 rx.Observable;
+import rx.Scheduler;
+import rx.Scheduler.Inner;
+import rx.Subscriber;
+import rx.schedulers.Schedulers;
+import rx.util.functions.Action1;
+
+public class OperatorRepeat<T> implements Operator<T, Observable<T>> {
+
+    private final Scheduler scheduler;
+
+    public OperatorRepeat(Scheduler scheduler) {
+        this.scheduler = scheduler;
+
+    }
+
+    public OperatorRepeat() {
+        this(Schedulers.trampoline());
+    }
+
+    @Override
+    public Subscriber<? super Observable<T>> call(final Subscriber<? super T> child) {
+        return new Subscriber<Observable<T>>(child) {
+
+            @Override
+            public void onCompleted() {
+                // ignore as we will keep repeating
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                child.onError(e);
+            }
+
+            @Override
+            public void onNext(final Observable<T> t) {
+                scheduler.schedule(new Action1<Inner>() {
+
+                    final Action1<Inner> self = this;
+
+                    @Override
+                    public void call(final Inner inner) {
+
+                        t.subscribe(new Subscriber<T>(child) {
+
+                            @Override
+                            public void onCompleted() {
+                                inner.schedule(self);
+                            }
+
+                            @Override
+                            public void onError(Throwable e) {
+                                child.onError(e);
+                            }
+
+                            @Override
+                            public void onNext(T t) {
+                                child.onNext(t);
+                            }
+
+                        });
+                    }
+
+                });
+            }
+
+        };
+    }
+}

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

@@ -31,7 +31,7 @@
 import rx.subscriptions.Subscriptions;
 import rx.util.functions.Func1;
 
-public class OperationRepeatTest {
+public class OperatorRepeatTest {
 
     @Test
     public void testRepetition() {