Merge pull request #828 from benjchristensen/repeat-operator

Repeat Operator

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,40 @@ 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));
+    }
+
+    /**
+     * Returns an Observable that repeats the sequence of items emitted by the source
+     * Observable at most count times.
+     * 
+     * @param count
+     *            the number of times the source Observable items are repeated,
+     *            a count of 0 will yield an empty sequence.
+     * @return an Observable that repeats the sequence of items emitted by the source
+     *         Observable at most count times.
+     */
+    public final Observable<T> repeat(long count) {
+        if (count < 0) {
+            throw new IllegalArgumentException("count >= 0 expected");
+        }
+        return nest().lift(new OperatorRepeat<T>(count));
+    }
+
+    /**
+     * Returns an Observable that repeats the sequence of items emitted by the source
+     * Observable at most count times on a particular scheduler.
+     * 
+     * @param count
+     *            the number of times the source Observable items are repeated,
+     *            a count of 0 will yield an empty sequence.
+     * @param scheduler
+     *            the scheduler to emit the items on
+     * @return an Observable that repeats the sequence of items emitted by the source
+     *         Observable at most count times on a particular scheduler.
+     */
+    public final Observable<T> repeat(long count, Scheduler scheduler) {
+        return nest().lift(new OperatorRepeat<T>(count, scheduler));
     }
 
     /**

rxjava-core/src/main/java/rx/observers/Observers.java

@@ -1,34 +1,35 @@
 package rx.observers;
 
 import rx.Observer;
+import rx.Subscriber;
 import rx.util.OnErrorNotImplementedException;
 import rx.util.functions.Action0;
 import rx.util.functions.Action1;
 
 public class Observers {
 
-    /**
-     * Create an empty Observer that ignores all events.
-     */
-    public static final <T> Observer<T> create() {
-        return new Observer<T>() {
+    private static final Observer<Object> EMPTY = new Observer<Object>() {
 
-            @Override
-            public final void onCompleted() {
-                // do nothing
-            }
+        @Override
+        public final void onCompleted() {
+            // do nothing
+        }
 
-            @Override
-            public final void onError(Throwable e) {
-                throw new OnErrorNotImplementedException(e);
-            }
+        @Override
+        public final void onError(Throwable e) {
+            throw new OnErrorNotImplementedException(e);
+        }
 
-            @Override
-            public final void onNext(T args) {
-                // do nothing
-            }
+        @Override
+        public final void onNext(Object args) {
+            // do nothing
+        }
 
-        };
+    };
+
+    @SuppressWarnings("unchecked")
+    public static <T> Observer<T> empty() {
+        return (Observer<T>) EMPTY;
     }
 
     /**

rxjava-core/src/main/java/rx/observers/Subscribers.java

@@ -8,6 +8,30 @@
 
 public class Subscribers {
 
+    private static final Subscriber<Object> EMPTY = new Subscriber<Object>() {
+
+        @Override
+        public final void onCompleted() {
+            // do nothing
+        }
+
+        @Override
+        public final void onError(Throwable e) {
+            throw new OnErrorNotImplementedException(e);
+        }
+
+        @Override
+        public final void onNext(Object args) {
+            // do nothing
+        }
+
+    };
+
+    @SuppressWarnings("unchecked")
+    public static <T> Subscriber<T> empty() {
+        return (Subscriber<T>) EMPTY;
+    }
+
     public static <T> Subscriber<T> from(final Observer<? super T> o) {
         return new Subscriber<T>() {
 
@@ -29,30 +53,6 @@ public void onNext(T t) {
         };
     }
 
-    /**
-     * Create an empty Subscriber that ignores all events.
-     */
-    public static final <T> Subscriber<T> create() {
-        return new Subscriber<T>() {
-
-            @Override
-            public final void onCompleted() {
-                // do nothing
-            }
-
-            @Override
-            public final void onError(Throwable e) {
-                throw new OnErrorNotImplementedException(e);
-            }
-
-            @Override
-            public final void onNext(T args) {
-                // do nothing
-            }
-
-        };
-    }
-
     /**
      * Create an Subscriber that receives `onNext` and ignores `onError` and `onCompleted`.
      */

rxjava-core/src/main/java/rx/observers/TestObserver.java

@@ -28,7 +28,6 @@
  */
 public class TestObserver<T> implements Observer<T> {
 
-    private final Observer<Object> EMPTY = new EmptyObserver<Object>();
 
     private final Observer<T> delegate;
     private final ArrayList<T> onNextEvents = new ArrayList<T>();
@@ -39,9 +38,8 @@ public TestObserver(Observer<T> delegate) {
         this.delegate = delegate;
     }
 
-    @SuppressWarnings("unchecked")
     public TestObserver() {
-        this.delegate = (Observer<T>) EMPTY;
+        this.delegate = Observers.empty();
     }
 
     @Override

rxjava-core/src/main/java/rx/observers/TestSubscriber.java

@@ -26,8 +26,6 @@
  */
 public class TestSubscriber<T> extends Subscriber<T> {
 
-    private final Subscriber<Object> EMPTY = Subscribers.create();
-
     private final TestObserver<T> testObserver;
 
     public TestSubscriber(Subscriber<T> delegate) {
@@ -38,9 +36,8 @@ public TestSubscriber(Observer<T> delegate) {
         this.testObserver = new TestObserver<T>(delegate);
     }
 
-    @SuppressWarnings("unchecked")
     public TestSubscriber() {
-        this.testObserver = new TestObserver<T>((Subscriber<T>) EMPTY);
+        this.testObserver = new TestObserver<T>(Subscribers.<T>empty());
     }
 
     @Override