package com.twitter.elephantbird.util;

import java.lang.reflect.Field;

import org.apache.hadoop.conf.Configuration;
import org.apache.thrift.TBase;

public class ThriftUtils {

  private static final String CLASS_CONF_PREFIX = "elephantbird.thrift.class.for.";

  public static void setClassConf(Configuration jobConf, Class<?> genericClass,
                                  Class<? extends TBase<?, ?>> thriftClass) {
    String name = CLASS_CONF_PREFIX + genericClass.getName();
    String existingThrift = jobConf.get(name);
    if (existingThrift != null) {
      if (existingThrift != thriftClass.getName()) {
        throw new RuntimeException(
            "Register different thriftClass for the same format class: "
            + genericClass.getName());
      }
    } else {
      jobConf.set(name, thriftClass.getName());
    }
  }


  /**
   * Verify that clazz is a Thrift class. i.e. is a subclass of TBase
   */
  private static void verifyAncestry(Class<?> tClass) {
    if (!TBase.class.isAssignableFrom(tClass)) {
      throw new ClassCastException(tClass.getName() + " is not a Thrift class");
    }
  }

  /**
   * Returns TypeRef for the Thrift class that was set using setClass(jobConf);
   */
  public static<M extends TBase<?, ?>> TypeRef<M> getTypeRef(Configuration jobConf, Class<?> genericClass) {
    String className = jobConf.get(CLASS_CONF_PREFIX + genericClass.getName());
    if (className == null) {
      throw new RuntimeException(CLASS_CONF_PREFIX + genericClass.getName() + " is not set");
    }

    Class<?> tClass = null;
    try {
      tClass = jobConf.getClassByName(className);
    } catch (ClassNotFoundException e) {
      throw new RuntimeException(e);
    }

    verifyAncestry(tClass);

    return new TypeRef<M>(tClass){};
  }

   /**
    * returns TypeRef for thrift class. Verifies that the class is
    * a Thrift class (i.e extends TBase).
    */
  public static <M extends TBase<?, ?>> TypeRef<M> getTypeRef(Class<?> tClass) {
    verifyAncestry(tClass);
    return new TypeRef<M>(tClass){};
  }

  /**
   * returns TypeRef for a thrift class.
   */
  public static <M extends TBase<?, ?>> TypeRef<M> getTypeRef(String thriftClassName, ClassLoader classLoader) {
    try {
      Class<?> tClass = classLoader == null ?
          Class.forName(thriftClassName) :
          Class.forName(thriftClassName, true, classLoader);

      verifyAncestry(tClass);

      return new TypeRef<M>(tClass){};
    } catch (ClassNotFoundException e) {
      throw new RuntimeException(e);
    }
  }

  /**
   * returns TypeRef for a thrift class.
   */
  public static<M extends TBase<?, ?>> TypeRef<M> getTypeRef(String thriftClassName) {
    return getTypeRef(thriftClassName, null);
  }

   /**
   * Returns value of a fieldName in an object.
   */
  public static <M> M getFieldValue(Object containingObject, String fieldName, Class<M> fieldClass) {
    return getFieldValue(containingObject.getClass(), containingObject, fieldName, fieldClass);
  }

  /**
   * Returns value of a static field with given name in containingClass.
   */
  public static <M> M getFieldValue(Class<?> containingClass, String fieldName, Class<M> fieldClass) {
    return getFieldValue(containingClass, null, fieldName, fieldClass);
  }

  private static <M> M getFieldValue(Class<?> containingClass, Object obj, String fieldName, Class<M> fieldClass) {
    try {
      Field field = containingClass.getDeclaredField(fieldName);
      return fieldClass.cast(field.get(obj));
    } catch (Exception e) {
      throw new RuntimeException("while trying to find " + fieldName + " in "
                                  +  containingClass.getName(), e);
    }
  }

  public static Class<?> getFiedlType(Class<?> containingClass, String fieldName) {
    try {
      Field field = containingClass.getDeclaredField(fieldName);
      return field.getType();
    } catch (NoSuchFieldException e) {
      throw new RuntimeException("while trying to find " + fieldName + " in "
                                 + containingClass, e);
    }
  }
}

@@ -560,32 +560,33 @@
 ll) %7B%0A      if (
+!
 existingThrift !
@@ -583,20 +583,24 @@
 ngThrift
- != 
+.equals(
 thriftCl
@@ -613,16 +613,17 @@
 tName())
+)
  %7B%0A     
@@ -670,16 +670,28 @@
    %22
-R
+Already r
 egister
+ed a
  dif
@@ -716,31 +716,8 @@
  for
- the same format class:
  %22%0A 
@@ -747,24 +747,102 @@
 ss.getName()
+%0A            + %22. old: %22 + existingThrift%0A            + %22 new: %22 + thriftClass
 );%0A      %7D%0A 

fix string comparision while setting thriftClass conf in ThriftUtils.