/**
 * 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.contrib.parser;

import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.ListIterator;
import java.util.StringTokenizer;

import org.elasticsearch.common.primitives.Ints;
import org.json.simple.parser.JSONParser;
import org.json.simple.parser.ParseException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.commons.lang3.ClassUtils;
import org.apache.hadoop.classification.InterfaceStability;

import com.google.common.collect.Lists;

import com.datatorrent.api.Context;
import com.datatorrent.api.Context.PortContext;
import com.datatorrent.api.DefaultOutputPort;
import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
import com.datatorrent.lib.parser.Parser;
import com.datatorrent.lib.util.FieldInfo;
import com.datatorrent.lib.util.FieldInfo.SupportType;
import com.datatorrent.lib.util.KeyValPair;
import com.datatorrent.lib.util.PojoUtils;

/**
 * Operator that parses a JSON string tuple and emits a POJO on the output port
 * and tuples that could not be parsed on error port.Upstream operator needs to
 * ensure that a full JSON record is emitted.<br>
 * <b>Properties</b><br>
 * <b>pojoClass</b>:POJO class <br>
 * <b>(optional)fieldMappingString</b>String of format
 * fieldNameInJson:fieldNameInPOJO:DataType<br>
 * <b>Ports</b> <br>
 * <b>in</b>:input tuple as a String. Each tuple represents a json string<br>
 * <b>out</b>:tuples that are validated as per the user defined POJO are emitted
 * as POJO on this port<br>
 * <b>err</b>:tuples that could not be parsed are emitted on this port as
 * KeyValPair<String,String><br>
 * Key being the tuple and Val being the reason
 * 
 * @displayName SimpleStreamingJsonParser
 * @category Parsers
 * @tags json pojo parser streaming
 *
 * @since 3.5.0
 */
@InterfaceStability.Evolving
public class StreamingJsonParser extends Parser<byte[], KeyValPair<String, String>>
{
  private transient JSONParser jsonParser;
  private transient String fieldMappingString;
  private transient List<FieldInfo> fieldInfos;
  private transient List<ActiveFieldInfo> columnFieldSetters;
  protected JsonKeyFinder finder;
  private static final String FIELD_SEPARATOR = ":";
  private static final String RECORD_SEPARATOR = ",";
  private transient ArrayList<String> columnFields;
  private transient Class<?> pojoClass;

  /**
   * @return POJO class
   */
  private Class<?> getPojoClass()
  {
    return pojoClass;
  }

  /**
   * Sets the POJO class
   */
  public void setPojoClass(Class<?> pojoClass)
  {
    this.pojoClass = pojoClass;
  }

  /**
   * Returns a string representing mapping from generic record to POJO fields
   */
  public String getFieldMappingString()
  {
    return fieldMappingString;
  }

  /**
   * Comma separated list mapping a field in JSON schema to POJO field eg :
   * fieldNameInPOJO:fieldNameInJSON:DataType
   */
  public void setFieldMappingString(String pojoFieldsToJsonMapping)
  {
    this.fieldMappingString = pojoFieldsToJsonMapping;
  }

  public StreamingJsonParser()
  {

  }

  @Override
  public void beginWindow(long windowId)
  {
    super.beginWindow(windowId);
  }

  @Override
  public void processTuple(byte[] tuple)
  {
    incomingTuplesCount++;
    Object obj = convert(tuple);
    if (obj != null) {
      output.emit(obj);
      emittedObjectCount++;
    }
  }

  /**
   * Parse an incoming tuple & return a POJO object
   */
  @Override
  public Object convert(byte[] tuple)
  {
    String str;
    if (tuple == null) {
      if (err.isConnected()) {
        err.emit(new KeyValPair<String, String>(null, "null tuple"));
      }
      errorTupleCount++;
      return null;
    }

    try {
      str = new String(tuple, "UTF-8");
    } catch (UnsupportedEncodingException e1) {
      if (err.isConnected()) {
        err.emit(new KeyValPair<String, String>(tuple.toString(), "Encoding not supported"));
      }
      errorTupleCount++;
      LOG.error("Encoding not supported", e1);
      throw new RuntimeException(e1);
    }

    try {
      finder.setKeyCount(0);
      finder.getKeyValMap().clear();
      while (!finder.isEnd()) {
        jsonParser.parse(str, finder, true);
        //stop parsing when the required keyCount is reached
        if (finder.getKeyCount() == columnFields.size()) {
          break;
        }
      }
      jsonParser.reset();
      return setPojoFields(finder.getKeyValMap());
    } catch (ParseException | IllegalAccessException | InstantiationException e) {
      if (err.isConnected()) {
        err.emit(new KeyValPair<String, String>(str, e.getMessage()));
      }
      errorTupleCount++;
      LOG.error("Exception in parsing the record", e);
      throw new RuntimeException(e);
    }

  }

  /**
   * Creates a map representing fieldName in POJO:field in JSON:Data type
   * 
   * @return List of FieldInfo
   */
  private List<FieldInfo> createFieldInfoMap(String str)
  {
    fieldInfos = new ArrayList<FieldInfo>();
    StringTokenizer strtok = new StringTokenizer(str, RECORD_SEPARATOR);

    while (strtok.hasMoreTokens()) {
      String[] token = strtok.nextToken().split(FIELD_SEPARATOR);
      try {
        fieldInfos.add(new FieldInfo(token[0], token[1], SupportType.valueOf(token[2])));
      } catch (Exception e) {
        LOG.error("Invalid support type", e);
      }
    }
    return fieldInfos;
  }

  @Override
  public KeyValPair<String, String> processErrorTuple(byte[] input)
  {
    throw new UnsupportedOperationException("Not supported");
  }

  /**
   * Class that maps fieldInfo to its getters or setters
   */
  protected static class ActiveFieldInfo
  {
    final FieldInfo fieldInfo;
    Object setterOrGetter;

    ActiveFieldInfo(FieldInfo fieldInfo)
    {
      this.fieldInfo = fieldInfo;
    }

  }

  /**
   * A list of {@link FieldInfo}s where each item maps a column name to a pojo
   * field name.
   */
  private List<FieldInfo> getFieldInfos()
  {
    return fieldInfos;
  }

  /**
   * Sets the {@link FieldInfo}s. A {@link FieldInfo} maps a store column to a
   * pojo field name.<br/>
   * The value from fieldInfo.column is assigned to
   * fieldInfo.pojoFieldExpression.
   *
   * @description $[].columnName name of the Output Field in POJO
   * @description $[].pojoFieldExpression expression to get the respective field
   *              from generic record
   * @useSchema $[].pojoFieldExpression outputPort.fields[].name
   */
  private void setFieldInfos(List<FieldInfo> fieldInfos)
  {
    this.fieldInfos = fieldInfos;
  }

  /**
   * Use reflection to generate field info values if the user has not provided
   * the inputs mapping
   * 
   * @return String representing the POJO field to JSON field mapping
   */
  private String generateFieldInfoInputs(Class<?> cls)
  {
    java.lang.reflect.Field[] fields = cls.getDeclaredFields();
    StringBuilder sb = new StringBuilder();
    for (int i = 0; i < fields.length; i++) {
      java.lang.reflect.Field f = fields[i];
      Class<?> c = ClassUtils.primitiveToWrapper(f.getType());
      sb.append(f.getName()).append(FIELD_SEPARATOR).append(f.getName()).append(FIELD_SEPARATOR)
          .append(c.getSimpleName().toUpperCase()).append(RECORD_SEPARATOR);
    }
    return sb.substring(0, sb.length() - 1);
  }

  /**
   * Adds the Active Fields to the columnFieldSetters {@link ActiveFieldInfo}s
   */
  private void initColumnFieldSetters(List<FieldInfo> fieldInfos)
  {
    for (FieldInfo fi : fieldInfos) {
      if (columnFieldSetters == null) {
        columnFieldSetters = Lists.newArrayList();
      }
      columnFieldSetters.add(new StreamingJsonParser.ActiveFieldInfo(fi));
    }
  }

  /**
   * Initialize the setters for generating the POJO
   */
  private void initializeActiveFieldSetters()
  {
    for (int i = 0; i < columnFieldSetters.size(); i++) {
      ActiveFieldInfo activeFieldInfo = columnFieldSetters.get(i);

      SupportType st = activeFieldInfo.fieldInfo.getType();

      switch (st) {
        case BOOLEAN:
          activeFieldInfo.setterOrGetter = PojoUtils.createSetterBoolean(getPojoClass(),
              activeFieldInfo.fieldInfo.getPojoFieldExpression());
          break;
        case DOUBLE:
          activeFieldInfo.setterOrGetter = PojoUtils.createSetterDouble(getPojoClass(),
              activeFieldInfo.fieldInfo.getPojoFieldExpression());
          break;
        case FLOAT:
          activeFieldInfo.setterOrGetter = PojoUtils.createSetterFloat(getPojoClass(),
              activeFieldInfo.fieldInfo.getPojoFieldExpression());
          break;
        case INTEGER:
          activeFieldInfo.setterOrGetter = PojoUtils.createSetterInt(getPojoClass(),
              activeFieldInfo.fieldInfo.getPojoFieldExpression());
          break;
        case STRING:
          activeFieldInfo.setterOrGetter = PojoUtils.createSetter(getPojoClass(),
              activeFieldInfo.fieldInfo.getPojoFieldExpression(), activeFieldInfo.fieldInfo.getType().getJavaType());
          break;
        case LONG:
          activeFieldInfo.setterOrGetter = PojoUtils.createSetterLong(getPojoClass(),
              activeFieldInfo.fieldInfo.getPojoFieldExpression());
          break;
        default:
          activeFieldInfo.setterOrGetter = PojoUtils.createSetter(getPojoClass(),
              activeFieldInfo.fieldInfo.getPojoFieldExpression(), Byte.class);
          break;
      }
      columnFieldSetters.get(i).setterOrGetter = activeFieldInfo.setterOrGetter;
    }
  }

  /**
   * Returns a POJO from a Generic Record Null is set as the default value if a
   * key is not found in the parsed JSON
   * 
   * @return Object
   */
  @SuppressWarnings("unchecked")
  private Object setPojoFields(HashMap<Object, Object> tuple) throws InstantiationException, IllegalAccessException
  {
    Object newObj = getPojoClass().newInstance();
    try {
      for (int i = 0; i < columnFieldSetters.size(); i++) {

        StreamingJsonParser.ActiveFieldInfo afi = columnFieldSetters.get(i);
        SupportType st = afi.fieldInfo.getType();
        Object val = null;

        try {
          val = tuple.get(afi.fieldInfo.getColumnName());
        } catch (Exception e) {
          LOG.error("Could not find field -" + afi.fieldInfo.getColumnName() + "- in the generic record", e);
          val = null;
        }

        //Nothing to set if a value is null
        if (val == null) {
          continue;
        }

        try {
          switch (st) {
            case BOOLEAN:
              ((PojoUtils.SetterBoolean<Object>)afi.setterOrGetter).set(newObj,
                  (boolean)tuple.get(afi.fieldInfo.getColumnName()));
              break;
            case DOUBLE:
              ((PojoUtils.SetterDouble<Object>)afi.setterOrGetter).set(newObj,
                  (double)tuple.get(afi.fieldInfo.getColumnName()));
              break;
            case INTEGER:
              int intVal = Ints.checkedCast((long)tuple.get(afi.fieldInfo.getColumnName()));
              ((PojoUtils.SetterInt<Object>)afi.setterOrGetter).set(newObj, intVal);
              break;
            case STRING:
              ((PojoUtils.Setter<Object, String>)afi.setterOrGetter).set(newObj,
                  new String(tuple.get(afi.fieldInfo.getColumnName()).toString()));
              break;
            case LONG:
              ((PojoUtils.SetterLong<Object>)afi.setterOrGetter).set(newObj,
                  (long)tuple.get(afi.fieldInfo.getColumnName()));
              break;
            default:
              throw new RuntimeException("Invalid Support Type");
          }
        } catch (Exception e) {
          LOG.error("Exception in setting value", e);
          throw new RuntimeException(e);
        }

      }
    } catch (Exception ex) {
      LOG.error("Generic Exception in setting value" + ex.getMessage());
      newObj = null;
    }
    return newObj;
  }

  @OutputPortFieldAnnotation(schemaRequired = true)
  public final transient DefaultOutputPort<Object> output = new DefaultOutputPort<Object>()
  {
    public void setup(PortContext context)
    {
      jsonParser = new JSONParser();
      finder = new JsonKeyFinder();
      columnFields = new ArrayList<String>();
      columnFieldSetters = Lists.newArrayList();

      setPojoClass(context.getValue(Context.PortContext.TUPLE_CLASS));

      if (getFieldMappingString() == null) {
        setFieldInfos(createFieldInfoMap(generateFieldInfoInputs(getPojoClass())));
      } else {
        setFieldInfos(createFieldInfoMap(getFieldMappingString()));
      }
      initColumnFieldSetters(getFieldInfos());
      initializeActiveFieldSetters();

      ListIterator<FieldInfo> itr = fieldInfos.listIterator();
      while (itr.hasNext()) {
        columnFields.add(itr.next().getColumnName());
      }
      finder.setMatchKeyList(columnFields);
    }
  };

  private static final Logger LOG = LoggerFactory.getLogger(StreamingJsonParser.class);

}