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

import java.math.BigDecimal;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;

import org.apache.apex.malhar.lib.util.PojoUtils;

import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.DataType;
import com.datastax.driver.core.LocalDate;
import com.datastax.driver.core.TypeCodec;

/**
 * Used to manage simple data type based getters for given cassandra columns
 *
 * @since 3.6.0
 */
public class CassandraPojoUtils
{
  /**
   * Resolves a getter that can be associated with the given field name in the Pojo matching to the given
   * data type of cassandra
   * @param tuplePayloadClass The tuple class that is used to build the getter from
   * @param getterExpr The name of the field representing the getter that needs to be generated
   * @param returnDataTypeOfGetter The Data type of the cassandra column
   * @param userDefinedTypesClass A map that can provide for a UDT class given a column name
   * @return The getter object that can be used to extract the value at runtime
   */
  public static Object resolveGetterForField(Class tuplePayloadClass, String getterExpr,
      DataType returnDataTypeOfGetter, Map<String,Class> userDefinedTypesClass)
  {
    Object getter = null;
    switch (returnDataTypeOfGetter.getName()) {
      case INT:
        getter = PojoUtils.createGetterInt(tuplePayloadClass, getterExpr);
        break;
      case BIGINT:
      case COUNTER:
        getter = PojoUtils.createGetterLong(tuplePayloadClass, getterExpr);
        break;
      case ASCII:
      case TEXT:
      case VARCHAR:
        getter = PojoUtils.createGetter(tuplePayloadClass, getterExpr, String.class);
        break;
      case BOOLEAN:
        getter = PojoUtils.createGetterBoolean(tuplePayloadClass, getterExpr);
        break;
      case FLOAT:
        getter = PojoUtils.createGetterFloat(tuplePayloadClass, getterExpr);
        break;
      case DOUBLE:
        getter = PojoUtils.createGetterDouble(tuplePayloadClass, getterExpr);
        break;
      case DECIMAL:
        getter = PojoUtils.createGetter(tuplePayloadClass, getterExpr, BigDecimal.class);
        break;
      case SET:
        getter = PojoUtils.createGetter(tuplePayloadClass, getterExpr, Set.class);
        break;
      case MAP:
        getter = PojoUtils.createGetter(tuplePayloadClass, getterExpr, Map.class);
        break;
      case LIST:
        getter = PojoUtils.createGetter(tuplePayloadClass, getterExpr, List.class);
        break;
      case TIMESTAMP:
        getter = PojoUtils.createGetter(tuplePayloadClass, getterExpr, Date.class);
        break;
      case UUID:
        getter = PojoUtils.createGetter(tuplePayloadClass, getterExpr, UUID.class);
        break;
      case UDT:
        getter = PojoUtils.createGetter(tuplePayloadClass, getterExpr, userDefinedTypesClass.get(getterExpr));
        break;
      default:
        getter = PojoUtils.createGetter(tuplePayloadClass, getterExpr, Object.class);
        break;
    }
    return getter;
  }

  /**
   * Populates a given bound statement column with a value give a POJO and the map representing the getters
   * @param boundStatement The statement that needs to be populated with the value
   * @param getters A map mapping the applicable getter for a given column name as key
   * @param dataType The data type of the cassandra column name
   * @param cassandraColName The name of the cassandra column
   * @param pojoPayload The POJO from which the value needs to be extracted
   * @param setNulls Whether nulls can be set explicitly
   * @param codecsForCassandraColumnNames A map giving column name to codec with key as column name
   */
  @SuppressWarnings(value = "unchecked")
  public static void populateBoundStatementWithValue(BoundStatement boundStatement, Map<String,Object> getters,
      DataType dataType, String cassandraColName, Object pojoPayload, boolean setNulls,
      Map<String,TypeCodec> codecsForCassandraColumnNames)
  {
    switch (dataType.getName()) {
      case BOOLEAN:
        PojoUtils.GetterBoolean<Object> boolGetter = ((PojoUtils.GetterBoolean<Object>)getters
            .get(cassandraColName));
        if (boolGetter != null) {
          final boolean bool = boolGetter.get(pojoPayload);
          boundStatement.setBool(cassandraColName, bool);
        } else {
          boundStatement.unset(cassandraColName);
        }
        break;
      case INT:
        PojoUtils.GetterInt<Object> inGetter = ((PojoUtils.GetterInt<Object>)getters.get(cassandraColName));
        if (inGetter != null) {
          final int intValue = inGetter.get(pojoPayload);
          boundStatement.setInt(cassandraColName, intValue);
        } else {
          boundStatement.unset(cassandraColName);
        }
        break;
      case BIGINT:
      case COUNTER:
        PojoUtils.GetterLong<Object> longGetter = ((PojoUtils.GetterLong<Object>)getters.get(cassandraColName));
        if (longGetter != null) {
          final long longValue = longGetter.get(pojoPayload);
          boundStatement.setLong(cassandraColName, longValue);
        } else {
          boundStatement.unset(cassandraColName);
        }
        break;
      case FLOAT:
        PojoUtils.GetterFloat<Object> floatGetter = ((PojoUtils.GetterFloat<Object>)getters.get(cassandraColName));
        if (floatGetter != null) {
          final float floatValue = floatGetter.get(pojoPayload);
          boundStatement.setFloat(cassandraColName, floatValue);
        } else {
          boundStatement.unset(cassandraColName);
        }
        break;
      case DOUBLE:
        PojoUtils.GetterDouble<Object> doubleGetter = ((PojoUtils.GetterDouble<Object>)getters
            .get(cassandraColName));
        if (doubleGetter != null) {
          final double doubleValue = doubleGetter.get(pojoPayload);
          boundStatement.setDouble(cassandraColName, doubleValue);
        } else {
          boundStatement.unset(cassandraColName);
        }
        break;
      case DECIMAL:
        PojoUtils.Getter<Object, BigDecimal> bigDecimalGetter = ((PojoUtils.Getter<Object, BigDecimal>)getters
            .get(cassandraColName));
        if (bigDecimalGetter != null) {
          final BigDecimal decimal = bigDecimalGetter.get(pojoPayload);
          if (decimal == null) {
            if (!setNulls) {
              boundStatement.unset(cassandraColName);
            } else {
              boundStatement.setDecimal(cassandraColName, null);
            }
          } else {
            boundStatement.setDecimal(cassandraColName, decimal);
          }
        } else {
          boundStatement.unset(cassandraColName);
        }
        break;
      case UUID:
        PojoUtils.Getter<Object, UUID> uuidGetter = ((PojoUtils.Getter<Object, UUID>)getters.get(cassandraColName));
        if (uuidGetter != null) {
          final UUID uuid = uuidGetter.get(pojoPayload);
          if (uuid == null) {
            if (!setNulls) {
              boundStatement.unset(cassandraColName);
            } else {
              boundStatement.setUUID(cassandraColName, null);
            }
          } else {
            boundStatement.setUUID(cassandraColName, uuid);
          }
        } else {
          boundStatement.unset(cassandraColName);
        }
        break;
      case ASCII:
      case VARCHAR:
      case TEXT:
        PojoUtils.Getter<Object, String> stringGetter = ((PojoUtils.Getter<Object, String>)getters
            .get(cassandraColName));
        if (stringGetter != null) {
          final String ascii = stringGetter.get(pojoPayload);
          if (ascii == null) {
            if (!setNulls) {
              boundStatement.unset(cassandraColName);
            } else {
              boundStatement.setString(cassandraColName, null);
            }
          } else {
            boundStatement.setString(cassandraColName, ascii);
          }
        } else {
          boundStatement.unset(cassandraColName);
        }
        break;
      case SET:
        PojoUtils.Getter<Object, Set<?>> getterForSet = ((PojoUtils.Getter<Object, Set<?>>)getters
            .get(cassandraColName));
        if (getterForSet != null) {
          final Set<?> set = getterForSet.get(pojoPayload);
          if (set == null) {
            if (!setNulls) {
              boundStatement.unset(cassandraColName);
            } else {
              boundStatement.setSet(cassandraColName, null);
            }
          } else {
            boundStatement.setSet(cassandraColName, set);
          }
        } else {
          boundStatement.unset(cassandraColName);
        }
        break;
      case MAP:
        PojoUtils.Getter<Object, Map<?, ?>> mapGetter = ((PojoUtils.Getter<Object, Map<?, ?>>)getters
            .get(cassandraColName));
        if (mapGetter != null) {
          final Map<?, ?> map = mapGetter.get(pojoPayload);
          if (map == null) {
            if (!setNulls) {
              boundStatement.unset(cassandraColName);
            } else {
              boundStatement.setMap(cassandraColName, null);
            }
          } else {
            boundStatement.setMap(cassandraColName, map);
          }
        } else {
          boundStatement.unset(cassandraColName);
        }
        break;
      case LIST:
        PojoUtils.Getter<Object, List<?>> listGetter = ((PojoUtils.Getter<Object, List<?>>)getters
            .get(cassandraColName));
        if (listGetter != null) {
          final List<?> list = listGetter.get(pojoPayload);
          if (list == null) {
            if (!setNulls) {
              boundStatement.unset(cassandraColName);
            } else {
              boundStatement.setList(cassandraColName, null);
            }
          } else {
            boundStatement.setList(cassandraColName, list);
          }
        } else {
          boundStatement.unset(cassandraColName);
        }
        break;
      case TIMESTAMP:
        PojoUtils.Getter<Object, Date> dateGetter = ((PojoUtils.Getter<Object, Date>)getters.get(cassandraColName));
        if (dateGetter != null) {
          final Date date = dateGetter.get(pojoPayload);
          if (date == null) {
            if (!setNulls) {
              boundStatement.unset(cassandraColName);
            } else {
              boundStatement.setMap(cassandraColName, null);
            }
          } else {
            boundStatement.setDate(cassandraColName, LocalDate.fromMillisSinceEpoch(date.getTime()));
          }
        } else {
          boundStatement.unset(cassandraColName);
        }
        break;
      case UDT:
        PojoUtils.Getter<Object, Object> udtGetter = ((PojoUtils.Getter<Object, Object>)getters
            .get(cassandraColName));
        if (udtGetter != null) {
          final Object udtPayload = udtGetter.get(pojoPayload);
          if (udtPayload == null) {
            if (!setNulls) {
              boundStatement.unset(cassandraColName);
            } else {
              boundStatement.setUDTValue(cassandraColName, null);
            }
          } else {
            boundStatement.set(cassandraColName, udtPayload, codecsForCassandraColumnNames
                .get(cassandraColName).getJavaType().getRawType());
          }
        } else {
          boundStatement.unset(cassandraColName);
        }
        break;
      default:
        throw new RuntimeException("Type not supported for " + dataType.getName());
    }
  }
}