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

import static org.apache.geode.internal.serialization.filter.SanctionedSerializables.loadSanctionedClassNames;
import static org.apache.geode.internal.serialization.filter.SanctionedSerializables.loadSanctionedSerializablesServices;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.EOFException;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.NotSerializableException;
import java.io.ObjectInput;
import java.io.ObjectInputStream;
import java.io.ObjectOutput;
import java.io.ObjectOutputStream;
import java.io.ObjectStreamClass;
import java.io.OutputStream;
import java.io.Serializable;
import java.io.UTFDataFormatException;
import java.lang.ref.WeakReference;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Modifier;
import java.lang.reflect.Proxy;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.net.InetAddress;
import java.net.SocketException;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.IdentityHashMap;
import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.Properties;
import java.util.ServiceLoader;
import java.util.Set;
import java.util.Stack;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.UUID;
import java.util.Vector;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.Logger;
import org.jetbrains.annotations.TestOnly;

import org.apache.geode.CancelException;
import org.apache.geode.CanonicalInstantiator;
import org.apache.geode.DataSerializable;
import org.apache.geode.DataSerializer;
import org.apache.geode.GemFireIOException;
import org.apache.geode.GemFireRethrowable;
import org.apache.geode.Instantiator;
import org.apache.geode.SerializationException;
import org.apache.geode.SystemFailure;
import org.apache.geode.ToDataException;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.annotations.internal.MakeNotStatic;
import org.apache.geode.annotations.internal.MutableForTesting;
import org.apache.geode.cache.CacheClosedException;
import org.apache.geode.cache.execute.Function;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.DMStats;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.LonerDistributionManager;
import org.apache.geode.distributed.internal.SerialDistributionMessage;
import org.apache.geode.internal.cache.EnumListenerEvent;
import org.apache.geode.internal.cache.EventID;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.PoolManagerImpl;
import org.apache.geode.internal.cache.tier.sockets.CacheClientNotifier;
import org.apache.geode.internal.cache.tier.sockets.CacheServerHelper;
import org.apache.geode.internal.cache.tier.sockets.ClientDataSerializerMessage;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.internal.cache.tier.sockets.OldClientSupportService;
import org.apache.geode.internal.cache.tier.sockets.Part;
import org.apache.geode.internal.classloader.ClassPathLoader;
import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.internal.serialization.BasicSerializable;
import org.apache.geode.internal.serialization.DSCODE;
import org.apache.geode.internal.serialization.DSFIDSerializer;
import org.apache.geode.internal.serialization.DSFIDSerializerFactory;
import org.apache.geode.internal.serialization.DataSerializableFixedID;
import org.apache.geode.internal.serialization.DataSerializableFixedIdRegistrant;
import org.apache.geode.internal.serialization.DeserializationContext;
import org.apache.geode.internal.serialization.DscodeHelper;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.internal.serialization.ObjectDeserializer;
import org.apache.geode.internal.serialization.ObjectSerializer;
import org.apache.geode.internal.serialization.SerializationContext;
import org.apache.geode.internal.serialization.SerializationVersions;
import org.apache.geode.internal.serialization.StaticSerialization;
import org.apache.geode.internal.serialization.VersionedDataStream;
import org.apache.geode.internal.serialization.filter.NullStreamSerialFilter;
import org.apache.geode.internal.serialization.filter.ReflectiveFacadeStreamSerialFilterFactory;
import org.apache.geode.internal.serialization.filter.SanctionedSerializablesService;
import org.apache.geode.internal.serialization.filter.SerializableObjectConfig;
import org.apache.geode.internal.serialization.filter.StreamSerialFilter;
import org.apache.geode.internal.serialization.filter.StreamSerialFilterFactory;
import org.apache.geode.internal.serialization.filter.UnableToSetSerialFilterException;
import org.apache.geode.internal.util.concurrent.CopyOnWriteHashMap;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.pdx.NonPortableClassException;
import org.apache.geode.pdx.PdxInstance;
import org.apache.geode.pdx.PdxSerializable;
import org.apache.geode.pdx.PdxSerializer;
import org.apache.geode.pdx.internal.AutoSerializableManager;
import org.apache.geode.pdx.internal.AutoSerializableManager.AutoClassInfo;
import org.apache.geode.pdx.internal.EnumInfo;
import org.apache.geode.pdx.internal.PdxInputStream;
import org.apache.geode.pdx.internal.PdxInstanceEnum;
import org.apache.geode.pdx.internal.PdxInstanceImpl;
import org.apache.geode.pdx.internal.PdxOutputStream;
import org.apache.geode.pdx.internal.PdxReaderImpl;
import org.apache.geode.pdx.internal.PdxType;
import org.apache.geode.pdx.internal.PdxWriterImpl;
import org.apache.geode.pdx.internal.TypeRegistry;
import org.apache.geode.util.internal.GeodeGlossary;

/**
 * Contains static methods for data serializing instances of internal GemFire classes. It also
 * contains the implementation of the distribution messaging (and shared memory management) needed
 * to support data serialization.
 *
 * @since GemFire 3.5
 */
public abstract class InternalDataSerializer extends DataSerializer {
  public static final boolean LOAD_CLASS_EACH_TIME =
      Boolean.getBoolean(GeodeGlossary.GEMFIRE_PREFIX + "loadClassOnEveryDeserialization");
  private static final Logger logger = LogService.getLogger();
  /**
   * Maps Class names to their DataSerializer. This is used to find a DataSerializer during
   * serialization.
   */
  @MakeNotStatic
  private static final Map<String, DataSerializer> classesToSerializers = new ConcurrentHashMap<>();

  /**
   * This list contains classes that Geode's classes subclass, such as antlr AST classes which are
   * used by our Object Query Language. It also contains certain classes that are DataSerializable
   * but end up being serialized as part of other serializable objects. VersionedObjectList, for
   * instance, is serialized as part of a partial putAll exception object.
   *
   * <p>
   * Do not java-serialize objects that Geode does not have complete control over. This leaves us
   * open to security attacks such as Gadget Chains and compromises the ability to do a rolling
   * upgrade from one version of Geode to the next.
   *
   * <p>
   * In general, you shouldn't use java serialization, and you should implement
   * DataSerializableFixedID for internal Geode objects. This gives you better control over
   * backward-compatibility.
   *
   * <p>
   * Do not add to this list unless absolutely necessary. Instead, put your classes either in the
   * sanctionedSerializables file for your module or in its excludedClasses file. Run
   * AnalyzeSerializables to generate the content for the file.
   *
   * <p>
   * Syntax is documented by the javadocs of {@code ObjectInputFilter.Config.createFilter}. In
   * Java 8, {@code ObjectInputFilter} is in package {@code sun.misc}. In Java 9 and above, it's
   * in package {@code java.io}.
   */
  private static final String SANCTIONED_SERIALIZABLES_DEPENDENCIES_PATTERN =
      // Java
      "java.**"

          // JMX
          + ";javax.management.**"

          // used for some old enums
          + ";javax.print.attribute.EnumSyntax"

          // Query AST objects
          + ";antlr.**"

          // old Admin API and old JMX
          + ";org.apache.commons.modeler.AttributeInfo"
          + ";org.apache.commons.modeler.FeatureInfo"
          + ";org.apache.commons.modeler.ManagedBean"
          + ";org.apache.geode.distributed.internal.DistributionConfigSnapshot"
          + ";org.apache.geode.distributed.internal.RuntimeDistributionConfigImpl"
          + ";org.apache.geode.distributed.internal.DistributionConfigImpl"

          // WindowedExportFunction and RegionSnapshotService
          + ";org.apache.geode.distributed.internal.membership.InternalDistributedMember"

          // putAll
          + ";org.apache.geode.internal.cache.persistence.PersistentMemberID"
          + ";org.apache.geode.internal.cache.persistence.DiskStoreID"
          + ";org.apache.geode.internal.cache.tier.sockets.VersionedObjectList"

          // security services
          + ";org.apache.shiro.**"

          // export logs
          + ";org.apache.logging.log4j.Level"
          + ";org.apache.logging.log4j.spi.StandardLevel"

          // jar deployment
          + ";com.sun.proxy.$Proxy*"
          + ";com.healthmarketscience.rmiio.RemoteInputStream"
          + ";javax.rmi.ssl.SslRMIClientSocketFactory"
          + ";javax.net.ssl.SSLHandshakeException"
          + ";javax.net.ssl.SSLException"
          + ";sun.security.validator.ValidatorException"
          + ";sun.security.provider.certpath.SunCertPathBuilderException"

          // geode-modules
          + ";org.apache.geode.modules.util.SessionCustomExpiry"
          + ";";

  private static final String serializationVersionTxt =
      System.getProperty(GeodeGlossary.GEMFIRE_PREFIX + "serializationVersion");
  /**
   * Change this constant to be the last one in SERIALIZATION_VERSION
   */
  @Immutable
  private static final SERIALIZATION_VERSION latestVersion = SERIALIZATION_VERSION.v662;
  @Immutable
  private static final SERIALIZATION_VERSION serializationVersion = calculateSerializationVersion();
  /**
   * Maps the id of a serializer to its {@code DataSerializer}.
   */
  @MakeNotStatic
  private static final ConcurrentMap<Integer, Object /* DataSerializer|Marker */> idsToSerializers =
      new ConcurrentHashMap<>();
  /**
   * Contains the classnames of the data serializers (and not the supported classes) not yet loaded
   * into the vm as keys and their corresponding holder instances as values.
   */
  @MakeNotStatic
  private static final ConcurrentHashMap<String, SerializerAttributesHolder> dsClassesToHolders =
      new ConcurrentHashMap<>();
  /**
   * Contains the id of the data serializers not yet loaded into the vm as keys and their
   * corresponding holder instances as values.
   */
  @MakeNotStatic
  private static final ConcurrentHashMap<Integer, SerializerAttributesHolder> idsToHolders =
      new ConcurrentHashMap<>();
  /**
   * Contains the classnames of supported classes as keys and their corresponding
   * SerializerAttributesHolder instances as values. This applies only to the data serializers which
   * have not been loaded into the vm.
   */
  @MakeNotStatic
  private static final ConcurrentHashMap<String, SerializerAttributesHolder> supportedClassesToHolders =
      new ConcurrentHashMap<>();
  private static final Object listenersSync = new Object();

  private static final ThreadLocal<Boolean> pdxSerializationInProgress = new ThreadLocal<>();
  @MakeNotStatic
  private static final CopyOnWriteHashMap<String, WeakReference<Class<?>>> classCache =
      LOAD_CLASS_EACH_TIME ? null : new CopyOnWriteHashMap<>();
  private static final Object cacheAccessLock = new Object();

  private static final String POST_GEODE_190_SERVER_CQIMPL =
      "org.apache.geode.cache.query.cq.internal.ServerCQImpl";
  private static final String PRE_GEODE_190_SERVER_CQIMPL =
      "org.apache.geode.cache.query.internal.cq.ServerCQImpl";

  @Immutable
  private static final StreamSerialFilter defaultSerializationFilter = new NullStreamSerialFilter();
  /**
   * A deserialization filter for ObjectInputStreams
   */
  @MakeNotStatic
  private static StreamSerialFilter serializationFilter = defaultSerializationFilter;
  /**
   * support for old GemFire clients and WAN sites - needed to enable moving from GemFire to Geode
   */
  @MakeNotStatic
  private static OldClientSupportService oldClientSupportService;

  @MakeNotStatic
  private static final DSFIDSerializer dsfidSerializer;

  @MakeNotStatic
  private static final DSFIDFactory dsfidFactory;

  /**
   * {@code RegistrationListener}s that receive callbacks when {@code DataSerializer}s and {@code
   * Instantiator}s are registered. Note: copy-on-write access used for this set
   */
  @MakeNotStatic
  private static volatile Set<RegistrationListener> listeners = new HashSet<>();

  static {
    dsfidSerializer = new DSFIDSerializerFactory().setObjectSerializer(new ObjectSerializer() {
      @Override
      public void writeObject(Object obj, DataOutput output) throws IOException {
        InternalDataSerializer.writeObject(obj, output);
      }

      @Override
      public void invokeToData(Object ds, DataOutput out) throws IOException {
        InternalDataSerializer.invokeToData(ds, out);
      }

    }).setObjectDeserializer(new ObjectDeserializer() {
      @Override
      public <T> T readObject(DataInput input) throws IOException, ClassNotFoundException {
        return InternalDataSerializer.readObject(input);
      }

      @Override
      public void invokeFromData(Object ds, DataInput in)
          throws IOException, ClassNotFoundException {
        InternalDataSerializer.invokeFromData(ds, in);
      }
    }).create();
    initializeWellKnownSerializers();
    dsfidFactory = new DSFIDFactory(dsfidSerializer);

    ServiceLoader<DataSerializableFixedIdRegistrant> loaders = ServiceLoader.load(
        DataSerializableFixedIdRegistrant.class);
    for (DataSerializableFixedIdRegistrant loader : loaders) {
      try {
        loader.register(dsfidSerializer);
      } catch (Exception ex) {
        logger.warn("Data serializable fixed ID loader '{}' failed",
            loader.getClass().getName(), ex);
      }
    }
  }

  /**
   * For backward compatibility we must swizzle the package of some classes that had to be moved
   * when GemFire was open- sourced. This preserves backward-compatibility.
   *
   * @param nameArg the fully qualified class name
   * @return the name of the class in this implementation
   */
  public static String processIncomingClassName(String nameArg) {
    final String name = StaticSerialization.processIncomingClassName(nameArg);
    // using identity comparison on purpose because we are on the hot path
    if (name != nameArg) {
      return name;
    }
    if (name.equals(PRE_GEODE_190_SERVER_CQIMPL)) {
      return POST_GEODE_190_SERVER_CQIMPL;
    }
    OldClientSupportService svc = getOldClientSupportService();
    if (svc != null) {
      return svc.processIncomingClassName(name);
    }
    return name;
  }

  /**
   * For backward compatibility we must swizzle the package of some classes that had to be moved
   * when GemFire was open- sourced. This preserves backward-compatibility.
   *
   * @param nameArg the fully qualified class name
   * @param out the consumer of the serialized object
   * @return the name of the class in this implementation
   */
  public static String processOutgoingClassName(final String nameArg, DataOutput out) {

    final String name = StaticSerialization.processOutgoingClassName(nameArg);
    // using identity comparison on purpose because we are on the hot path
    if (name != nameArg) {
      return name;
    }

    if (out instanceof VersionedDataStream) {
      VersionedDataStream vout = (VersionedDataStream) out;
      KnownVersion version = vout.getVersion();
      if (null != version) {
        if (version.isOlderThan(KnownVersion.GEODE_1_9_0)) {
          if (name.equals(POST_GEODE_190_SERVER_CQIMPL)) {
            return PRE_GEODE_190_SERVER_CQIMPL;
          }
        }
      }
    }
    OldClientSupportService svc = getOldClientSupportService();
    if (svc != null) {
      return svc.processOutgoingClassName(name, out);
    }
    return name;
  }

  /**
   * Initializes the optional serialization "accept list" if the user has requested it in the
   * DistributionConfig
   *
   * @param config the DistributedSystem configuration
   */
  public static void initializeSerializationFilter(SerializableObjectConfig config) {
    initializeSerializationFilter(config, loadSanctionedSerializablesServices());
  }

  /**
   * Initializes the optional serialization "accept list" if the user has requested it in the
   * DistributionConfig
   *
   * @param config the DistributedSystem configuration
   * @param services SanctionedSerializablesService that might have classes to acceptlist
   */
  @VisibleForTesting
  public static void initializeSerializationFilter(SerializableObjectConfig config,
      Collection<SanctionedSerializablesService> services) {
    logger.info("initializing InternalDataSerializer with {} services", services.size());

    StreamSerialFilterFactory objectInputFilterFactory =
        new ReflectiveFacadeStreamSerialFilterFactory();

    serializationFilter =
        objectInputFilterFactory.create(config, loadSanctionedClassNames(services));
  }

  @TestOnly
  static void clearSerializationFilter() {
    serializationFilter = defaultSerializationFilter;
  }

  private static SERIALIZATION_VERSION calculateSerializationVersion() {
    if (serializationVersionTxt == null || serializationVersionTxt.isEmpty()) {
      return latestVersion;
    } else if (serializationVersionTxt.startsWith("6.6.0")
        || serializationVersionTxt.startsWith("6.6.1")) {
      return SERIALIZATION_VERSION.v660;
    } else if (serializationVersionTxt.startsWith("6.6.2")) {
      return SERIALIZATION_VERSION.v662;
    } else {
      return SERIALIZATION_VERSION.vINVALID;
    }
  }

  public static boolean is662SerializationEnabled() {
    return serializationVersion.ordinal() >= SERIALIZATION_VERSION.v662.ordinal();
  }

  public static void checkSerializationVersion() {
    if (serializationVersion == SERIALIZATION_VERSION.vINVALID) {
      throw new IllegalArgumentException(
          "The system property \"gemfire.serializationVersion\" was set to \""
              + serializationVersionTxt
              + "\" which is not a valid serialization version. Valid versions must start with \"6.6.0\", \"6.6.1\", or \"6.6.2\"");
    }
  }

  private static void initializeWellKnownSerializers() {
    // ArrayBlockingQueue does not have zero-arg constructor
    // LinkedBlockingQueue does have zero-arg constructor but no way to get capacity

    classesToSerializers.put("java.lang.String", new WellKnownPdxDS() {
      @Override
      public boolean toData(Object o, DataOutput out) throws IOException {
        try {
          writeString((String) o, out);
        } catch (UTFDataFormatException ex) {
          // See bug 30428
          String s = "While writing a String of length " + ((String) o).length();
          UTFDataFormatException ex2 = new UTFDataFormatException(s);
          ex2.initCause(ex);
          throw ex2;
        }
        return true;
      }
    });
    classesToSerializers.put("java.net.InetAddress", new WellKnownDS() {
      @Override
      public boolean toData(Object o, DataOutput out) throws IOException {
        InetAddress address = (InetAddress) o;
        out.writeByte(DSCODE.INET_ADDRESS.toByte());
        writeInetAddress(address, out);
        return true;
      }
    });
    classesToSerializers.put("java.net.Inet4Address", new WellKnownDS() {
      @Override
      public boolean toData(Object o, DataOutput out) throws IOException {
        InetAddress address = (InetAddress) o;
        out.writeByte(DSCODE.INET_ADDRESS.toByte());
        writeInetAddress(address, out);
        return true;
      }
    });
    classesToSerializers.put("java.net.Inet6Address", new WellKnownDS() {
      @Override
      public boolean toData(Object o, DataOutput out) throws IOException {
        InetAddress address = (InetAddress) o;
        out.writeByte(DSCODE.INET_ADDRESS.toByte());
        writeInetAddress(address, out);
        return true;
      }
    });
    classesToSerializers.put("java.lang.Class", new WellKnownDS() {
      @Override
      public boolean toData(Object o, DataOutput out) throws IOException {
        Class c = (Class) o;
        if (c.isPrimitive()) {
          StaticSerialization.writePrimitiveClass(c, out);
        } else {
          out.writeByte(DSCODE.CLASS.toByte());
          writeClass(c, out);
        }
        return true;
      }
    });
    classesToSerializers.put("java.lang.Boolean", new WellKnownPdxDS() {
      @Override
      public boolean toData(Object o, DataOutput out) throws IOException {
        Boolean value = (Boolean) o;
        out.writeByte(DSCODE.BOOLEAN.toByte());
        writeBoolean(value, out);
        return true;
      }
    });
    classesToSerializers.put("java.lang.Character", new WellKnownPdxDS() {
      @Override
      public boolean toData(Object o, DataOutput out) throws IOException {
        Character value = (Character) o;
        out.writeByte(DSCODE.CHARACTER.toByte());
        writeCharacter(value, out);
        return true;
      }
    });
    classesToSerializers.put("java.lang.Byte", new WellKnownPdxDS() {
      @Override
      public boolean toData(Object o, DataOutput out) throws IOException {
        Byte value = (Byte) o;
        out.writeByte(DSCODE.BYTE.toByte());
        writeByte(value, out);
        return true;
      }
    });
    classesToSerializers.put("java.lang.Short", new WellKnownPdxDS() {
      @Override
      public boolean toData(Object o, DataOutput out) throws IOException {
        Short value = (Short) o;
        out.writeByte(DSCODE.SHORT.toByte());
        writeShort(value, out);
        return true;
      }
    });
    classesToSerializers.put("java.lang.Integer", new WellKnownPdxDS() {
      @Override
      public boolean toData(Object o, DataOutput out) throws IOException {
        Integer value = (Integer) o;
        out.writeByte(DSCODE.INTEGER.toByte());
        writeInteger(value, out);
        return true;
      }
    });
    classesToSerializers.put("java.lang.Long", new WellKnownPdxDS() {
      @Override
      public boolean toData(Object o, DataOutput out) throws IOException {
        Long value = (Long) o;
        out.writeByte(DSCODE.LONG.toByte());
        writeLong(value, out);
        return true;
      }
    });
    classesToSerializers.put("java.lang.Float", new WellKnownPdxDS() {
      @Override
      public boolean toData(Object o, DataOutput out) throws IOException {
        Float value = (Float) o;
        out.writeByte(DSCODE.FLOAT.toByte());
        writeFloat(value, out);
        return true;
      }
    });
    classesToSerializers.put("java.lang.Double", new WellKnownPdxDS() {
      @Override
      public boolean toData(Object o, DataOutput out) throws IOException {
        Double value = (Double) o;
        out.writeByte(DSCODE.DOUBLE.toByte());
        writeDouble(value, out);
        return true;
      }
    });
    classesToSerializers.put("[Z", // boolean[]
        new WellKnownPdxDS() {
          @Override
          public boolean toData(Object o, DataOutput out) throws IOException {
            out.writeByte(DSCODE.BOOLEAN_ARRAY.toByte());
            writeBooleanArray((boolean[]) o, out);
            return true;
          }
        });
    classesToSerializers.put("[B", // byte[]
        new WellKnownPdxDS() {
          @Override
          public boolean toData(Object o, DataOutput out) throws IOException {
            byte[] array = (byte[]) o;
            out.writeByte(DSCODE.BYTE_ARRAY.toByte());
            writeByteArray(array, out);
            return true;
          }
        });
    classesToSerializers.put("[C", // char[]
        new WellKnownPdxDS() {
          @Override
          public boolean toData(Object o, DataOutput out) throws IOException {
            out.writeByte(DSCODE.CHAR_ARRAY.toByte());
            writeCharArray((char[]) o, out);
            return true;
          }
        });
    classesToSerializers.put("[D", // double[]
        new WellKnownPdxDS() {
          @Override
          public boolean toData(Object o, DataOutput out) throws IOException {
            double[] array = (double[]) o;
            out.writeByte(DSCODE.DOUBLE_ARRAY.toByte());
            writeDoubleArray(array, out);
            return true;
          }
        });
    classesToSerializers.put("[F", // float[]
        new WellKnownPdxDS() {
          @Override
          public boolean toData(Object o, DataOutput out) throws IOException {
            float[] array = (float[]) o;
            out.writeByte(DSCODE.FLOAT_ARRAY.toByte());
            writeFloatArray(array, out);
            return true;
          }
        });
    classesToSerializers.put("[I", // int[]
        new WellKnownPdxDS() {
          @Override
          public boolean toData(Object o, DataOutput out) throws IOException {
            int[] array = (int[]) o;
            out.writeByte(DSCODE.INT_ARRAY.toByte());
            writeIntArray(array, out);
            return true;
          }
        });
    classesToSerializers.put("[J", // long[]
        new WellKnownPdxDS() {
          @Override
          public boolean toData(Object o, DataOutput out) throws IOException {
            long[] array = (long[]) o;
            out.writeByte(DSCODE.LONG_ARRAY.toByte());
            writeLongArray(array, out);
            return true;
          }
        });
    classesToSerializers.put("[S", // short[]
        new WellKnownPdxDS() {
          @Override
          public boolean toData(Object o, DataOutput out) throws IOException {
            short[] array = (short[]) o;
            out.writeByte(DSCODE.SHORT_ARRAY.toByte());
            writeShortArray(array, out);
            return true;
          }
        });
    classesToSerializers.put("[Ljava.lang.String;", // String[]
        new WellKnownPdxDS() {
          @Override
          public boolean toData(Object o, DataOutput out) throws IOException {
            String[] array = (String[]) o;
            out.writeByte(DSCODE.STRING_ARRAY.toByte());
            writeStringArray(array, out);
            return true;
          }
        });

    WellKnownDS TIME_UNIT_SERIALIZER = new WellKnownDS() {
      @Override
      public boolean toData(Object o, DataOutput out) throws IOException {
        TimeUnit timeUnit = (TimeUnit) o;
        switch (timeUnit) {
          case NANOSECONDS: {
            out.writeByte(DSCODE.TIME_UNIT.toByte());
            out.writeByte(StaticSerialization.TIME_UNIT_NANOSECONDS);
            break;
          }
          case MICROSECONDS: {
            out.writeByte(DSCODE.TIME_UNIT.toByte());
            out.writeByte(StaticSerialization.TIME_UNIT_MICROSECONDS);
            break;
          }
          case MILLISECONDS: {
            out.writeByte(DSCODE.TIME_UNIT.toByte());
            out.writeByte(StaticSerialization.TIME_UNIT_MILLISECONDS);
            break;
          }
          case SECONDS: {
            out.writeByte(DSCODE.TIME_UNIT.toByte());
            out.writeByte(StaticSerialization.TIME_UNIT_SECONDS);
            break;
          }
          // handles all other timeunits
          default: {
            writeGemFireEnum(timeUnit, out);
          }
        }
        return true;
      }
    };

    // in java 9 and above, TimeUnit implementation changes. the class name of these units are the
    // same now.
    if (TimeUnit.NANOSECONDS.getClass().getName().equals(TimeUnit.SECONDS.getClass().getName())) {
      classesToSerializers.put(TimeUnit.class.getName(), TIME_UNIT_SERIALIZER);
    }
    // in java 8 and below
    else {
      classesToSerializers.put(TimeUnit.NANOSECONDS.getClass().getName(), TIME_UNIT_SERIALIZER);
      classesToSerializers.put(TimeUnit.MICROSECONDS.getClass().getName(), TIME_UNIT_SERIALIZER);
      classesToSerializers.put(TimeUnit.MILLISECONDS.getClass().getName(), TIME_UNIT_SERIALIZER);
      classesToSerializers.put(TimeUnit.SECONDS.getClass().getName(), TIME_UNIT_SERIALIZER);
    }
    classesToSerializers.put("java.util.Date", new WellKnownPdxDS() {
      @Override
      public boolean toData(Object o, DataOutput out) throws IOException {
        Date date = (Date) o;
        out.writeByte(DSCODE.DATE.toByte());
        writeDate(date, out);
        return true;
      }
    });
    classesToSerializers.put("java.io.File", new WellKnownDS() {
      @Override
      public boolean toData(Object o, DataOutput out) throws IOException {
        File file = (File) o;
        out.writeByte(DSCODE.FILE.toByte());
        writeFile(file, out);
        return true;
      }
    });
    classesToSerializers.put("java.util.ArrayList", new WellKnownPdxDS() {
      @Override
      public boolean toData(Object o, DataOutput out) throws IOException {
        ArrayList list = (ArrayList) o;
        out.writeByte(DSCODE.ARRAY_LIST.toByte());
        writeArrayList(list, out);
        return true;
      }
    });
    classesToSerializers.put("java.util.LinkedList", new WellKnownDS() {
      @Override
      public boolean toData(Object o, DataOutput out) throws IOException {
        LinkedList list = (LinkedList) o;
        out.writeByte(DSCODE.LINKED_LIST.toByte());
        writeLinkedList(list, out);
        return true;
      }
    });
    classesToSerializers.put("java.util.Vector", new WellKnownPdxDS() {
      @Override
      public boolean toData(Object o, DataOutput out) throws IOException {
        out.writeByte(DSCODE.VECTOR.toByte());
        writeVector((Vector) o, out);
        return true;
      }
    });
    classesToSerializers.put("java.util.Stack", new WellKnownDS() {
      @Override
      public boolean toData(Object o, DataOutput out) throws IOException {
        out.writeByte(DSCODE.STACK.toByte());
        writeStack((Stack) o, out);
        return true;
      }
    });
    classesToSerializers.put("java.util.HashSet", new WellKnownPdxDS() {
      @Override
      public boolean toData(Object o, DataOutput out) throws IOException {
        HashSet list = (HashSet) o;
        out.writeByte(DSCODE.HASH_SET.toByte());
        writeHashSet(list, out);
        return true;
      }
    });
    classesToSerializers.put("java.util.LinkedHashSet", new WellKnownPdxDS() {
      @Override
      public boolean toData(Object o, DataOutput out) throws IOException {
        out.writeByte(DSCODE.LINKED_HASH_SET.toByte());
        writeLinkedHashSet((LinkedHashSet) o, out);
        return true;
      }
    });
    classesToSerializers.put("java.util.HashMap", new WellKnownPdxDS() {
      @Override
      public boolean toData(Object o, DataOutput out) throws IOException {
        HashMap list = (HashMap) o;
        out.writeByte(DSCODE.HASH_MAP.toByte());
        writeHashMap(list, out);
        return true;
      }
    });
