/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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 org.apache.apex.malhar.stream.sample.cookbook;

import java.util.ArrayList;
import java.util.List;

import org.apache.apex.malhar.lib.function.Function;
import org.apache.apex.malhar.lib.util.KeyValPair;
import org.apache.apex.malhar.lib.window.TriggerOption;
import org.apache.apex.malhar.lib.window.Tuple;
import org.apache.apex.malhar.lib.window.WindowOption;
import org.apache.apex.malhar.lib.window.accumulation.ReduceFn;
import org.apache.apex.malhar.stream.api.ApexStream;
import org.apache.apex.malhar.stream.api.CompositeStreamTransform;
import org.apache.apex.malhar.stream.api.WindowedStream;
import org.apache.apex.malhar.stream.api.impl.StreamFactory;

import org.apache.hadoop.conf.Configuration;

import com.datatorrent.api.Context;
import com.datatorrent.api.DAG;
import com.datatorrent.api.DefaultInputPort;
import com.datatorrent.api.DefaultOutputPort;
import com.datatorrent.api.InputOperator;
import com.datatorrent.api.StreamingApplication;
import com.datatorrent.api.annotation.ApplicationAnnotation;
import com.datatorrent.common.util.BaseOperator;

import static org.apache.apex.malhar.stream.api.Option.Options.name;

/**
 * An example that reads the public 'Shakespeare' data, and for each word in
 * the dataset that is over a given length, generates a string containing the
 * list of play names in which that word appears
 *
 * <p>Concepts: the combine transform, which lets you combine the values in a
 * key-grouped Collection
 *
 *
 * @since 3.5.0
 */
@ApplicationAnnotation(name = "CombinePerKeyExamples")
public class CombinePerKeyExamples implements StreamingApplication
{
  // Use the shakespeare public BigQuery sample
  private static final String SHAKESPEARE_TABLE = "publicdata:samples.shakespeare";
  // We'll track words >= this word length across all plays in the table.
  private static final int MIN_WORD_LENGTH = 0;

  /**
   * Examines each row in the input table. If the word is greater than or equal to MIN_WORD_LENGTH,
   * outputs word, play_name.
   */
  static class ExtractLargeWordsFn implements Function.MapFunction<SampleBean, KeyValPair<String, String>>
  {

    @Override
    public KeyValPair<String, String> f(SampleBean input)
    {
      String playName = input.getCorpus();
      String word = input.getWord();
      if (word.length() >= MIN_WORD_LENGTH) {
        return new KeyValPair<>(word, playName);
      } else {
        return null;
      }
    }
  }


  /**
   * Prepares the output data which is in same bean
   */
  static class FormatShakespeareOutputFn implements Function.MapFunction<Tuple.WindowedTuple<KeyValPair<String, String>>, SampleBean>
  {
    @Override
    public SampleBean f(Tuple.WindowedTuple<KeyValPair<String, String>> input)
    {
      return new SampleBean(input.getValue().getKey(), input.getValue().getValue());
    }
  }

  /**
   * A reduce function to concat two strings together.
   */
  public static class Concat extends ReduceFn<String>
  {
    @Override
    public String reduce(String input1, String input2)
    {
      return input1 + ", " + input2;
    }
  }

  /**
   * Reads the public 'Shakespeare' data, and for each word in the dataset
   * over a given length, generates a string containing the list of play names
   * in which that word appears.
   */
  private static class PlaysForWord extends CompositeStreamTransform<ApexStream<SampleBean>, WindowedStream<SampleBean>>
  {

    @Override
    public WindowedStream<SampleBean> compose(ApexStream<SampleBean> inputStream)
    {
      return inputStream
          // Extract words from the input SampleBeam stream.
          .map(new ExtractLargeWordsFn(), name("ExtractLargeWordsFn"))

          // Apply window and trigger option to the streams.
          .window(new WindowOption.GlobalWindow(), new TriggerOption().accumulatingFiredPanes().withEarlyFiringsAtEvery(1))

          // Apply reduceByKey transformation to concat the names of all the plays that a word has appeared in together.
          .reduceByKey(new Concat(), new Function.ToKeyValue<KeyValPair<String,String>, String, String>()
          {
            @Override
            public Tuple<KeyValPair<String, String>> f(KeyValPair<String, String> input)
            {
              return new Tuple.PlainTuple<KeyValPair<String, String>>(input);
            }
          }, name("Concat"))

          // Format the output back to a SampleBeam object.
          .map(new FormatShakespeareOutputFn(), name("FormatShakespeareOutputFn"));
    }
  }


  /**
   * A Java Beam class that contains information about a word appears in a corpus written by Shakespeare.
   */
  public static class SampleBean
  {

    public SampleBean()
    {

    }

    public SampleBean(String word, String corpus)
    {
      this.word = word;
      this.corpus = corpus;
    }

    @Override
    public String toString()
    {
      return this.word + " : "  + this.corpus;
    }

    private String word;

    private String corpus;

    public void setWord(String word)
    {
      this.word = word;
    }

    public String getWord()
    {
      return word;
    }

    public void setCorpus(String corpus)
    {
      this.corpus = corpus;
    }

    public String getCorpus()
    {
      return corpus;
    }
  }

  /**
   * A dummy info generator to generate {@link SampleBean} objects to mimic reading from real 'Shakespeare'
   * data.
   */
  public static class SampleInput extends BaseOperator implements InputOperator
  {

    public final transient DefaultOutputPort<SampleBean> beanOutput = new DefaultOutputPort();
    private String[] words = new String[]{"A", "B", "C", "D", "E", "F", "G"};
    private String[] corpuses = new String[]{"1", "2", "3", "4", "5", "6", "7", "8"};
    private static int i;

    @Override
    public void setup(Context.OperatorContext context)
    {
      super.setup(context);
      i = 0;
    }

    @Override
    public void emitTuples()
    {
      while (i < 1) {
        for (String word : words) {
          for (String corpus : corpuses) {
            try {
              Thread.sleep(50);
              beanOutput.emit(new SampleBean(word, corpus));
            } catch (Exception e) {
              // Ignore it
            }
          }
        }
        i++;
      }

    }
  }

  public static class Collector extends BaseOperator
  {
    private static List<SampleBean> result;
    private static boolean done = false;

    public static List<SampleBean> getResult()
    {
      return result;
    }

    public static boolean isDone()
    {
      return done;
    }

    @Override
    public void setup(Context.OperatorContext context)
    {
      result = new ArrayList<>();
      done = false;
    }

    public final transient DefaultInputPort<SampleBean> input = new DefaultInputPort<SampleBean>()
    {
      @Override
      public void process(SampleBean tuple)
      {
        if (tuple.getWord().equals("F")) {
          done = true;
        }
        result.add(tuple);
      }
    };
  }

  /**
   * Populate dag using High-Level API.
   * @param dag
   * @param conf
   */
  @Override
  public void populateDAG(DAG dag, Configuration conf)
  {
    SampleInput input = new SampleInput();
    Collector collector = new Collector();
    StreamFactory.fromInput(input, input.beanOutput, name("input"))
      .addCompositeStreams(new PlaysForWord())
      .print(name("console"))
      .endWith(collector, collector.input, name("Collector"))
      .populateDag(dag);

  }
}