/**
 * 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.api.impl;

import java.lang.reflect.Field;
import java.lang.reflect.Modifier;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;

import org.joda.time.Duration;

import org.apache.apex.malhar.lib.function.Function;
import org.apache.apex.malhar.lib.function.Function.FlatMapFunction;
import org.apache.apex.malhar.lib.function.FunctionOperator;
import org.apache.apex.malhar.lib.io.ConsoleOutputOperator;
import org.apache.apex.malhar.lib.window.TriggerOption;
import org.apache.apex.malhar.lib.window.WindowOption;

import org.apache.apex.malhar.stream.api.ApexStream;
import org.apache.apex.malhar.stream.api.CompositeStreamTransform;
import org.apache.apex.malhar.stream.api.Option;
import org.apache.apex.malhar.stream.api.WindowedStream;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.hadoop.classification.InterfaceStability;

import com.datatorrent.api.Attribute;
import com.datatorrent.api.Context;
import com.datatorrent.api.DAG;
import com.datatorrent.api.LocalMode;
import com.datatorrent.api.Operator;
import com.datatorrent.stram.StramLocalCluster;
import com.datatorrent.stram.plan.logical.LogicalPlan;

/**
 * Default stream implementation for ApexStream interface. <br>
 * It creates the dag(execution plan) from stream api <br>
 * The dag won't be constructed until {@link #populateDag(DAG)} is called
 *
 * @since 3.4.0
 */
@InterfaceStability.Evolving
public class ApexStreamImpl<T> implements ApexStream<T>
{

  private static Set<Attribute<?>> OPERATOR_ATTRIBUTES;

  private static Set<Attribute<?>> DAG_ATTRIBUTES;

  private static Set<Attribute<?>> INPUT_ATTRIBUTES;

  private static Set<Attribute<?>> OUTPUT_ATTRIBUTES;

  static {

    OPERATOR_ATTRIBUTES = new HashSet<>();
    DAG_ATTRIBUTES = new HashSet<>();
    INPUT_ATTRIBUTES = new HashSet<>();
    OUTPUT_ATTRIBUTES = new HashSet<>();

    try {
      for (Field field : Context.OperatorContext.class.getDeclaredFields()) {
        if (field.getType() == Attribute.class) {
          OPERATOR_ATTRIBUTES.add((Attribute)field.get(Context.OperatorContext.class));
        }
      }

      for (Field field : Context.DAGContext.class.getDeclaredFields()) {
        if (field.getType() == Attribute.class) {
          DAG_ATTRIBUTES.add((Attribute)field.get(Context.DAGContext.class));
        }
      }
    } catch (IllegalAccessException e) {
      //Ignore here
    }

    INPUT_ATTRIBUTES.add(Context.PortContext.PARTITION_PARALLEL);
    INPUT_ATTRIBUTES.add(Context.PortContext.AUTO_RECORD);
    INPUT_ATTRIBUTES.add(Context.PortContext.STREAM_CODEC);
    INPUT_ATTRIBUTES.add(Context.PortContext.TUPLE_CLASS);


    OUTPUT_ATTRIBUTES.add(Context.PortContext.QUEUE_CAPACITY);
    OUTPUT_ATTRIBUTES.add(Context.PortContext.BUFFER_MEMORY_MB);
    OUTPUT_ATTRIBUTES.add(Context.PortContext.SPIN_MILLIS);
    OUTPUT_ATTRIBUTES.add(Context.PortContext.UNIFIER_SINGLE_FINAL);
    OUTPUT_ATTRIBUTES.add(Context.PortContext.IS_OUTPUT_UNIFIED);
    OUTPUT_ATTRIBUTES.add(Context.PortContext.AUTO_RECORD);
    OUTPUT_ATTRIBUTES.add(Context.PortContext.STREAM_CODEC);
    OUTPUT_ATTRIBUTES.add(Context.PortContext.TUPLE_CLASS);

  }

  /**
   * The extension point of the stream
   *
   * @param <T>
   */
  public static class Brick<T>
  {

    private Operator.OutputPort<T> lastOutput;

    private DagMeta.NodeMeta nodeMeta;

    private Pair<Operator.OutputPort, Operator.InputPort> lastStream;

    public Operator.OutputPort<T> getLastOutput()
    {
      return lastOutput;
    }

    public void setLastOutput(Operator.OutputPort<T> lastOutput)
    {
      this.lastOutput = lastOutput;
    }

    public void setLastStream(Pair<Operator.OutputPort, Operator.InputPort> lastStream)
    {
      this.lastStream = lastStream;
    }

    public Pair<Operator.OutputPort, Operator.InputPort> getLastStream()
    {
      return lastStream;
    }
  }


  /**
   * Graph behind the stream
   */
  protected DagMeta graph;

  /**
   * Right now the stream only support single extension point
   * You can have multiple downstream operators connect to this single extension point though
   */
  protected Brick<T> lastBrick;

  public Brick<T> getLastBrick()
  {
    return lastBrick;
  }

  public void setLastBrick(Brick<T> lastBrick)
  {
    this.lastBrick = lastBrick;
  }

  public ApexStreamImpl()
  {
    graph = new DagMeta();
  }

  public ApexStreamImpl(ApexStreamImpl<T> apexStream)
  {
    //copy the variables over to the new ApexStreamImpl
    graph = apexStream.graph;
    lastBrick = apexStream.lastBrick;
  }

  public ApexStreamImpl(DagMeta graph)
  {
    this(graph, null);
  }

  public ApexStreamImpl(DagMeta graph, Brick<T> lastBrick)
  {
    this.graph = graph;
    this.lastBrick = lastBrick;
  }

  @Override
  public <O, STREAM extends ApexStream<O>> STREAM map(Function.MapFunction<T, O> mf, Option... opts)
  {
    FunctionOperator.MapFunctionOperator<T, O> opt = new FunctionOperator.MapFunctionOperator<>(mf);
    return addOperator(opt, opt.input, opt.output, opts);
  }


  @Override
  public <O, STREAM extends ApexStream<O>> STREAM flatMap(FlatMapFunction<T, O> flatten, Option... opts)
  {
    FunctionOperator.FlatMapFunctionOperator<T, O> opt = new FunctionOperator.FlatMapFunctionOperator<>(flatten);
    return addOperator(opt, opt.input, opt.output, opts);
  }

  @Override
  @SuppressWarnings("unchecked")
  public <STREAM extends ApexStream<T>> STREAM filter(final Function.FilterFunction<T> filter, Option... opts)
  {
    FunctionOperator.FilterFunctionOperator<T> filterFunctionOperator = new FunctionOperator.FilterFunctionOperator<>(filter);
    return addOperator(filterFunctionOperator, filterFunctionOperator.input, filterFunctionOperator.output, opts);
  }

  public <STREAM extends ApexStream<Map.Entry<Object, Integer>>> STREAM countByElement()
  {
    return null;
  }

  @Override
  public <O, STREAM extends ApexStream<O>> STREAM endWith(Operator op, Operator.InputPort<T> inputPort, Option... opts)
  {
    return (STREAM)addOperator(op, inputPort, null, opts);
  }

  @Override
  @SuppressWarnings("unchecked")
  public  <O, STREAM extends ApexStream<O>> STREAM addOperator(Operator op, Operator.InputPort<T> inputPort, Operator.OutputPort<O> outputPort, Option... opts)
  {

    checkArguments(op, inputPort, outputPort);

    DagMeta.NodeMeta nm = null;

    if (lastBrick == null) {
      nm = graph.addNode(op, null, null, inputPort, opts);
    } else {
      nm = graph.addNode(op, lastBrick.nodeMeta, lastBrick.lastOutput, inputPort, opts);
    }

    Brick<O> newBrick = new Brick<>();
    newBrick.nodeMeta = nm;
    newBrick.setLastOutput(outputPort);
    if (lastBrick != null) {
      newBrick.lastStream = Pair.<Operator.OutputPort, Operator.InputPort>of(lastBrick.lastOutput, inputPort);
    }

    if (this.getClass() == ApexStreamImpl.class || this.getClass() == ApexWindowedStreamImpl.class) {
      return (STREAM)newStream(this.graph, newBrick);
    } else {
      try {
        return (STREAM)this.getClass().getConstructor(ApexStreamImpl.class).newInstance(newStream(this.graph, newBrick));
      } catch (Exception e) {
        throw new RuntimeException("You have to override the default constructor with ApexStreamImpl as default parameter", e);
      }
    }

  }

  @Override
  public <O, INSTREAM extends ApexStream<T>, OUTSTREAM extends ApexStream<O>> OUTSTREAM addCompositeStreams(CompositeStreamTransform<INSTREAM, OUTSTREAM> compositeStreamTransform)
  {
    return compositeStreamTransform.compose((INSTREAM)this);
  }


  /* Check to see if inputPort and outputPort belongs to the operator */
  private void checkArguments(Operator op, Operator.InputPort inputPort, Operator.OutputPort outputPort)
  {
    if (op == null) {
      throw new IllegalArgumentException("Operator can not be null");
    }

    boolean foundInput = inputPort == null;
    boolean foundOutput = outputPort == null;
    for (Field f : op.getClass().getFields()) {
      int modifiers = f.getModifiers();
      if (!Modifier.isPublic(modifiers) || !Modifier.isTransient(modifiers)) {
        continue;
      }
      Object obj = null;
      try {
        obj = f.get(op);
      } catch (IllegalAccessException e) {
        // NonAccessible field is not a valid port object
      }
      if (obj == outputPort) {
        foundOutput = true;
      }
      if (obj == inputPort) {
        foundInput = true;
      }
    }
    if (!foundInput || !foundOutput) {
      throw new IllegalArgumentException("Input port " + inputPort + " and/or Output port " + outputPort + " is/are not owned by Operator " + op);
    }

  }

  @Override
  public <STREAM extends ApexStream<T>> STREAM union(ApexStream<T>... others)
  {
    throw new UnsupportedOperationException();
  }

  @Override
  @SuppressWarnings("unchecked")
  public ApexStreamImpl<T> print(Option... opts)
  {
    ConsoleOutputOperator consoleOutputOperator = new ConsoleOutputOperator();
    addOperator(consoleOutputOperator, (Operator.InputPort<T>)consoleOutputOperator.input, null, opts);
    return this;
  }

  @Override
  @SuppressWarnings("unchecked")
  public ApexStreamImpl<T> print()
  {
    ConsoleOutputOperator consoleOutputOperator = new ConsoleOutputOperator();
    addOperator(consoleOutputOperator,
        (Operator.InputPort<T>)consoleOutputOperator.input, null, Option.Options.name(IDGenerator.generateOperatorIDWithUUID(consoleOutputOperator.getClass())));
    return this;
  }

  @Override
  @SuppressWarnings("unchecked")
  public ApexStream<T> printErr()
  {
    //TODO need to make ConsoleOutputOperator support stderr
    throw new UnsupportedOperationException();
  }

  @Override
  @SuppressWarnings("unchecked")
  public ApexStream<T> with(Attribute attribute, Object value)
  {
    if (OPERATOR_ATTRIBUTES.contains(attribute)) {
      lastBrick.nodeMeta.operatorAttributes.add(Pair.of(attribute, value));
    }

    if (INPUT_ATTRIBUTES.contains(attribute)) {
      if (lastBrick.lastStream != null) {
        List<Pair<Attribute, Object>> attrs = lastBrick.nodeMeta.inputPortAttributes.get(lastBrick.lastStream.getRight());
        if (attrs == null) {
          attrs = new LinkedList<>();
        }
        attrs.add(Pair.of(attribute, value));
        lastBrick.nodeMeta.inputPortAttributes.put(lastBrick.lastStream.getRight(), attrs);
      }
    }

    if (OUTPUT_ATTRIBUTES.contains(attribute)) {
      if (lastBrick.lastStream != null) {

        for (DagMeta.NodeMeta parent : lastBrick.nodeMeta.getParent()) {
          parent.getNodeStreams().containsKey(lastBrick.lastStream.getLeft());
          List<Pair<Attribute, Object>> attrs = parent.outputPortAttributes.get(lastBrick.lastStream.getLeft());
          if (attrs == null) {
            attrs = new LinkedList<>();
          }
          attrs.add(Pair.of(attribute, value));
          lastBrick.nodeMeta.outputPortAttributes.put(lastBrick.lastStream.getLeft(), attrs);
        }
      }
    }

    setGlobalAttribute(attribute, value);

    return this;
  }

  @Override
  @SuppressWarnings("unchecked")
  public ApexStream<T> setGlobalAttribute(Attribute attribute, Object value)
  {
    graph.dagAttributes.add(Pair.of(attribute, value));
    return this;
  }

  @Override
  @SuppressWarnings("unchecked")
  public ApexStream<T> with(DAG.Locality locality)
  {
    if (lastBrick.lastStream != null) {
      for (DagMeta.NodeMeta parent : lastBrick.nodeMeta.getParent()) {
        Pair<List<Operator.InputPort>, DAG.Locality> p = parent.getNodeStreams().get(lastBrick.lastStream.getLeft());
        if (p != null) {
          p.setValue(locality);
        }
      }
    }
    return this;
  }

  @Override
  @SuppressWarnings("unchecked")
  public ApexStream<T> with(String propName, Object value)
  {
    try {
      BeanUtils.setProperty(lastBrick.nodeMeta.getOperator(), propName, value);
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
    return this;
  }


  @Override
  public DAG createDag()
  {
    LogicalPlan dag = new LogicalPlan();
    populateDag(dag);
    return dag;
  }

  @Override
  public void populateDag(DAG dag)
  {
    graph.buildDAG(dag);
  }

  @Override
  public void runEmbedded(boolean async, long duration, Callable<Boolean> exitCondition)
  {
    LocalMode lma = LocalMode.newInstance();
    populateDag(lma.getDAG());
    DAG dag = lma.getDAG();
    LocalMode.Controller lc = lma.getController();
    if (lc instanceof StramLocalCluster) {
      ((StramLocalCluster)lc).setExitCondition(exitCondition);
    }
    if (async) {
      lc.runAsync();
    } else {
      if (duration >= 0) {
        lc.run(duration);
      } else {
        lc.run();
      }
    }

  }


  @Override
  public void run()
  {
    throw new UnsupportedOperationException();
    //TODO need an api to submit the StreamingApplication to cluster
  }

  @Override
  public WindowedStream<T> window(WindowOption option)
  {
    return window(option, null, null);
  }

  @Override
  public WindowedStream<T> window(WindowOption windowOption, TriggerOption triggerOption)
  {
    return window(windowOption, triggerOption, null);
  }

  @Override
  public WindowedStream<T> window(WindowOption windowOption, TriggerOption triggerOption, Duration allowLateness)
  {
    ApexWindowedStreamImpl<T> windowedStream = new ApexWindowedStreamImpl<>();
    windowedStream.lastBrick = lastBrick;
    windowedStream.graph = graph;
    windowedStream.windowOption = windowOption;
    windowedStream.triggerOption = triggerOption;
    windowedStream.allowedLateness = allowLateness;
    return windowedStream;
  }

  protected <O> ApexStream<O> newStream(DagMeta graph, Brick<O> newBrick)
  {
    ApexStreamImpl<O> newstream = new ApexStreamImpl<>();
    newstream.graph = graph;
    newstream.lastBrick = newBrick;
    return newstream;
  }

}