/*
 * Copyright 2016-2017 David Karnok
 *
 * 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 hu.akarnokd.rxjava2.parallel;

import java.io.IOException;

import org.junit.Test;

import hu.akarnokd.rxjava2.test.TestHelper;
import io.reactivex.Flowable;
import io.reactivex.functions.Function;

public class ParallelSumTest {

    @Test
    public void normalInteger() {
        Flowable.range(1, 5)
        .parallel(1)
        .compose(ParallelTransformers.<Integer>sumInteger())
        .sequential()
        .test()
        .assertResult(15);
    }

    @Test
    public void normalLong() {
        Flowable.rangeLong(1, 5)
        .parallel(1)
        .compose(ParallelTransformers.<Long>sumLong())
        .sequential()
        .test()
        .assertResult(15L);
    }

    @Test
    public void normalDouble() {
        Flowable.range(1, 5)
        .map(new Function<Integer, Double>() {
            @Override
            public Double apply(Integer v) throws Exception {
                return v + 0.5d;
            }
        })
        .parallel(1)
        .compose(ParallelTransformers.<Double>sumDouble())
        .sequential()
        .test()
        .assertResult(17.5d);
    }

    @Test
    public void emptyInteger() {
        Flowable.<Integer>empty()
        .parallel(1)
        .compose(ParallelTransformers.<Integer>sumInteger())
        .sequential()
        .test()
        .assertResult();
    }

    @Test
    public void errorInteger() {
        Flowable.<Integer>error(new IOException())
        .parallel(1)
        .compose(ParallelTransformers.<Integer>sumInteger())
        .sequential()
        .test()
        .assertFailure(IOException.class);
    }

    @Test
    public void emptyLong() {
        Flowable.<Long>empty()
        .parallel(1)
        .compose(ParallelTransformers.<Long>sumLong())
        .sequential()
        .test()
        .assertResult();
    }

    @Test
    public void errorLong() {
        Flowable.<Long>error(new IOException())
        .parallel(1)
        .compose(ParallelTransformers.<Long>sumLong())
        .sequential()
        .test()
        .assertFailure(IOException.class);
    }

    @Test
    public void emptyDouble() {
        Flowable.<Double>empty()
        .parallel(1)
        .compose(ParallelTransformers.<Double>sumDouble())
        .sequential()
        .test()
        .assertResult();
    }

    @Test
    public void errorDouble() {
        Flowable.<Double>error(new IOException())
        .parallel(1)
        .compose(ParallelTransformers.<Double>sumDouble())
        .sequential()
        .test()
        .assertFailure(IOException.class);
    }

    @Test
    public void invalidRailsInteger() {
        TestHelper.checkInvalidParallelSubscribers(Flowable.range(1, 5)
                .parallel(1)
                .compose(ParallelTransformers.<Integer>sumInteger()));
    }

    @Test
    public void invalidRailsLong() {
        TestHelper.checkInvalidParallelSubscribers(Flowable.range(1, 5)
                .parallel(1)
                .compose(ParallelTransformers.<Integer>sumLong()));
    }

    @Test
    public void invalidRailsDouble() {
        TestHelper.checkInvalidParallelSubscribers(Flowable.range(1, 5)
                .parallel(1)
                .compose(ParallelTransformers.<Integer>sumDouble()));
    }

}

@@ -680,20 +680,17 @@
 g.junit.
-Test
+*
 ;%0A%0Aimpor
@@ -788,24 +788,60 @@
 nctions.
-Function
+*;%0Aimport io.reactivex.schedulers.Schedulers
 ;%0A%0Apubli
@@ -3859,10 +3859,623 @@
 %0A    %7D%0A%0A
+    @Test%0A    public void sumALot() %7B%0A        Long%5B%5D array = new Long%5B1000 * 1000%5D;%0A        for (int i = 0; i %3C array.length; i++) %7B%0A            array%5Bi%5D = (long)i;%0A        %7D%0A%0A        long n = Flowable.fromArray(array)%0A        .parallel()%0A        .runOn(Schedulers.computation())%0A        .compose(ParallelTransformers.%3CLong%3EsumLong())%0A        .reduce(new BiFunction%3CLong, Long, Long%3E() %7B%0A            @Override%0A            public Long apply(Long a, Long b) throws Exception %7B%0A                return a + b;%0A            %7D%0A        %7D)%0A        .blockingLast();%0A%0A        Assert.assertEquals(999999L * 500000L, n);%0A    %7D%0A
 %7D%0A

Add a sum (0..1M] test