package com.github.davidmoten.rx2.internal.flowable;

import java.util.List;
import java.util.function.ToIntFunction;
import java.util.stream.Collectors;

import org.junit.Assert;
import org.junit.Test;

import com.github.davidmoten.rx2.flowable.Transformers;

import io.reactivex.Flowable;
import io.reactivex.functions.Function;

public final class FlowableReduceTest {

    static final Function<List<Integer>, Integer> sum = (new Function<List<Integer>, Integer>() {
        @Override
        public Integer apply(List<Integer> list) throws Exception {
            return list.stream().collect(Collectors.summingInt( //
                    new ToIntFunction<Integer>() {
                @Override
                public int applyAsInt(Integer x) {
                    return x;
                }
            }));
        }
    });

    static final Function<Flowable<Integer>, Flowable<Integer>> reducer = new Function<Flowable<Integer>, Flowable<Integer>>() {

        @Override
        public Flowable<Integer> apply(Flowable<Integer> f) throws Exception {
            return f.buffer(2).map(sum);
        }
    };

    @Test
    public void testEmpty() {
        int result = Flowable.<Integer> empty() //
                .to(Transformers.reduce(reducer, 2)) //
                .blockingGet(-1);
        Assert.assertEquals(-1, result);
    }

    @Test
    public void testOne() {
        check(1, 2);
    }

    @Test
    public void testCompletesFirstLevel() {
        check(2, 2);
    }

    @Test
    public void testCompletesSecondLevel() {
        check(3, 2);
    }

    @Test
    public void testCompletesThirdLevel() {
        check(4, 2);
    }

    @Test(timeout = 200000000)
    public void testCompletesFourLevels() {
        check(8, 2);
    }

    @Test
    public void testCompletesVariableLevelsWithVaryingDepths() {
        for (int maxDepthConcurrent = 1; maxDepthConcurrent < 5; maxDepthConcurrent++) {
            for (int n = 5; n <= 100; n++) {
                check(n, maxDepthConcurrent);
            }
        }
    }

    private static void check(int n, int maxDepthConcurrent) {
        int result = Flowable.range(1, n) //
                .to(Transformers.reduce(reducer, maxDepthConcurrent)) //
                .blockingGet(-1);
        Assert.assertEquals(sum(n), result);
    }

    private static int sum(int n) {
        int sum = 0;
        for (int i = 1; i <= n; i++) {
            sum += i;
        }
        return sum;
    }

}

@@ -323,16 +323,59 @@
 unction;
+%0Aimport io.reactivex.schedulers.Schedulers;
 %0A%0Apublic
@@ -1158,16 +1158,345 @@
     %7D;%0A%0A
+    static final Function%3CFlowable%3CInteger%3E, Flowable%3CInteger%3E%3E reducerAsync = new Function%3CFlowable%3CInteger%3E, Flowable%3CInteger%3E%3E() %7B%0A%0A        @Override%0A        public Flowable%3CInteger%3E apply(Flowable%3CInteger%3E f) throws Exception %7B%0A            return f.subscribeOn(Schedulers.computation()).buffer(2).map(sum);%0A        %7D%0A    %7D;%0A%0A
     @Tes
@@ -2042,29 +2042,8 @@
 Test
-(timeout = 200000000)
 %0A   
@@ -2145,48 +2145,12 @@
 test
-CompletesVariableLevelsWithVaryingDepths
+Many
 () %7B
@@ -2369,159 +2369,701 @@
     
-private static void check(int n, int maxDepthConcurrent) %7B%0A        int result = Flowable.range(1, n) //%0A                .to(Transformers.reduce(reducer
+@Test%0A    public void testManyAsync() %7B%0A        for (int maxDepthConcurrent = 1; maxDepthConcurrent %3C 5; maxDepthConcurrent++) %7B%0A            for (int n = 5; n %3C= 100; n++) %7B%0A                checkAsync(n, maxDepthConcurrent);%0A            %7D%0A        %7D%0A    %7D%0A%0A    %0A    private static void check(int n, int maxDepthConcurrent) %7B%0A        int result = Flowable.range(1, n) //%0A                .to(Transformers.reduce(reducer, maxDepthConcurrent)) //%0A                .blockingGet(-1);%0A        Assert.assertEquals(sum(n), result);%0A    %7D%0A%0A    private static void checkAsync(int n, int maxDepthConcurrent) %7B%0A        int result = Flowable.range(1, n) //%0A                .to(Transformers.reduce(reducerAsync
 , ma

add test