/*
 * Copyright 2022 Google LLC
 *
 * 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
 *
 *     https://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 basics;

import com.google.common.primitives.Ints;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Combine;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.stream.IntStream;


public class CombinerInterface {
    private static final Logger LOG = LoggerFactory.getLogger(CombinerInterface.class);

    public static class MeanCombine extends Combine.CombineFn<Integer, int[], Float> { //Types: Input, Accum, Output
        @Override
        public int[] createAccumulator() {
            // Called at Combiner creation, once per CombineFn
            int[] accumulator = new int[2];
            return accumulator;
        }

        @Override
        public int[] addInput(int[] accumulator, Integer input) {
            // Called for every element
            accumulator[0] += input; // Add input to running sum
            accumulator[1]++; // Increase counter
            return accumulator;
        }

        @Override
        public int[] mergeAccumulators(Iterable<int[]> accumulators) {
            // Merges the accumulator created across workers
            int[] finalSum = new int[2];
            Iterator<int[]> ite = accumulators.iterator();
            while (ite.hasNext()) {
                int[] val = ite.next();
                finalSum[0] += val[0];
                finalSum[1] += val[1];
            }
            return finalSum;
        }

        @Override
        public Float extractOutput(int[] accumulator) {
            // Takes input from mergeAccumulators and outputs value
            return Float.intBitsToFloat(accumulator[0]) / Float.intBitsToFloat(accumulator[1]);

        }
    }

    public static void main(String[] args) {

        PipelineOptions options = PipelineOptionsFactory.fromArgs(args).withValidation().as(PipelineOptions.class);

        Pipeline p = Pipeline.create(options);

        int[] rangeIntegers = IntStream.range(0, 10000).toArray();
        Iterable<Integer> elements = Ints.asList(rangeIntegers);
        p
                .apply("Create Elements", Create.of(elements))
                .apply(Combine.globally(new MeanCombine()))
                .apply("Log", ParDo.of(new DoFn<Float, Float>() {
                            @ProcessElement
                            public void processElement(@Element Float mean) {
                                LOG.info(mean.toString());
                            }
                        })
                );


        p.run();
    }
}
