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

import java.io.DataInput;
import java.io.DataOutput;
import java.io.File;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.lang.reflect.Array;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Comparator;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.IdentityHashMap;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.LinkedList;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.Stack;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.Vector;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

import org.apache.logging.log4j.Logger;

import org.apache.geode.admin.RegionNotFoundException;
import org.apache.geode.annotations.internal.MakeNotStatic;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.Region;
import org.apache.geode.internal.HeapDataOutputStream;
import org.apache.geode.internal.InternalDataSerializer;
import org.apache.geode.internal.ObjToByteArraySerializer;
import org.apache.geode.internal.cache.CachedDeserializable;
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.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.internal.offheap.StoredObject;
import org.apache.geode.internal.serialization.DSCODE;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.internal.serialization.StaticSerialization;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.pdx.PdxInstance;

/**
 * Provides static helper methods for reading and writing non-primitive data when working with a
 * {@link DataSerializable}. For instance, classes that implement <code>DataSerializable</code> can
 * use the <code>DataSerializer</code> in their <code>toData</code> and <code>fromData</code>
 * methods:
 *
 * <!-- The source code for the Employee class resides in tests/com/examples/ds/Employee.java Please
 * keep the below code snippet in sync with that file. -->
 *
 * <PRE>
 * public class Employee implements DataSerializable {
 *   private int id;
 *   private String name;
 *   private Date birthday;
 *   private Company employer;
 *
 *   public void toData(DataOutput out) throws IOException {
 *     out.writeInt(this.id);
 *     out.writeUTF(this.name);
 *     DataSerializer.writeDate(this.birthday, out);
 *     DataSerializer.writeObject(this.employer, out);
 *   }
 *
 *   public void fromData(DataInput in) throws IOException, ClassNotFoundException {
 *
 *     this.id = in.readInt();
 *     this.name = in.readUTF();
 *     this.birthday = DataSerializer.readDate(in);
 *     this.employer = (Company) DataSerializer.readObject(in);
 *   }
 * }
 *
 * </PRE>
 *
 * <P>
 *
 * Instances of <code>DataSerializer</code> are used to data serialize objects (such as instances of
 * standard Java classes or third-party classes for which the source code is not available) that do
 * not implement the <code>DataSerializable</code> interface.
 *
 * <P>
 *
 * The following <code>DataSerializer</code> data serializes instances of <code>Company</code>. In
 * order for the data serialization framework to consult this custom serializer, it must be
 * {@linkplain #register(Class) registered} with the framework.
 *
 * <!-- The source code for the CompanySerializer class resides in
 * tests/com/examples/ds/CompanySerializer.java Please keep the below code snippet in sync with that
 * file. -->
 *
 * <PRE>
public class CompanySerializer extends DataSerializer {

  static {
    DataSerializer.register(CompanySerializer.class);
  }

  &#47;**
   * May be invoked reflectively if instances of Company are
   * distributed to other VMs.
   *&#47;
  public CompanySerializer() {

  }

  public Class[] getSupportedClasses() {
    return new Class[] { Company.class };
  }
  public int getId() {
    return 42;
  }

  public boolean toData(Object o, DataOutput out)
    throws IOException {
    if (o instanceof Company) {
      Company company = (Company) o;
      out.writeUTF(company.getName());

      // Let's assume that Address is java.io.Serializable
      Address address = company.getAddress();
      writeObject(address, out);
      return true;

    } else {
      return false;
    }
  }

  public Object fromData(DataInput in)
    throws IOException, ClassNotFoundException {

    String name = in.readUTF();
    Address address = (Address) readObject(in);
    return new Company(name, address);
  }
}
 * </PRE>
 *
 * Just like {@link Instantiator}s, a <code>DataSerializer</code> may be sent to other members of
 * the distributed system when it is {@linkplain #register(Class) registered}. The data
 * serialization framework does not require that a <code>DataSerializer</code> be
 * {@link Serializable}, but it does require that it provide a {@linkplain #DataSerializer()
 * zero-argument constructor}.
 *
 * @see #writeObject(Object, DataOutput)
 * @see #readObject
 *
 * @since GemFire 3.5
 */
public abstract class DataSerializer {

  private static final Logger logger = LogService.getLogger();

  /** The eventId of this <code>DataSerializer</code> */
  private EventID eventId;

  /** The originator of this <code>DataSerializer</code> */
  private ClientProxyMembershipID context;

  /**
   * @deprecated Use Boolean.getBoolean("DataSerializer.TRACE_SERIALIZABLE") instead.
   */
  @Deprecated
  protected static final boolean TRACE_SERIALIZABLE =
      Boolean.getBoolean("DataSerializer.TRACE_SERIALIZABLE");

  /* Used to prevent standard Java serialization when sending data to a non-Java client */
  protected static final ThreadLocal<Boolean> DISALLOW_JAVA_SERIALIZATION = new ThreadLocal<>();

  /**
   * Writes an instance of <code>Class</code> to a <code>DataOutput</code>. This method will handle
   * a <code>null</code> value and not throw a <code>NullPointerException</code>.
   *
   * @param c the <code>Class</code> to write
   * @param out the {@link DataInput} to write to
   *
   * @throws IOException A problem occurs while writing to <code>out</code>
   *
   * @see #readClass
   */
  public static void writeClass(Class<?> c, DataOutput out) throws IOException {

    InternalDataSerializer.checkOut(out);

    if (logger.isTraceEnabled(LogMarker.SERIALIZER_VERBOSE)) {
      logger.trace(LogMarker.SERIALIZER_VERBOSE, "Writing Class {}", c);
    }

    if (c == null || c.isPrimitive()) {
      StaticSerialization.writePrimitiveClass(c, out);
    } else {
      // non-primitive classes have a second CLASS byte
      // if readObject/writeObject is called:
      // the first CLASS byte indicates it's a Class, the second
      // one indicates it's a non-primitive Class
      out.writeByte(DSCODE.CLASS.toByte());
      String cname = c.getName();
      cname = InternalDataSerializer.processOutgoingClassName(cname, out);
      writeString(cname, out);
    }
  }

  /**
   * Writes class name to a <code>DataOutput</code>. This method will handle a <code>null</code>
   * value and not throw a <code>NullPointerException</code>.
   *
   * @param className the class name to write
   * @param out the {@link DataInput} to write to
   *
   * @throws IOException A problem occurs while writing to <code>out</code>
   *
   * @see #readNonPrimitiveClassName(DataInput)
   */
  public static void writeNonPrimitiveClassName(String className, DataOutput out)
      throws IOException {

    InternalDataSerializer.checkOut(out);

    if (logger.isTraceEnabled(LogMarker.SERIALIZER_VERBOSE)) {
      logger.trace(LogMarker.SERIALIZER_VERBOSE, "Writing Class name {}", className);
    }

    writeString(InternalDataSerializer.processOutgoingClassName(className, out), out);
  }

  /**
   * Reads an instance of <code>Class</code> from a <code>DataInput</code>. The class will be loaded
   * using the {@linkplain Thread#getContextClassLoader current content class loader}. The return
   * value may be <code>null</code>.
   *
   * @param in the {@link DataInput} to read from
   * @return the deserialized <code>Class</code>
   *
   * @throws IOException A problem occurs while reading from <code>in</code>
   * @throws ClassNotFoundException The class cannot be loaded
   */
  public static Class<?> readClass(DataInput in) throws IOException, ClassNotFoundException {

    InternalDataSerializer.checkIn(in);

    byte typeCode = in.readByte();
    if (typeCode == DSCODE.CLASS.toByte()) {
      String className = readString(in);
      return InternalDataSerializer.getCachedClass(className);
    } else {
      return StaticSerialization.decodePrimitiveClass(typeCode);
    }
  }

  /**
   * Reads name of an instance of <code>Class</code> from a <code>DataInput</code>.
   *
   * The return value may be <code>null</code>.
   *
   * @param in the {@link DataInput} to read from
   * @return the deserialized Class name
   *
   * @throws IOException A problem occurs while reading from <code>in</code>
   * @see #writeNonPrimitiveClassName(String, DataOutput)
   */
  public static String readNonPrimitiveClassName(DataInput in) throws IOException {

    InternalDataSerializer.checkIn(in);

    return InternalDataSerializer.processIncomingClassName(readString(in));
  }

  /**
   * Writes an instance of Region. A Region is serialized as just a reference to a full path only.
   * It will be recreated on the other end by calling {@link CacheFactory#getAnyInstance} and then
   * calling <code>getRegion</code> on it. This method will handle a <code>null</code> value and not
   * throw a <code>NullPointerException</code>.
   *
   * @param rgn the Region to write
   * @param out the {@link DataInput} to write to
   *
   * @throws IOException if a problem occurs while reading from <code>in</code>
   */
  public static void writeRegion(Region<?, ?> rgn, DataOutput out) throws IOException {
    writeString((rgn != null) ? rgn.getFullPath() : null, out);
  }

  /**
   * Reads an instance of Region. A Region is serialized as a reference to a full path only. It is
   * recreated on the other end by calling {@link CacheFactory#getAnyInstance} and then calling
   * <code>getRegion</code> on it. The return value may be <code>null</code>.
   *
   * @param <K> the type of keys in the region
   * @param <V> the type of values in the region
   * @param in the input stream
   * @return the Region instance
   * @throws org.apache.geode.cache.CacheClosedException if a cache has not been created or the only
   *         created one is closed.
   * @throws RegionNotFoundException if there is no region by this name in the Cache
   * @throws IOException if a problem occurs while reading from <code>in</code>
   * @throws ClassNotFoundException if the class of one of the Region's elements cannot be found.
   */
  public static <K, V> Region<K, V> readRegion(DataInput in)
      throws IOException, ClassNotFoundException {
    String fullPath = readString(in);
    Region<K, V> rgn = null;
    if (fullPath != null) {
      // use getExisting to fix bug 43151
      rgn = ((Cache) GemFireCacheImpl.getExisting("Needed cache to find region."))
          .getRegion(fullPath);
      if (rgn == null) {
        throw new RegionNotFoundException(
            String.format(
                "Region ' %s ' could not be found while reading a DataSerializer stream",
                fullPath));
      }
    }
    return rgn;
  }

  /**
   * Writes an instance of <code>Date</code> to a <code>DataOutput</code>. Note that even though
   * <code>date</code> may be an instance of a subclass of <code>Date</code>, <code>readDate</code>
   * will always return an instance of <code>Date</code>, <B>not</B> an instance of the subclass. To
   * preserve the class type of <code>date</code>,\ {@link #writeObject(Object, DataOutput)} should
   * be used for data serialization. This method will handle a <code>null</code> value and not throw
   * a <code>NullPointerException</code>.
   *
   * @param date the <code>Date</code> to write
   * @param out the {@link DataInput} to write to
   *
   * @throws IOException A problem occurs while writing to <code>out</code>
   *
   * @see #readDate
   */
  public static void writeDate(Date date, DataOutput out) throws IOException {

    InternalDataSerializer.checkOut(out);

    if (logger.isTraceEnabled(LogMarker.SERIALIZER_VERBOSE)) {
      logger.trace(LogMarker.SERIALIZER_VERBOSE, "Writing Date {}", date);
    }

    long v;
    if (date == null) {
      v = -1L;
    } else {
      v = date.getTime();
      if (v == -1L) {
        throw new IllegalArgumentException(
            "Dates whose getTime returns -1 can not be DataSerialized.");
      }
    }
    out.writeLong(v);
  }

  /**
   * Reads an instance of <code>Date</code> from a <code>DataInput</code>. The return value may be
   * <code>null</code>.
   *
   * @param in the {@link DataInput} to read from
   * @return the deserialized <code>Date</code>
   *
   * @throws IOException A problem occurs while reading from <code>in</code>
   */
  public static Date readDate(DataInput in) throws IOException {
    InternalDataSerializer.checkIn(in);

    long time = in.readLong();
    Date date = null;
    if (time != -1L) {
      date = new Date(time);
    }

    if (logger.isTraceEnabled(LogMarker.SERIALIZER_VERBOSE)) {
      logger.trace(LogMarker.SERIALIZER_VERBOSE, "Read Date {}", date);
    }

    return date;
  }

  /**
   * Writes an instance of <code>File</code> to a <code>DataOutput</code>. Note that even though
   * <code>file</code> may be an instance of a subclass of <code>File</code>, <code>readFile</code>
   * will always return an instance of <code>File</code>, <B>not</B> an instance of the subclass. To
   * preserve the class type of <code>file</code>, {@link #writeObject(Object, DataOutput)} should
   * be used for data serialization. This method will handle a <code>null</code> value and not throw
   * a <code>NullPointerException</code>.
   *
   * @param file the <code>File</code> to write
   * @param out the {@link DataInput} to write to
   *
   * @throws IOException A problem occurs while writing to <code>out</code>
   *
   * @see #readFile
   * @see File#getCanonicalPath
   */
  public static void writeFile(File file, DataOutput out) throws IOException {

    InternalDataSerializer.checkOut(out);

    if (logger.isTraceEnabled(LogMarker.SERIALIZER_VERBOSE)) {
      logger.trace(LogMarker.SERIALIZER_VERBOSE, "Writing File {}", file);
    }

    writeString((file != null) ? file.getCanonicalPath() : null, out);
  }

  /**
   * Reads an instance of <code>File</code> from a <code>DataInput</code>. The return value may be
   * <code>null</code>.
   *
   * @param in the {@link DataInput} to read from
   * @return the deserialized <code>File</code>
   *
   * @throws IOException A problem occurs while reading from <code>in</code>
   */
  public static File readFile(DataInput in) throws IOException {
    InternalDataSerializer.checkIn(in);

    String s = readString(in);
    File file = null;
    if (s != null) {
      file = new File(s);
    }
    if (logger.isTraceEnabled(LogMarker.SERIALIZER_VERBOSE)) {
      logger.trace(LogMarker.SERIALIZER_VERBOSE, "Read File {}", file);
    }

    return file;
  }

  /**
   * Writes an instance of <code>InetAddress</code> to a <code>DataOutput</code>. The
   * <code>InetAddress</code> is data serialized by writing its {@link InetAddress#getAddress byte}
   * representation to the <code>DataOutput</code>. {@link #readInetAddress} converts the
   * <code>byte</code> representation to an instance of <code>InetAddress</code> using
   * {@link InetAddress#getAddress}. As a result, if <code>address</code> is an instance of a
   * user-defined subclass of <code>InetAddress</code> (that is, not an instance of one of the
   * subclasses from the <code>java.net</code> package), its class will not be preserved. In order
   * to be able to read an instance of the user-defined class,
   * {@link #writeObject(Object, DataOutput)} should be used. This method will handle a
   * <code>null</code> value and not throw a <code>NullPointerException</code>.
   *
   * @param address the <code>InetAddress</code> to write
   * @param out the {@link DataInput} to write to
   *
   * @throws IOException A problem occurs while writing to <code>out</code>
   *
   * @see #readInetAddress
   */
  public static void writeInetAddress(InetAddress address, DataOutput out) throws IOException {

    InternalDataSerializer.checkOut(out);

    if (logger.isTraceEnabled(LogMarker.SERIALIZER_VERBOSE)) {
      logger.trace(LogMarker.SERIALIZER_VERBOSE, "Writing InetAddress {}", address);
    }

    StaticSerialization.writeInetAddress(address, out);
  }

  /**
   * Reads an instance of <code>InetAddress</code> from a <code>DataInput</code>. The return value
   * may be <code>null</code>.
   *
   * @param in the {@link DataInput} to read from
   * @return the deserialized <code>InetAddress</code>
   *
   * @throws IOException A problem occurs while reading from <code>in</code> or the address read
   *         from <code>in</code> is unknown
   *
   * @see InetAddress#getAddress
   */
  public static InetAddress readInetAddress(DataInput in) throws IOException {

    InternalDataSerializer.checkIn(in);
    InetAddress address = StaticSerialization.readInetAddress(in);

    if (logger.isTraceEnabled(LogMarker.SERIALIZER_VERBOSE)) {
      logger.trace(LogMarker.SERIALIZER_VERBOSE, "Read InetAddress {}", address);
    }

    return address;
  }

  /**
   * Writes an instance of <code>String</code> to a <code>DataOutput</code>. This method will handle
   * a <code>null</code> value and not throw a <code>NullPointerException</code>.
   * <p>
   * As of 5.7 strings longer than 0xFFFF can be serialized.
   *
   * @param value the <code>String</code> to write
   * @param out the {@link DataInput} to write to
   *
   * @throws IOException A problem occurs while writing to <code>out</code>
   *
   * @see #readString
   */
  public static void writeString(String value, DataOutput out) throws IOException {

    InternalDataSerializer.checkOut(out);

    final boolean isTraceSerialzerVerbose = logger.isTraceEnabled(LogMarker.SERIALIZER_VERBOSE);
    if (isTraceSerialzerVerbose) {
      logger.trace(LogMarker.SERIALIZER_VERBOSE, "Writing String \"{}\"", value);
    }

    StaticSerialization.writeString(value, out);
  }

  /**
   * Reads an instance of <code>String</code> from a <code>DataInput</code>. The return value may be
   * <code>null</code>.
   *
   * @param in the {@link DataInput} to read from
   * @return the deserialized <code>String</code>
   *
   * @throws IOException A problem occurs while reading from <code>in</code>
   *
   * @see #writeString
   */
  public static String readString(DataInput in) throws IOException {
    return StaticSerialization.readString(in);
  }

  /**
   * Writes an instance of <code>Boolean</code> to a <code>DataOutput</code>.
   *
   * @param value the <code>Boolean</code> to write
   * @param out the {@link DataInput} to write to
   *
   * @throws IOException A problem occurs while writing to <code>out</code>
   * @throws NullPointerException if value is null.
   *
   * @see #readBoolean
   */
  public static void writeBoolean(Boolean value, DataOutput out) throws IOException {

    InternalDataSerializer.checkOut(out);

    if (logger.isTraceEnabled(LogMarker.SERIALIZER_VERBOSE)) {
      logger.trace(LogMarker.SERIALIZER_VERBOSE, "Writing Boolean {}", value);
    }

    out.writeBoolean(value);
  }

  /**
   * Reads an instance of <code>Boolean</code> from a <code>DataInput</code>.
   *
   * @param in the {@link DataInput} to read from
   * @return the deserialized <code>Boolean</code>
   *
   * @throws IOException A problem occurs while reading from <code>in</code>
   */
  public static Boolean readBoolean(DataInput in) throws IOException {
    InternalDataSerializer.checkIn(in);

    Boolean value = in.readBoolean();
    if (logger.isTraceEnabled(LogMarker.SERIALIZER_VERBOSE)) {
      logger.trace(LogMarker.SERIALIZER_VERBOSE, "Read Boolean {}", value);
    }
    return value;
  }

  /**
   * Writes an instance of <code>Character</code> to a <code>DataOutput</code>.
   *
   * @param value the <code>Character</code> to write
   * @param out the {@link DataInput} to write to
   *
   * @throws IOException A problem occurs while writing to <code>out</code>
   * @throws NullPointerException if value is null.
   *
   * @see #readCharacter
   */
  public static void writeCharacter(Character value, DataOutput out) throws IOException {

    InternalDataSerializer.checkOut(out);

    if (logger.isTraceEnabled(LogMarker.SERIALIZER_VERBOSE)) {
      logger.trace(LogMarker.SERIALIZER_VERBOSE, "Writing Character {}", value);
    }

    out.writeChar(value);
  }

  /**
   * Reads an instance of <code>Character</code> from a <code>DataInput</code>.
   *
   * @param in the {@link DataInput} to read from
   * @return the deserialized <code>Character</code>
   *
   * @throws IOException A problem occurs while reading from <code>in</code>
   */
  public static Character readCharacter(DataInput in) throws IOException {

    InternalDataSerializer.checkIn(in);

    Character value = in.readChar();
    if (logger.isTraceEnabled(LogMarker.SERIALIZER_VERBOSE)) {
      logger.trace(LogMarker.SERIALIZER_VERBOSE, "Read Character {}", value);
    }
    return value;
  }

  /**
   * Writes an instance of <code>Byte</code> to a <code>DataOutput</code>.
   *
   * @param value the <code>Byte</code> to write
   * @param out the {@link DataInput} to write to
   *
   * @throws IOException A problem occurs while writing to <code>out</code>
   * @throws NullPointerException if value is null.
   *
   * @see #readByte
   */
  public static void writeByte(Byte value, DataOutput out) throws IOException {

    InternalDataSerializer.checkOut(out);

    if (logger.isTraceEnabled(LogMarker.SERIALIZER_VERBOSE)) {
      logger.trace(LogMarker.SERIALIZER_VERBOSE, "Writing Byte {}", value);
    }

    out.writeByte(value);
  }

  /**
   * Reads an instance of <code>Byte</code> from a <code>DataInput</code>.
   *
   * @param in the {@link DataInput} to read from
   * @return the deserialized <code>Byte</code>
   *
   * @throws IOException A problem occurs while reading from <code>in</code>
   */
  public static Byte readByte(DataInput in) throws IOException {
    InternalDataSerializer.checkIn(in);

    Byte value = in.readByte();
    if (logger.isTraceEnabled(LogMarker.SERIALIZER_VERBOSE)) {
      logger.trace(LogMarker.SERIALIZER_VERBOSE, "Read Byte {}", value);
    }
    return value;
  }

  /**
   * Writes an instance of <code>Short</code> to a <code>DataOutput</code>.
   *
   * @param value the <code>Short</code> to write
   * @param out the {@link DataInput} to write to
   *
   * @throws IOException A problem occurs while writing to <code>out</code>
   * @throws NullPointerException if value is null.
   *
   * @see #readShort
   */
  public static void writeShort(Short value, DataOutput out) throws IOException {

    InternalDataSerializer.checkOut(out);

    if (logger.isTraceEnabled(LogMarker.SERIALIZER_VERBOSE)) {
      logger.trace(LogMarker.SERIALIZER_VERBOSE, "Writing Short {}", value);
    }

    out.writeShort(value);
  }

  /**
   * Reads an instance of <code>Short</code> from a <code>DataInput</code>.
   *
   * @param in the {@link DataInput} to read from
   * @return the deserialized <code>Short</code>
   *
   * @throws IOException A problem occurs while reading from <code>in</code>
   */
  public static Short readShort(DataInput in) throws IOException {
    InternalDataSerializer.checkIn(in);

    Short value = in.readShort();
    if (logger.isTraceEnabled(LogMarker.SERIALIZER_VERBOSE)) {
      logger.trace(LogMarker.SERIALIZER_VERBOSE, "Read Short {}", value);
    }
    return value;
  }

  /**
   * Writes an instance of <code>Integer</code> to a <code>DataOutput</code>.
   *
   * @param value the <code>Integer</code> to write
   * @param out the {@link DataInput} to write to
   *
   * @throws IOException A problem occurs while writing to <code>out</code>
   * @throws NullPointerException if value is null.
   *
   * @see #readInteger
   */
  public static void writeInteger(Integer value, DataOutput out) throws IOException {

    InternalDataSerializer.checkOut(out);

    if (logger.isTraceEnabled(LogMarker.SERIALIZER_VERBOSE)) {
      logger.trace(LogMarker.SERIALIZER_VERBOSE, "Writing Integer {}", value);
    }

    out.writeInt(value);
  }

  /**
   * Reads an instance of <code>Integer</code> from a <code>DataInput</code>.
   *
   * @param in the {@link DataInput} to read from
   * @return the deserialized <code>Integer</code>
   *
   * @throws IOException A problem occurs while reading from <code>in</code>
   */
  public static Integer readInteger(DataInput in) throws IOException {
    InternalDataSerializer.checkIn(in);

    Integer value = in.readInt();
    if (logger.isTraceEnabled(LogMarker.SERIALIZER_VERBOSE)) {
      logger.trace(LogMarker.SERIALIZER_VERBOSE, "Read Integer {}", value);
    }
    return value;
  }

  /**
   * Writes an instance of <code>Long</code> to a <code>DataOutput</code>.
   *
   * @param value the <code>Long</code> to write
   * @param out the {@link DataInput} to write to
   *
   * @throws IOException A problem occurs while writing to <code>out</code>
   * @throws NullPointerException if value is null.
   *
   * @see #readLong
   */
  public static void writeLong(Long value, DataOutput out) throws IOException {

    InternalDataSerializer.checkOut(out);

    if (logger.isTraceEnabled(LogMarker.SERIALIZER_VERBOSE)) {
      logger.trace(LogMarker.SERIALIZER_VERBOSE, "Writing Long {}", value);
    }

    out.writeLong(value);
  }

  /**
   * Reads an instance of <code>Long</code> from a <code>DataInput</code>.
   *
   * @param in the {@link DataInput} to read from
   * @return the deserialized <code>Long</code>
   *
   * @throws IOException A problem occurs while reading from <code>in</code>
   */
  public static Long readLong(DataInput in) throws IOException {
    InternalDataSerializer.checkIn(in);

    Long value = in.readLong();
    if (logger.isTraceEnabled(LogMarker.SERIALIZER_VERBOSE)) {
      logger.trace(LogMarker.SERIALIZER_VERBOSE, "Read Long {}", value);
    }
    return value;
  }

  /**
   * Writes an instance of <code>Float</code> to a <code>DataOutput</code>.
   *
   * @param value the <code>Float</code> to write
   * @param out the {@link DataInput} to write to
   *
   * @throws IOException A problem occurs while writing to <code>out</code>
   * @throws NullPointerException if value is null.
   *
   * @see #readFloat
   */
  public static void writeFloat(Float value, DataOutput out) throws IOException {

    InternalDataSerializer.checkOut(out);

    if (logger.isTraceEnabled(LogMarker.SERIALIZER_VERBOSE)) {
      logger.trace(LogMarker.SERIALIZER_VERBOSE, "Writing Float {}", value);
    }

    out.writeFloat(value);
  }

  /**
   * Reads an instance of <code>Float</code> from a <code>DataInput</code>.
   *
   * @param in the {@link DataInput} to read from
   * @return the deserialized <code>Float</code>
   *
   * @throws IOException A problem occurs while reading from <code>in</code>
   */
  public static Float readFloat(DataInput in) throws IOException {
    InternalDataSerializer.checkIn(in);

    Float value = in.readFloat();
    if (logger.isTraceEnabled(LogMarker.SERIALIZER_VERBOSE)) {
      logger.trace(LogMarker.SERIALIZER_VERBOSE, "Read Float {}", value);
    }
    return value;
  }

  /**
   * Writes an instance of <code>Double</code> to a <code>DataOutput</code>.
   *
   * @param value the <code>Double</code> to write
   * @param out the {@link DataInput} to write to
   *
   * @throws IOException A problem occurs while writing to <code>out</code>
   * @throws NullPointerException if value is null.
   *
   * @see #readDouble
   */
  public static void writeDouble(Double value, DataOutput out) throws IOException {

    InternalDataSerializer.checkOut(out);

    if (logger.isTraceEnabled(LogMarker.SERIALIZER_VERBOSE)) {
      logger.trace(LogMarker.SERIALIZER_VERBOSE, "Writing Double {}", value);
    }

    out.writeDouble(value);
  }

  /**
   * Reads an instance of <code>Double</code> from a <code>DataInput</code>.
   *
   * @param in the {@link DataInput} to read from
   * @return the deserialized <code>Double</code>
   *
   * @throws IOException A problem occurs while reading from <code>in</code>
   */
  public static Double readDouble(DataInput in) throws IOException {
    InternalDataSerializer.checkIn(in);

    Double value = in.readDouble();
    if (logger.isTraceEnabled(LogMarker.SERIALIZER_VERBOSE)) {
      logger.trace(LogMarker.SERIALIZER_VERBOSE, "Read Double {}", value);
    }
    return value;
  }

  /**
   * Writes a primitive <code>boolean</code> to a <code>DataOutput</code>.
   *
   * @param value the primitive <code>boolean</code> to write
   * @param out the {@link DataInput} to write to
   *
   * @throws IOException A problem occurs while writing to <code>out</code>
   *
   * @see DataOutput#writeBoolean
   * @since GemFire 5.1
   */
  public static void writePrimitiveBoolean(boolean value, DataOutput out) throws IOException {

    InternalDataSerializer.checkOut(out);

    if (logger.isTraceEnabled(LogMarker.SERIALIZER_VERBOSE)) {
      logger.trace(LogMarker.SERIALIZER_VERBOSE, "Writing Boolean {}", value);
    }

    out.writeBoolean(value);
  }

  /**
   * Reads a primitive <code>boolean</code> from a <code>DataInput</code>.
   *
   * @param in the {@link DataInput} to read from
   * @return the deserialized primitive <code>boolean</code>
   *
   * @throws IOException A problem occurs while reading from <code>in</code>
   * @see DataInput#readBoolean
   * @since GemFire 5.1
   */
  public static boolean readPrimitiveBoolean(DataInput in) throws IOException {
    InternalDataSerializer.checkIn(in);

    boolean value = in.readBoolean();
    if (logger.isTraceEnabled(LogMarker.SERIALIZER_VERBOSE)) {
      logger.trace(LogMarker.SERIALIZER_VERBOSE, "Read Boolean {}", value);
    }
    return value;
  }

  /**
   * Writes a primitive <code>byte</code> to a <code>DataOutput</code>.
   *
   * @param value the primitive <code>byte</code> to write
   * @param out the {@link DataInput} to write to
   *
   * @throws IOException A problem occurs while writing to <code>out</code>
   *
   * @see DataOutput#writeByte
   * @since GemFire 5.1
   */
  public static void writePrimitiveByte(byte value, DataOutput out) throws IOException {

    InternalDataSerializer.checkOut(out);

    if (logger.isTraceEnabled(LogMarker.SERIALIZER_VERBOSE)) {
      logger.trace(LogMarker.SERIALIZER_VERBOSE, "Writing Byte {}", value);
    }

    out.writeByte(value);
  }

  /**
   * Reads a primitive <code>byte</code> from a <code>DataInput</code>.
   *
   * @param in the {@link DataInput} to read from
   * @return the deserialized primitive <code>byte</code>
   *
   * @throws IOException A problem occurs while reading from <code>in</code>
   * @see DataInput#readByte
   * @since GemFire 5.1
   */
  public static byte readPrimitiveByte(DataInput in) throws IOException {
    InternalDataSerializer.checkIn(in);

    byte value = in.readByte();
    if (logger.isTraceEnabled(LogMarker.SERIALIZER_VERBOSE)) {
      logger.trace(LogMarker.SERIALIZER_VERBOSE, "Read Byte {}", value);
    }
    return value;
  }

  /**
   * Writes a primitive <code>char</code> to a <code>DataOutput</code>.
   *
