/*
 * 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.geode.pdx.internal;

import java.io.Externalizable;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.lang.ref.WeakReference;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.util.ArrayList;
import java.util.Date;
import java.util.Enumeration;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.regex.Pattern;

import org.apache.logging.log4j.Logger;

import org.apache.geode.CancelException;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.cache.RegionService;
import org.apache.geode.internal.CopyOnWriteHashSet;
import org.apache.geode.internal.PdxSerializerObject;
import org.apache.geode.internal.util.concurrent.CopyOnWriteWeakHashMap;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.pdx.FieldType;
import org.apache.geode.pdx.NonPortableClassException;
import org.apache.geode.pdx.PdxReader;
import org.apache.geode.pdx.PdxSerializationException;
import org.apache.geode.pdx.PdxWriter;
import org.apache.geode.pdx.ReflectionBasedAutoSerializer;
import org.apache.geode.unsafe.internal.sun.misc.Unsafe;
import org.apache.geode.util.internal.GeodeGlossary;

/**
 * The core of auto serialization which is used in both aspect and reflection-based
 * auto-serialization. This simple manager class is a singleton which tracks the relevant fields for
 * each class which is to be auto-serialized. This class used to be a singleton. But now every
 * instance of ReflectionBasedAutoSerializer will have its own instance of this class. We allow
 * instances of this class to be found so that tests can access internal apis that are not exposed
 * on the public ReflectionBasedAutoSerializer.
 *
 * @since GemFire 6.6
 */

public class AutoSerializableManager {
  private static final Logger logger = LogService.getLogger();

  private static final String INIT_CLASSES_PARAM = "classes";
  private static final String INIT_CHECK_PORTABILITY_PARAM = "check-portability";
  private static final String OPT_IDENTITY = "identity";
  private static final String OPT_EXCLUDE = "exclude";

  /*
   * Map of class and list of fields, we're interested in, for that class.
   */
  private final Map<Class<?>, AutoClassInfo> classMap =
      new CopyOnWriteWeakHashMap<>();

  /*
   * Mapping between class patterns and identity field patterns.
   */
  private final List<String[]> identityPatterns = new CopyOnWriteArrayList<>();

  /*
   * Mapping between class patterns and patterns of fields to exclude
   */
  private final List<String[]> excludePatterns = new CopyOnWriteArrayList<>();

  /*
   * This is an internal parameter which, when set either as a system property or via cache.xml will
   * not evaluate any hardcoded excludes. This helps with testing as well as possibly debugging
   * future customer issues.
   */
  public static final String NO_HARDCODED_EXCLUDES_PARAM =
      GeodeGlossary.GEMFIRE_PREFIX + "auto.serialization.no.hardcoded.excludes";

  private boolean noHardcodedExcludes = Boolean.getBoolean(NO_HARDCODED_EXCLUDES_PARAM);


  /*
   * Holds a set of regex patterns which match the list of classes we're interested in.
   */
  private final Set<Pattern> classPatterns = new LinkedHashSet<>();

  /*
   * Hardcoded set of patterns which we always exclude.
   */
  private final Set<Pattern> hardcodedExclusions = new HashSet<Pattern>() {
    {
      add(Pattern.compile("org\\.apache\\.geode\\..*"));
      add(Pattern.compile("com\\.gemstone\\..*"));
      add(Pattern.compile("java\\..*"));
      add(Pattern.compile("javax\\..*"));
    }
  };

  /*
   * Cache of class names which have been determined to be excluded from serialization. Built up
   * within isRelevant().
   */
  private final Set<String> cachedExcludedClasses = new CopyOnWriteHashSet<>();

  /*
   * Cache of class names which have been determined to be included for serialization. Built up
   * within isRelevant().
   */
  private final Set<String> cachedIncludedClasses = new CopyOnWriteHashSet<>();

  /*
   * Used to hold the class names which have triggered a warning because they have been determined
   * that they should be auto serialized based on a pattern much but were not because that either do
   * not have a public no-arg constructor or have explicit java serialization code.
   */
  private final Set<String> loggedNoAutoSerializeMsg = new CopyOnWriteHashSet<>();


  private final ReflectionBasedAutoSerializer owner;

  public ReflectionBasedAutoSerializer getOwner() {
    return owner;
  }

  public static AutoSerializableManager create(ReflectionBasedAutoSerializer owner,
      boolean checkPortability, String... patterns) {
    AutoSerializableManager result = new AutoSerializableManager(owner);
    result.reconfigure(checkPortability, patterns);
    return result;
  }

  private AutoSerializableManager(ReflectionBasedAutoSerializer owner) {
    this.owner = owner;
  }

  public Map<Class<?>, AutoClassInfo> getClassMap() {
    return classMap;
  }

  private boolean checkPortability;

  public void setCheckPortability(boolean b) {
    checkPortability = b;
  }

  public boolean getCheckPortability() {
    return checkPortability;
  }

  public void resetCachedTypes() {
    classMap.clear();
  }

  public void resetCaches() {
    identityPatterns.clear();
    excludePatterns.clear();
    resetCachedTypes();
  }

  public void resetAll() {
    resetCaches();
    checkPortability = false;
    classPatterns.clear();
    cachedIncludedClasses.clear();
    cachedExcludedClasses.clear();
    loggedNoAutoSerializeMsg.clear();
    noHardcodedExcludes = Boolean.getBoolean(NO_HARDCODED_EXCLUDES_PARAM);
  }

  /*
   * Helper method to determine whether the class of a given object is a class which we are
   * interested in (de)serializing.
   *
   *
   * @return true if the object should be considered for serialization or false otherwise
   */
  private boolean isRelevant(Class<?> clazz) {
    String className = clazz.getName();

    // Do some short-circuiting if possible
    if (cachedIncludedClasses.contains(className)) {
      return true;
    } else if (cachedExcludedClasses.contains(className)) {
      return false;
    }

    boolean result = getOwner().isClassAutoSerialized(clazz);
    if (result) {
      cachedIncludedClasses.add(className);
    } else {
      cachedExcludedClasses.add(className);
    }
    return result;
  }

  /**
   * Determines whether objects of the named class are excluded from auto-serialization. The classes
   * that are excluded are Geode-internal classes and Java classes.
   *
   * @param className Class name including packages.
   * @return <code>true</code> if the named class is excluded.
   */
  public boolean isExcluded(String className) {
    if (!noHardcodedExcludes) {
      for (Pattern p : hardcodedExclusions) {
        if (p.matcher(className).matches()) {
          return true;
        }
      }
    }
    return false;
  }

  public boolean defaultIsClassAutoSerialized(Class<?> clazz) {
    if (clazz.isEnum()) {
      return false;
    }
    String className = clazz.getName();
    if (isExcluded(className)) {
      if (!PdxSerializerObject.class.isAssignableFrom(clazz)) {
        return false;
      }
    }

    for (Pattern p : classPatterns) {
      if (p.matcher(className).matches()) {
        return hasValidConstructor(clazz, p) && !needsStandardSerialization(clazz, p);
      }
    }
    return false;
  }

  /*
   * Helper method to determine whether a class has a default constructor. That's needed so that it
   * can be re-instantiated by PDX on de-serialization.
   *
   */
  private boolean hasValidConstructor(Class<?> clazz, Pattern matchedPattern) {
    if (unsafe != null && !USE_CONSTRUCTOR) {
      // unsafe allows us to create instances without a constructor
      return true;
    }
    try {
      clazz.getConstructor();
      return true;
    } catch (NoSuchMethodException nex) {
      String className = clazz.getName();
      if (!loggedNoAutoSerializeMsg.contains(className)) {
        loggedNoAutoSerializeMsg.add(className);
        logger.warn(
            "Class {} matched with '{}' cannot be auto-serialized due to missing public no-arg constructor. Will attempt using Java serialization.",
            className, matchedPattern.pattern());
      }
      return false;
    }
  }

  private boolean needsStandardSerialization(Class<?> clazz, Pattern matchedPattern) {
    if (Serializable.class.isAssignableFrom(clazz)) {
      if (Externalizable.class.isAssignableFrom(clazz)) {
        String className = clazz.getName();
        if (!loggedNoAutoSerializeMsg.contains(className)) {
          loggedNoAutoSerializeMsg.add(className);
          logger.warn(
              "Class {} matched with '{}' cannot be auto-serialized because it is Externalizable. Java serialization will be used instead of auto-serialization.",
              className, matchedPattern.pattern());
        }
        return true;
      } else {
        if (getPrivateMethod(clazz, "writeObject", new Class[] {ObjectOutputStream.class},
            Void.TYPE)) {
          String className = clazz.getName();
          if (!loggedNoAutoSerializeMsg.contains(className)) {
            loggedNoAutoSerializeMsg.add(className);
            logger.warn(
                "Class {} matched with '{}' cannot be auto-serialized because it has a writeObject(ObjectOutputStream) method. Java serialization will be used instead of auto-serialization.",
                className, matchedPattern.pattern());
          }
          return true;
        } else if (getInheritableMethod(clazz, "writeReplace", null, Object.class)) {
          String className = clazz.getName();
          if (!loggedNoAutoSerializeMsg.contains(className)) {
            loggedNoAutoSerializeMsg.add(className);
            logger.warn(
                "Class {} matched with '{}' cannot be auto-serialized because it has a writeReplace() method. Java serialization will be used instead of auto-serialization.",
                className, matchedPattern.pattern());
          }
          return true;
        }
      }
    }
    return false;
  }

  /**
   * Returns true if a non-private static method with given signature defined by given class, or
   * false if none found.
   */
  private static boolean getPrivateMethod(Class cl, String name, Class[] argTypes,
      Class returnType) {
    try {
      Method meth = cl.getDeclaredMethod(name, argTypes);
      int mods = meth.getModifiers();
      return ((meth.getReturnType() == returnType) && ((mods & Modifier.STATIC) == 0)
          && ((mods & Modifier.PRIVATE) != 0));
    } catch (NoSuchMethodException ex) {
      return false;
    }
  }

  /**
   * Returns true if a non-static, non-abstract method with given signature provided it is defined
   * by or accessible (via inheritance) by the given class, or false if no match found.
   */
  private static boolean getInheritableMethod(Class cl, String name, Class[] argTypes,
      Class returnType) {
    Method meth = null;
    Class defCl = cl;
    while (defCl != null) {
      try {
        meth = defCl.getDeclaredMethod(name, argTypes);
        break;
      } catch (NoSuchMethodException ex) {
        defCl = defCl.getSuperclass();
      }
    }

    if ((meth == null) || (meth.getReturnType() != returnType)) {
      return false;
    }
    int mods = meth.getModifiers();
    if ((mods & (Modifier.STATIC | Modifier.ABSTRACT)) != 0) {
      return false;
    } else if ((mods & (Modifier.PUBLIC | Modifier.PROTECTED)) != 0) {
      return true;
    } else if ((mods & Modifier.PRIVATE) != 0) {
      return (cl == defCl);
    } else {
      return packageEquals(cl, defCl);
    }
  }

  /**
   * Returns true if classes are defined in the same runtime package, false otherwise.
   */
  private static boolean packageEquals(Class cl1, Class cl2) {
    return (cl1.getClassLoader() == cl2.getClassLoader()
        && getPackageName(cl1).equals(getPackageName(cl2)));
  }

  /**
   * Returns package name of given class.
   */
  private static String getPackageName(Class cl) {
    String s = cl.getName();
    int i = s.lastIndexOf('[');
    if (i >= 0) {
      s = s.substring(i + 2);
    }
    i = s.lastIndexOf('.');
    return (i >= 0) ? s.substring(0, i) : "";
  }

  /**
   * Given a class, figure out which fields we're interested in serializing. The class' entire
   * hierarchy will be traversed and used. Transients and statics will be ignored.
   *
   * @param clazz the <code>Class</code> we're interested in
   * @return a list of fields to be used when this class is (de)serialized
   */
  public List<PdxFieldWrapper> getFields(Class<?> clazz) {
    return getClassInfo(clazz).getFields();

  }

  public AutoClassInfo getExistingClassInfo(Class<?> clazz) {
    return classMap.get(clazz);
  }

  public AutoClassInfo getClassInfo(Class<?> clazz) {
    Class<?> tmpClass = clazz;
    AutoClassInfo classInfo = getExistingClassInfo(tmpClass);
    if (classInfo == null) {
      synchronized (classMap) {
        classInfo = classMap.get(tmpClass);
        if (classInfo != null) {
          return classInfo;
        }

        List<PdxFieldWrapper> fieldList = new ArrayList<>();
        List<PdxFieldWrapper> variableLenFields = new ArrayList<>();

        while (tmpClass != Object.class) {
          Field[] fields = tmpClass.getDeclaredFields();
          for (Field f : fields) {
            if (getOwner().isFieldIncluded(f, clazz)) {
              // Should this be reset at some point?
              f.setAccessible(true);
              FieldType ft = getOwner().getFieldType(f, clazz);
              PdxFieldWrapper fw = PdxFieldWrapper.create(this, f, ft,
                  getOwner().getFieldName(f, clazz), getOwner().transformFieldValue(f, clazz),
                  getOwner().isIdentityField(f, clazz));
              if (ft.isFixedWidth()) {
                fieldList.add(fw);
              } else {
                variableLenFields.add(fw);
              }
            }
          }
          tmpClass = tmpClass.getSuperclass();
        }

        fieldList.addAll(variableLenFields);
        classInfo = new AutoClassInfo(clazz, fieldList);
        logger.info("Auto serializer generating type for {} for fields: {}", clazz,
            classInfo.toFormattedString());
        classMap.put(clazz, classInfo);
      } // end sync
    }
    return classInfo;
  }

  public boolean defaultIsIdentityField(Field f, Class<?> clazz) {
    return fieldMatches(f, clazz.getName(), identityPatterns);
  }

  public boolean defaultIsFieldIncluded(Field f, Class<?> clazz) {
    return !Modifier.isTransient(f.getModifiers()) && !Modifier.isStatic(f.getModifiers())
        && !fieldMatches(f, clazz.getName(), excludePatterns);
  }

  public FieldType defaultGetFieldType(Field f, Class<?> clazz) {
    return FieldType.get(f.getType());
  }

  private static class FieldWrapper {
    private final Field field;

    public FieldWrapper(Field f) {
      field = f;
    }

    public Field getField() {
      return field;
    }

    public int getInt(Object o) throws IllegalArgumentException, IllegalAccessException {
      return field.getInt(o);
    }

    public void setInt(Object o, int v) throws IllegalArgumentException, IllegalAccessException {
      field.setInt(o, v);
    }

    public boolean getBoolean(Object o) throws IllegalArgumentException, IllegalAccessException {
      return field.getBoolean(o);
    }

    public void setBoolean(Object o, boolean v)
        throws IllegalArgumentException, IllegalAccessException {
      field.setBoolean(o, v);
    }

    public byte getByte(Object o) throws IllegalArgumentException, IllegalAccessException {
      return field.getByte(o);
    }

    public void setByte(Object o, byte v) throws IllegalArgumentException, IllegalAccessException {
      field.setByte(o, v);
    }

    public short getShort(Object o) throws IllegalArgumentException, IllegalAccessException {
      return field.getShort(o);
    }

    public void setShort(Object o, short v)
        throws IllegalArgumentException, IllegalAccessException {
      field.setShort(o, v);
    }

    public char getChar(Object o) throws IllegalArgumentException, IllegalAccessException {
      return field.getChar(o);
    }

    public void setChar(Object o, char v) throws IllegalArgumentException, IllegalAccessException {
      field.setChar(o, v);
    }

    public long getLong(Object o) throws IllegalArgumentException, IllegalAccessException {
      return field.getLong(o);
    }

    public void setLong(Object o, long v) throws IllegalArgumentException, IllegalAccessException {
      field.setLong(o, v);
    }

    public float getFloat(Object o) throws IllegalArgumentException, IllegalAccessException {
      return field.getFloat(o);
    }

    public void setFloat(Object o, float v)
        throws IllegalArgumentException, IllegalAccessException {
      field.setFloat(o, v);
    }

    public double getDouble(Object o) throws IllegalArgumentException, IllegalAccessException {
      return field.getDouble(o);
    }

    public void setDouble(Object o, double v)
        throws IllegalArgumentException, IllegalAccessException {
      field.setDouble(o, v);
    }

    public Object getObject(Object o) throws IllegalArgumentException, IllegalAccessException {
      return field.get(o);
    }

    public void setObject(Object o, Object v)
        throws IllegalArgumentException, IllegalAccessException {
      field.set(o, v);
    }

    @Override
    public String toString() {
      return field.toString();
    }
  }

  private static class UnsafeFieldWrapper extends FieldWrapper {
    private final long offset;

    public UnsafeFieldWrapper(Field f) {
      super(f);
      offset = unsafe.objectFieldOffset(f);
    }

    @Override
    public int getInt(Object o) throws IllegalArgumentException, IllegalAccessException {
      return unsafe.getInt(o, offset);
    }

    @Override
    public void setInt(Object o, int v) throws IllegalArgumentException, IllegalAccessException {
      unsafe.putInt(o, offset, v);
    }

    @Override
    public boolean getBoolean(Object o) throws IllegalArgumentException, IllegalAccessException {
      return unsafe.getBoolean(o, offset);
    }

    @Override
    public void setBoolean(Object o, boolean v)
        throws IllegalArgumentException, IllegalAccessException {
      unsafe.putBoolean(o, offset, v);
    }

    @Override
    public byte getByte(Object o) throws IllegalArgumentException, IllegalAccessException {
      return unsafe.getByte(o, offset);
    }

    @Override
    public void setByte(Object o, byte v) throws IllegalArgumentException, IllegalAccessException {
      unsafe.putByte(o, offset, v);
    }

    @Override
    public short getShort(Object o) throws IllegalArgumentException, IllegalAccessException {
      return unsafe.getShort(o, offset);
    }

    @Override
    public void setShort(Object o, short v)
        throws IllegalArgumentException, IllegalAccessException {
      unsafe.putShort(o, offset, v);
    }

    @Override
    public char getChar(Object o) throws IllegalArgumentException, IllegalAccessException {
      return unsafe.getChar(o, offset);
    }

    @Override
    public void setChar(Object o, char v) throws IllegalArgumentException, IllegalAccessException {
      unsafe.putChar(o, offset, v);
    }

    @Override
    public long getLong(Object o) throws IllegalArgumentException, IllegalAccessException {
      return unsafe.getLong(o, offset);
    }

    @Override
    public void setLong(Object o, long v) throws IllegalArgumentException, IllegalAccessException {
      unsafe.putLong(o, offset, v);
    }

    @Override
    public float getFloat(Object o) throws IllegalArgumentException, IllegalAccessException {
      return unsafe.getFloat(o, offset);
    }

    @Override
    public void setFloat(Object o, float v)
        throws IllegalArgumentException, IllegalAccessException {
      unsafe.putFloat(o, offset, v);
    }

    @Override
    public double getDouble(Object o) throws IllegalArgumentException, IllegalAccessException {
      return unsafe.getDouble(o, offset);
    }

    @Override
    public void setDouble(Object o, double v)
        throws IllegalArgumentException, IllegalAccessException {
      unsafe.putDouble(o, offset, v);
    }

    @Override
    public Object getObject(Object o) throws IllegalArgumentException, IllegalAccessException {
      return unsafe.getObject(o, offset);
    }

    @Override
    public void setObject(Object o, Object v)
        throws IllegalArgumentException, IllegalAccessException {
      unsafe.putObject(o, offset, v);
    }
  }

  // unsafe will be null if the Unsafe class is not available or SAFE was requested.
  // We attempt to use Unsafe by default for best performance.
  @Immutable
  private static final Unsafe unsafe;
  static {
    Unsafe tmp = null;
    // only use Unsafe if SAFE was not explicitly requested
    if (!Boolean.getBoolean(GeodeGlossary.GEMFIRE_PREFIX + "AutoSerializer.SAFE")) {
      try {
        tmp = new Unsafe();
        // only throw an exception if UNSAFE was explicitly requested
      } catch (RuntimeException ex) {
        if (Boolean.getBoolean(GeodeGlossary.GEMFIRE_PREFIX + "AutoSerializer.UNSAFE")) {
          throw ex;
        }
      } catch (Error ex) {
        if (Boolean.getBoolean(GeodeGlossary.GEMFIRE_PREFIX + "AutoSerializer.UNSAFE")) {
          throw ex;
        }
      }
    }
    unsafe = tmp;
  }

  public abstract static class PdxFieldWrapper {
    private final FieldWrapper field;
    private final String fieldName;
    private final boolean transformValue;
    private final AutoSerializableManager owner;
    private final boolean isIdentityField;

    protected PdxFieldWrapper(AutoSerializableManager owner, Field f, String name,
        boolean transformValue, boolean isIdentityField) {
      FieldWrapper tmp;
      if (unsafe != null) {
        tmp = new UnsafeFieldWrapper(f);
      } else {
        tmp = new FieldWrapper(f);
      }
      field = tmp;
      fieldName = name;
      this.transformValue = transformValue;
      this.owner = owner;
      this.isIdentityField = isIdentityField;
    }

    public static PdxFieldWrapper create(AutoSerializableManager owner, Field f, FieldType ft,
        String name, boolean transformValue, boolean isIdentityField) {
      switch (ft) {
        case INT:
          return new IntField(owner, f, name, transformValue, isIdentityField);
        case BYTE:
          return new ByteField(owner, f, name, transformValue, isIdentityField);
        case LONG:
          return new LongField(owner, f, name, transformValue, isIdentityField);
        case BOOLEAN:
          return new BooleanField(owner, f, name, transformValue, isIdentityField);
        case CHAR:
          return new CharField(owner, f, name, transformValue, isIdentityField);
        case SHORT:
          return new ShortField(owner, f, name, transformValue, isIdentityField);
        case DOUBLE:
          return new DoubleField(owner, f, name, transformValue, isIdentityField);
        case FLOAT:
          return new FloatField(owner, f, name, transformValue, isIdentityField);
        case STRING:
          return new StringField(owner, f, name, transformValue, isIdentityField);
        case DATE:
          return new DateField(owner, f, name, transformValue, isIdentityField);
        case BYTE_ARRAY:
          return new ByteArrayField(owner, f, name, transformValue, isIdentityField);
        case STRING_ARRAY:
          return new StringArrayField(owner, f, name, transformValue, isIdentityField);
        case ARRAY_OF_BYTE_ARRAYS:
          return new ByteArrayArrayField(owner, f, name, transformValue, isIdentityField);
        case BOOLEAN_ARRAY:
          return new BooleanArrayField(owner, f, name, transformValue, isIdentityField);
        case CHAR_ARRAY:
          return new CharArrayField(owner, f, name, transformValue, isIdentityField);
        case SHORT_ARRAY:
          return new ShortArrayField(owner, f, name, transformValue, isIdentityField);
        case INT_ARRAY:
          return new IntArrayField(owner, f, name, transformValue, isIdentityField);
        case LONG_ARRAY:
          return new LongArrayField(owner, f, name, transformValue, isIdentityField);
        case FLOAT_ARRAY:
          return new FloatArrayField(owner, f, name, transformValue, isIdentityField);
        case DOUBLE_ARRAY:
          return new DoubleArrayField(owner, f, name, transformValue, isIdentityField);
        case OBJECT_ARRAY:
          return new ObjectArrayField(owner, f, name, transformValue, isIdentityField);
        case OBJECT:
          return new ObjectField(owner, f, name, transformValue, isIdentityField);
        default:
          throw new IllegalStateException("unhandled field type " + ft);
      }
    }

    public boolean getCheckPortability() {
      return owner.getCheckPortability();
    }

    public Field getField() {
      return field.getField();
    }

    public String getName() {
      return fieldName;
    }

    public boolean transform() {
      return transformValue;
    }

    public abstract void serialize(PdxWriterImpl writer, Object obj, boolean optimizeWrite);

    public abstract void serializeValue(PdxWriterImpl writer, Object newValue,
        boolean optimizeWrite);

    public abstract void deserialize(InternalPdxReader reader, Object obj);

    public abstract void orderedDeserialize(InternalPdxReader reader, Object obj);

    protected Object readTransformIf(Object o, Object serializedValue)
        throws IllegalArgumentException, IllegalAccessException {
      if (!transform()) {
        return serializedValue;
      }
      return readTransform(o, serializedValue);
    }

    protected Object readTransform(Object o, Object serializedValue)
        throws IllegalArgumentException, IllegalAccessException {
      return owner.getOwner().readTransform(getField(), o.getClass(), serializedValue);
    }

    protected void handleException(boolean serialization, Object obj, Exception ex) {
      AutoSerializableManager.handleException(ex, serialization, getName(), obj);
    }

    protected int getInt(Object o) throws IllegalArgumentException, IllegalAccessException {
      return field.getInt(o);
    }

    protected void setInt(Object o, int v) throws IllegalArgumentException, IllegalAccessException {
      field.setInt(o, v);
    }

    protected boolean getBoolean(Object o) throws IllegalArgumentException, IllegalAccessException {
      return field.getBoolean(o);
    }

    protected void setBoolean(Object o, boolean v)
        throws IllegalArgumentException, IllegalAccessException {
      field.setBoolean(o, v);
    }

    protected byte getByte(Object o) throws IllegalArgumentException, IllegalAccessException {
      return field.getByte(o);
    }

    protected void setByte(Object o, byte v)
        throws IllegalArgumentException, IllegalAccessException {
      field.setByte(o, v);
    }

    protected short getShort(Object o) throws IllegalArgumentException, IllegalAccessException {
      return field.getShort(o);
    }

    protected void setShort(Object o, short v)
        throws IllegalArgumentException, IllegalAccessException {
      field.setShort(o, v);
    }

    protected char getChar(Object o) throws IllegalArgumentException, IllegalAccessException {
      return field.getChar(o);
    }

    protected void setChar(Object o, char v)
        throws IllegalArgumentException, IllegalAccessException {
      field.setChar(o, v);
    }

    protected long getLong(Object o) throws IllegalArgumentException, IllegalAccessException {
      return field.getLong(o);
    }

    protected void setLong(Object o, long v)
        throws IllegalArgumentException, IllegalAccessException {
      field.setLong(o, v);
    }

    protected float getFloat(Object o) throws IllegalArgumentException, IllegalAccessException {
      return field.getFloat(o);
    }

    protected void setFloat(Object o, float v)
        throws IllegalArgumentException, IllegalAccessException {
      field.setFloat(o, v);
    }

    protected double getDouble(Object o) throws IllegalArgumentException, IllegalAccessException {
      return field.getDouble(o);
    }

    protected void setDouble(Object o, double v)
        throws IllegalArgumentException, IllegalAccessException {
      field.setDouble(o, v);
    }

    protected Object getObject(Object o) throws IllegalArgumentException, IllegalAccessException {
      return field.getObject(o);
    }

    protected void setObject(Object o, Object v)
        throws IllegalArgumentException, IllegalAccessException {
      field.setObject(o, v);
    }

    public boolean isIdentityField() {
      return isIdentityField;
    }

    @Override
    public String toString() {
      return fieldName + ": " + field;
    }
  }

  public static class IntField extends PdxFieldWrapper {
    public IntField(AutoSerializableManager owner, Field f, String name, boolean transformValue,
        boolean isIdentityField) {
      super(owner, f, name, transformValue, isIdentityField);
    }

    @Override
    public void serialize(PdxWriterImpl writer, Object obj, boolean optimizeWrite) {
      try {
        if (optimizeWrite) {
          writer.writeInt(getInt(obj));
        } else {
          writer.writeInt(getName(), getInt(obj));
        }
      } catch (Exception ex) {
        handleException(true, obj, ex);
      }
    }

    @Override
    public void serializeValue(PdxWriterImpl writer, Object newValue, boolean optimizeWrite) {
      if (optimizeWrite) {
        writer.writeInt((Integer) newValue);
      } else {
        writer.writeInt(getName(), (Integer) newValue);
      }
    }

    @Override
    public void deserialize(InternalPdxReader reader, Object obj) {
      PdxField pf = reader.getPdxField(getName());
      if (pf != null) {
        try {
          if (transform()) {
            setObject(obj, readTransform(obj, reader.readInt(pf)));
          } else {
            setInt(obj, reader.readInt(pf));
          }
        } catch (Exception ex) {
          handleException(false, obj, ex);
        }
      }
    }

    @Override
    public void orderedDeserialize(InternalPdxReader reader, Object obj) {
      try {
        if (transform()) {
          setObject(obj, readTransform(obj, reader.readInt()));
        } else {
          setInt(obj, reader.readInt());
        }
      } catch (Exception ex) {
        handleException(false, obj, ex);
      }
    }
  }
  public static class ByteField extends PdxFieldWrapper {
    public ByteField(AutoSerializableManager owner, Field f, String name, boolean transformValue,
        boolean isIdentityField) {
      super(owner, f, name, transformValue, isIdentityField);
    }

    @Override
    public void serialize(PdxWriterImpl writer, Object obj, boolean optimizeWrite) {
      try {
        if (optimizeWrite) {
          writer.writeByte(getByte(obj));
        } else {
          writer.writeByte(getName(), getByte(obj));
        }
      } catch (Exception ex) {
        handleException(true, obj, ex);
      }
    }

    @Override
    public void serializeValue(PdxWriterImpl writer, Object newValue, boolean optimizeWrite) {
      if (optimizeWrite) {
        writer.writeByte((Byte) newValue);
      } else {
        writer.writeByte(getName(), (Byte) newValue);
      }
    }

    @Override
    public void deserialize(InternalPdxReader reader, Object obj) {
      PdxField pf = reader.getPdxField(getName());
      if (pf != null) {
        try {
