/**
 * 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 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.computation()).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();
    }

}

@@ -777,20 +777,16 @@
 ubscribe
-Func
 ;%0Aimport
@@ -795,16 +795,76 @@
 x.Observ
+able.OnSubscribeFunc;%0Aimport rx.Observer;%0Aimport rx.Subscrib
 er;%0Aimpo
@@ -954,16 +954,48 @@
 iptions;
+%0Aimport rx.util.functions.Func1;
 %0A%0Apublic
@@ -2029,10 +2029,1175 @@
 %0A    %7D%0A%0A
+    @Test%0A    public void testRepeatTakeWithSubscribeOn() throws InterruptedException %7B%0A%0A        final AtomicInteger counter = new AtomicInteger();%0A        Observable%3CInteger%3E oi = Observable.create(new OnSubscribe%3CInteger%3E() %7B%0A%0A            @Override%0A            public void call(Subscriber%3C? super Integer%3E sub) %7B%0A                System.out.println(%22invoked!%22);%0A                counter.incrementAndGet();%0A                sub.onNext(1);%0A                sub.onNext(2);%0A                sub.onCompleted();%0A            %7D%0A        %7D).subscribeOn(Schedulers.newThread());%0A%0A        Object%5B%5D ys = oi.repeat(Schedulers.newThread()).map(new Func1%3CInteger, Integer%3E() %7B%0A%0A            @Override%0A            public Integer call(Integer t1) %7B%0A                System.out.println(%22t1: %22 + t1);%0A                try %7B%0A                    Thread.sleep(50);%0A                %7D catch (InterruptedException e) %7B%0A                    e.printStackTrace();%0A                %7D%0A                return t1;%0A            %7D%0A%0A        %7D).take(4).toList().toBlockingObservable().last().toArray();%0A%0A        assertEquals(2, counter.get());%0A        assertArrayEquals(new Object%5B%5D %7B 1, 2, 1, 2 %7D, ys);%0A    %7D%0A%0A
 %7D%0A

Test Repeat with SubscribeOn