content: /**
   * 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.avro;
  
  import java.util.ArrayList;
  import java.util.List;
  import java.util.StringTokenizer;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import org.apache.apex.malhar.lib.util.FieldInfo;
  import org.apache.apex.malhar.lib.util.FieldInfo.SupportType;
  import org.apache.apex.malhar.lib.util.PojoUtils;
  import org.apache.avro.AvroRuntimeException;
  import org.apache.avro.generic.GenericRecord;
  import org.apache.commons.lang3.ClassUtils;
  import org.apache.hadoop.classification.InterfaceStability;
  
  import com.google.common.annotations.VisibleForTesting;
  import com.google.common.collect.Lists;
  
  import com.datatorrent.api.AutoMetric;
  import com.datatorrent.api.Context;
  import com.datatorrent.api.Context.PortContext;
  import com.datatorrent.api.DefaultInputPort;
  import com.datatorrent.api.DefaultOutputPort;
  import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
  import com.datatorrent.common.util.BaseOperator;
  
  /**
   * <p>
   * AvroToPojo
   * </p>
   * A generic implementation for conversion from Avro to POJO. The POJO class
   * name & field mapping should be provided by the user.<br>
   * If this mapping is not provided then reflection is used to determine this
   * mapping.<br>
   * As of now only primitive types are supported.<br>
   * Error records are emitted on the errorPort if connected
   *
   * @displayName Avro To Pojo
   * @category Converter
   * @tags avro
   *
   * @since 3.4.0
   */
  @InterfaceStability.Evolving
  public class AvroToPojo extends BaseOperator
  {
  
    private transient Class<?> pojoClass;
  
    private static final String FIELD_SEPARATOR = ":";
    private static final String RECORD_SEPARATOR = ",";
  
    private String genericRecordToPOJOFieldsMapping = null;
  
    private transient List<FieldInfo> fieldInfos;
  
    private transient List<ActiveFieldInfo> columnFieldSetters;
  
    @AutoMetric
    @VisibleForTesting
    int recordCount = 0;
  
    @AutoMetric
    @VisibleForTesting
    int errorCount = 0;
  
    @AutoMetric
    @VisibleForTesting
    int fieldErrorCount = 0;
  
    public final transient DefaultOutputPort<GenericRecord> errorPort = new DefaultOutputPort<GenericRecord>();
  
    /**
     * Returns a string representing mapping from generic record to POJO fields
     */
    public String getGenericRecordToPOJOFieldsMapping()
    {
      return genericRecordToPOJOFieldsMapping;
    }
  
    /**
     * Comma separated list mapping a field in Avro schema to POJO field eg :
     * orderId:orderId:INTEGER,total:total:DOUBLE
     */
    public void setGenericRecordToPOJOFieldsMapping(String genericRecordToPOJOFieldsMapping)
    {
      this.genericRecordToPOJOFieldsMapping = genericRecordToPOJOFieldsMapping;
    }
  
    public final transient DefaultInputPort<GenericRecord> data = new DefaultInputPort<GenericRecord>()
    {
      @Override
      public void process(GenericRecord tuple)
      {
        processTuple(tuple);
      }
    };
  
    /**
     * Converts given Generic Record and to a POJO and emits it
     */
    protected void processTuple(GenericRecord tuple)
    {
      try {
        Object obj = getPOJOFromGenericRecord(tuple);
  
        if (obj != null) {
          output.emit(obj);
          recordCount++;
        } else if (errorPort.isConnected()) {
          errorPort.emit(tuple);
          errorCount++;
        }
  
      } catch (InstantiationException | IllegalAccessException e) {
        LOG.error("Could not initialize object of class -" + getClass().getName(), e);
        errorCount++;
      }
    }
  
    /**
     * Returns a POJO from a Generic Record
     *
     * @return Object
     */
    @SuppressWarnings("unchecked")
    private Object getPOJOFromGenericRecord(GenericRecord tuple) throws InstantiationException, IllegalAccessException
    {
      Object newObj = getPojoClass().newInstance();
  
      try {
        for (int i = 0; i < columnFieldSetters.size(); i++) {
  
          AvroToPojo.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;
            fieldErrorCount++;
          }
  
          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 FLOAT:
                ((PojoUtils.SetterFloat<Object>)afi.setterOrGetter).set(newObj,
                    (float)tuple.get(afi.fieldInfo.getColumnName()));
                break;
  
              case INTEGER:
                ((PojoUtils.SetterInt<Object>)afi.setterOrGetter).set(newObj,
                    (int)tuple.get(afi.fieldInfo.getColumnName()));
                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 AvroRuntimeException("Invalid Support Type");
  
            }
          } catch (AvroRuntimeException e) {
            LOG.error("Exception in setting value", e);
            fieldErrorCount++;
          }
  
        }
      } catch (Exception ex) {
        LOG.error("Generic Exception in setting value" + ex.getMessage());
        errorCount++;
        newObj = null;
      }
      return newObj;
    }
  
    /**
     * Use reflection to generate field info values if the user has not provided
     * the inputs mapping
     *
     * @return String representing the POJO field to Avro 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);
    }
  
    /**
     * Creates a map representing fieldName in POJO:field in Generic Record: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;
    }
  
    @OutputPortFieldAnnotation(schemaRequired = true)
    public final transient DefaultOutputPort<Object> output = new DefaultOutputPort<Object>()
    {
      public void setup(PortContext context)
      {
        setPojoClass(context.getValue(Context.PortContext.TUPLE_CLASS));
  
        columnFieldSetters = Lists.newArrayList();
  
        /**
         * Check if the mapping of Generic record fields to POJO is given, else
         * use reflection
         */
        if (getGenericRecordToPOJOFieldsMapping() == null) {
          setFieldInfos(createFieldInfoMap(generateFieldInfoInputs(getPojoClass())));
        } else {
          setFieldInfos(createFieldInfoMap(getGenericRecordToPOJOFieldsMapping()));
        }
  
        initColumnFieldSetters(getFieldInfos());
        initializeActiveFieldSetters();
      }
    };
  
    @Override
    public void endWindow()
    {
      errorCount = 0;
      fieldErrorCount = 0;
      recordCount = 0;
  
    }
  
    private Class<?> getPojoClass()
    {
      return pojoClass;
    }
  
    public void setPojoClass(Class<?> pojoClass)
    {
      this.pojoClass = pojoClass;
    }
  
    /**
     * 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;
    }
  
    /**
     * Add 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 AvroToPojo.ActiveFieldInfo(fi));
      }
    }
  
    /**
     * 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;
    }
  
    /**
     * 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;
      }
    }
  
    private static final Logger LOG = LoggerFactory.getLogger(AvroToPojo.class);
  
  }
  