package com.github.davidmoten.rx;

import static org.junit.Assert.assertEquals;

import java.util.Arrays;
import java.util.Date;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.junit.Test;

import rx.Observable;
import rx.Scheduler.Worker;
import rx.functions.Action0;
import rx.functions.Action1;
import rx.functions.Func0;
import rx.schedulers.Schedulers;
import rx.schedulers.TestScheduler;

public class ObsTest {

    @Test
    public void testCachedScheduledReset() {
        TestScheduler scheduler = new TestScheduler();
        Worker worker = scheduler.createWorker();
        try {
            final AtomicInteger count = new AtomicInteger(0);
            Observable<Integer> source = Observable.defer(new Func0<Observable<Integer>>() {
                @Override
                public Observable<Integer> call() {
                    return Observable.just(count.incrementAndGet());
                }
            })
                    // cache
                    .compose(Transformers.<Integer> cache(5, TimeUnit.MINUTES, worker));
            assertEquals(1, (int) source.toBlocking().single());
            scheduler.advanceTimeBy(1, TimeUnit.MINUTES);
            assertEquals(1, (int) source.toBlocking().single());
            scheduler.advanceTimeBy(1, TimeUnit.MINUTES);
            assertEquals(1, (int) source.toBlocking().single());
            scheduler.advanceTimeBy(3, TimeUnit.MINUTES);
            assertEquals(2, (int) source.toBlocking().single());
            assertEquals(2, (int) source.toBlocking().single());
        } finally {
            worker.unsubscribe();
        }
    }

    @Test
    public void testCachedResetableScheduledReset() {
        TestScheduler scheduler = new TestScheduler();
        final AtomicInteger count = new AtomicInteger(0);
        Observable<Integer> o = Observable.defer(new Func0<Observable<Integer>>() {
            @Override
            public Observable<Integer> call() {
                return Observable.just(count.incrementAndGet());
            }
        });
        CloseableObservableWithReset<Integer> cached = Obs.cache(o, 5, TimeUnit.MINUTES, scheduler);
        assertEquals(1, (int) cached.observable().toBlocking().single());
        scheduler.advanceTimeBy(1, TimeUnit.MINUTES);
        assertEquals(1, (int) cached.observable().toBlocking().single());
        cached.reset();
        scheduler.advanceTimeBy(5, TimeUnit.MINUTES);
        assertEquals(2, (int) cached.observable().toBlocking().single());
        scheduler.advanceTimeBy(1, TimeUnit.MINUTES);
        assertEquals(2, (int) cached.observable().toBlocking().single());
        cached.reset();
        assertEquals(2, (int) cached.observable().toBlocking().single());
        scheduler.advanceTimeBy(5, TimeUnit.MINUTES);
        assertEquals(3, (int) cached.observable().toBlocking().single());
        cached.close();
    }

    @Test
    public void testRepeating() {
        assertEquals(1000, (int) Obs.repeating(1000).take(2000).toBlocking().last());
    }

    public static void main(String[] args) throws InterruptedException {
        Observable<Date> source = Observable.defer(new Func0<Observable<Date>>() {
            @Override
            public Observable<Date> call() {
                return Observable.just(new Date()).doOnNext(new Action1<Date>() {
                    @Override
                    public void call(Date d) {
                        System.out.println("source emits " + d);
                    }
                });
            }
        });
        final CloseableObservableWithReset<Date> cached = Obs.cache(source, 5, TimeUnit.SECONDS,
                Schedulers.computation());
        Observable<Date> o = cached.observable().doOnSubscribe(new Action0() {
            @Override
            public void call() {
                cached.reset();
            }
        });
        for (int i = 0; i < 30; i++) {
            o.doOnNext(new Action1<Date>() {
                @Override
                public void call(Date t) {
                    System.out.println(t);
                }
            }).subscribe();
            Thread.sleep((i % 5 + 1) * 1000);
        }
        cached.close();
    }

    @Test
    public void testPermutations() {
        Obs.permutations(Observable.range(0,4).toList().toBlocking().single()).forEach(Actions.println());
    }
}

@@ -4286,24 +4286,63 @@
 () %7B%0A       
+ assertEquals(24,%0A                (int)
  Obs.permuta
@@ -4366,16 +4366,17 @@
 range(0,
+ 
 4).toLis
@@ -4405,32 +4405,61 @@
 e())
-.forEach(Actions.println
+%0A                        .count().toBlocking().single
 ());

fix test