Operator Repeat and other operator fixes

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

@@ -63,13 +63,11 @@
 import rx.operators.OperationMergeDelayError;
 import rx.operators.OperationMinMax;
 import rx.operators.OperationMulticast;
-import rx.operators.OperationObserveOn;
 import rx.operators.OperationOnErrorResumeNextViaFunction;
 import rx.operators.OperationOnErrorResumeNextViaObservable;
 import rx.operators.OperationOnErrorReturn;
 import rx.operators.OperationOnExceptionResumeNextViaObservable;
 import rx.operators.OperationParallelMerge;
-import rx.operators.OperationRepeat;
 import rx.operators.OperationReplay;
 import rx.operators.OperationRetry;
 import rx.operators.OperationSample;
@@ -103,7 +101,9 @@
 import rx.operators.OperatorGroupBy;
 import rx.operators.OperatorMap;
 import rx.operators.OperatorMerge;
+import rx.operators.OperatorObserveOn;
 import rx.operators.OperatorParallel;
+import rx.operators.OperatorRepeat;
 import rx.operators.OperatorTake;
 import rx.operators.OperatorTimestamp;
 import rx.operators.OperatorToObservableList;
@@ -5172,7 +5172,7 @@ public final <R> ConnectableObservable<R> multicast(Subject<? super T, ? extends
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Observable-Utility-Operators#wiki-observeon">RxJava Wiki: observeOn()</a>
      */
     public final Observable<T> observeOn(Scheduler scheduler) {
-        return create(OperationObserveOn.observeOn(this, scheduler));
+        return lift(new OperatorObserveOn<T>(scheduler));
     }
 
     /**
@@ -5549,7 +5549,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 from(this).lift(new OperatorRepeat<T>(-1));
     }
 
     /**
@@ -5566,9 +5566,38 @@ 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 repeat().observeOn(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 from(this).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 repeat(count).observeOn(scheduler);
+    }
+
     /**
      * Returns a {@link ConnectableObservable} that shares a single subscription to the underlying
      * Observable that will replay all of its items and notifications to any future {@link Observer}.

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

@@ -49,7 +49,7 @@ public Subscriber<? super T> call(final Subscriber<? super GroupedObservable<K,
         return new Subscriber<T>(new CompositeSubscription()) {
             private final Map<K, PublishSubject<T>> groups = new HashMap<K, PublishSubject<T>>();
             private final AtomicInteger completionCounter = new AtomicInteger(0);
-
+            private boolean completeOnce;
             @Override
             public void onCompleted() {
                 // if we receive onCompleted from our parent we onComplete children
@@ -59,7 +59,10 @@ public void onCompleted() {
 
                 if (completionCounter.get() == 0) {
                     // special case if no children are running (such as an empty sequence, or just getting the groups and not subscribing)
-                    childObserver.onCompleted();
+                    if (!completeOnce) {
+                        completeOnce = true;
+                        childObserver.onCompleted();
+                    }
                 }
             }
 
@@ -135,7 +138,10 @@ private void completeInner() {
                     for (PublishSubject<T> ps : groups.values()) {
                         ps.onCompleted();
                     }
-                    childObserver.onCompleted();
+                    if (!completeOnce) {
+                        completeOnce = true;
+                        childObserver.onCompleted();
+                    }
                 }
             }
 

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

@@ -0,0 +1,67 @@
+/**
+ * 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.Notification;
+import rx.Scheduler;
+import rx.Subscriber;
+import rx.subscriptions.CompositeSubscription;
+import rx.util.functions.Action0;
+
+/**
+ * Move the observation of events to another thread via Scheduler.
+ * @param <T> the item type
+ */
+public class OperatorObserveOn<T> implements Operator<T, T> {
+    final Scheduler scheduler;
+    public OperatorObserveOn(Scheduler scheduler) {
+        this.scheduler = scheduler;
+    }
+
+    @Override
+    public Subscriber<? super T> call(final Subscriber<? super T> t1) {
+        final QueueDrain qd = new QueueDrain(t1);
+        final CompositeSubscription csub = new CompositeSubscription();
+        t1.add(csub);
+        return new Subscriber<T>(t1) {
+            /** Dispatch the notification value. */
+            void run(final Notification<T> nt) {
+                qd.enqueue(new Action0() {
+                    @Override
+                    public void call() {
+                        nt.accept(t1);
+                    }
+                });
+                qd.tryDrainAsync(scheduler, csub);
+            }
+            @Override
+            public void onNext(final T args) {
+                run(Notification.createOnNext(args));
+            }
+
+            @Override
+            public void onError(final Throwable e) {
+                run(Notification.<T>createOnError(e));
+            }
+
+            @Override
+            public void onCompleted() {
+                run(Notification.<T>createOnCompleted());
+            }
+        };
+    }
+
+}