Added test

Author: AppliedDuality

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

@@ -969,7 +969,18 @@ public static Observable<Integer> range(int start, int count, Scheduler schedule
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229428(v=vs.103).aspx">MSDN: Observable.Repeat</a>
      */
     public Observable<T> repeat() {
-        return create(rx.operators.OperationRepeat.repeat(this));
+        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));
     }
 
     /**

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

@@ -18,35 +18,36 @@
 
 import rx.Observable;
 import rx.Observer;
+import rx.Scheduler;
 import rx.Subscription;
-import rx.concurrency.Schedulers;
-import rx.subscriptions.SerialSubscription;
+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> seed) {
-        return new OperationRepeat(seed);
+    public static <T> Observable.OnSubscribeFunc<T> repeat(Observable<T> source, Scheduler scheduler) {
+        return new OperationRepeat(source, scheduler);
     }
 
-    private OperationRepeat(Observable<T> source) {
+    private OperationRepeat(Observable<T> source, Scheduler scheduler) {
         this.source = source;
+        this.scheduler = scheduler;
     }
 
     @Override
     public Subscription onSubscribe(final Observer<? super T> observer) {
-        final SerialSubscription subscription = new SerialSubscription();
-        subscription.setSubscription(Schedulers.currentThread().schedule(new Action1<Action0>() {
+        final MultipleAssignmentSubscription subscription = new MultipleAssignmentSubscription();
+        subscription.setSubscription(scheduler.schedule(new Action1<Action0>() {
             @Override
             public void call(final Action0 self) {
                 subscription.setSubscription(source.subscribe(new Observer<T>() {
 
                     @Override
                     public void onCompleted() {
-                        subscription.getSubscription().unsubscribe();
                         self.call();
                     }
 

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

@@ -15,35 +15,36 @@
  */
 package rx;
 
-import static org.junit.Assert.*;
-import static org.mockito.Matchers.*;
-import static org.mockito.Mockito.*;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicReference;
-
 import org.junit.Before;
 import org.junit.Test;
 import org.mockito.InOrder;
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
-
 import rx.Observable.OnSubscribeFunc;
+import rx.concurrency.Schedulers;
 import rx.concurrency.TestScheduler;
 import rx.observables.ConnectableObservable;
 import rx.subscriptions.BooleanSubscription;
 import rx.subscriptions.Subscriptions;
-import rx.util.functions.Action0;
 import rx.util.functions.Action1;
 import rx.util.functions.Func1;
 import rx.util.functions.Func2;
 
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.Assert.*;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyInt;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.*;
+
 public class ObservableTests {
 
     @Mock
@@ -958,4 +959,11 @@ public void testRangeWithScheduler() {
         inOrder.verify(aObserver, times(1)).onCompleted();
         inOrder.verifyNoMoreInteractions();
     }
+
+    @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);
+    }
 }