/*
 * 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.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInput;
import java.io.DataInputStream;
import java.io.DataOutput;
import java.io.DataOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.sql.Timestamp;
import java.util.ArrayList;
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.Properties;
import java.util.Random;
import java.util.Stack;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.UUID;
import java.util.Vector;
import java.util.concurrent.TimeUnit;

import org.junit.Before;
import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.rules.TestName;

import org.apache.geode.CanonicalInstantiator;
import org.apache.geode.DataSerializable;
import org.apache.geode.DataSerializer;
import org.apache.geode.Instantiator;
import org.apache.geode.SystemFailure;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.internal.serialization.VersionedDataInputStream;
import org.apache.geode.internal.serialization.VersionedDataOutputStream;
import org.apache.geode.internal.statistics.StatArchiveWriter;
import org.apache.geode.internal.tcp.ByteBufferInputStream;
import org.apache.geode.test.junit.categories.SerializationTest;

/**
 * Tests the functionality of the {@link DataSerializable} class.
 *
 * @since GemFire 3.0
 */
@Category({SerializationTest.class})
public class DataSerializableJUnitTest implements Serializable {

  /** A <code>ByteArrayOutputStream</code> that data is serialized to */
  private transient ByteArrayOutputStream baos;

  @Rule
  public transient TestName testName = new TestName();

  /**
   * Creates a new <code>ByteArrayOutputStream</code> for this test to work with.
   */
  @Before
  public void setUp() {
    baos = new ByteArrayOutputStream();
  }

  public void tearDown() {
    baos = null;
  }

  /**
   * Returns a <code>DataOutput</code> to write to
   */
  private DataOutputStream getDataOutput() {
    return new DataOutputStream(baos);
  }

  /**
   * Returns a <code>DataInput</code> to read from
   */
  private DataInput getDataInput() {
    // changed this to use ByteBufferInputStream to give us better
    // test coverage of this class.
    ByteBuffer bb = ByteBuffer.wrap(baos.toByteArray());
    return new ByteBufferInputStream(bb);
  }

  private DataInputStream getDataInputStream() {
    ByteBuffer bb = ByteBuffer.wrap(baos.toByteArray());
    ByteBufferInputStream bbis = new ByteBufferInputStream(bb);
    return new DataInputStream(bbis);
  }

  /**
   * Returns a random number generator
   */
  private Random getRandom() {
    long seed = Long.getLong("SEED", System.currentTimeMillis());
    System.out.println("SEED for " + testName.getMethodName() + ": " + seed);
    return new Random(seed);
  }

  /**
   * Tests data serializing a {@link Class}
   */
  @Test
  public void testClass() throws Exception {
    Class<? extends DataSerializableJUnitTest> c = getClass();

    DataOutputStream out = getDataOutput();
    DataSerializer.writeClass(c, out);
    out.flush();

    DataInput in = getDataInput();
    Class<?> c2 = DataSerializer.readClass(in);
    assertEquals(c, c2);
  }

  /**
   * Tests data serializing a {@link Class} using {@link DataSerializer#writeObject}.
   */
  @Test
  public void testClassObject() throws Exception {
    Class<? extends DataSerializableJUnitTest> c = getClass();

    DataOutputStream out = getDataOutput();
    DataSerializer.writeObject(c, out);
    out.flush();

    DataInput in = getDataInput();
    Class<?> c2 = DataSerializer.readObject(in);
    assertEquals(c, c2);
  }

  @Test
  public void testBigInteger() throws Exception {
    BigInteger o = new BigInteger("12345678901234567890");

    DataOutputStream out = getDataOutput();
    DataSerializer.writeObject(o, out, false);
    out.flush();

    DataInput in = getDataInput();
    BigInteger o2 = DataSerializer.readObject(in);
    assertEquals(o, o2);
  }

  @Test
  public void testBigDecimal() throws Exception {
    BigDecimal o = new BigDecimal("1234567890.1234567890");

    DataOutputStream out = getDataOutput();
    DataSerializer.writeObject(o, out, false);
    out.flush();

    DataInput in = getDataInput();
    BigDecimal o2 = DataSerializer.readObject(in);
    assertEquals(o, o2);
  }

  @Test
  public void testUUID() throws Exception {
    UUID o = UUID.randomUUID();

    DataOutputStream out = getDataOutput();
    DataSerializer.writeObject(o, out, false);
    out.flush();

    DataInput in = getDataInput();
    UUID o2 = DataSerializer.readObject(in);
    assertEquals(o, o2);
  }

  @Test
  public void testTimestamp() throws Exception {
    Timestamp o = new Timestamp(new Date().getTime() + 79);

    DataOutputStream out = getDataOutput();
    DataSerializer.writeObject(o, out, false);
    out.flush();

    DataInput in = getDataInput();
    Timestamp o2 = DataSerializer.readObject(in);
    assertEquals(o, o2);
  }

  /**
   * Tests data serializing a {@link Date}
   */
  @Test
  public void testDate() throws Exception {
    Date date = new Date();

    DataOutputStream out = getDataOutput();
    DataSerializer.writeDate(date, out);
    out.flush();

    DataInput in = getDataInput();
    Date date2 = DataSerializer.readDate(in);
    assertEquals(date, date2);
  }

  /**
   * Tests data serializing a {@link Date} using {@link DataSerializer#writeObject}.
   */
  @Test
  public void testDateObject() throws Exception {
    Date date = new Date();

    DataOutputStream out = getDataOutput();
    DataSerializer.writeObject(date, out);
    out.flush();

    DataInput in = getDataInput();
    Date date2 = DataSerializer.readObject(in);
    assertEquals(date, date2);
  }

  /**
   * Tests data serializing a {@link File}
   */
  @Test
  public void testFile() throws Exception {
    File file = new File(System.getProperty("user.dir"));

    DataOutputStream out = getDataOutput();
    DataSerializer.writeFile(file, out);
    out.flush();

    DataInput in = getDataInput();
    File file2 = DataSerializer.readFile(in);
    assertEquals(file, file2);
  }

  /**
   * Tests data serializing a {@link File} using {@link DataSerializer#writeObject}.
   */
  @Test
  public void testFileObject() throws Exception {
    File file = new File(System.getProperty("user.dir"));

    DataOutputStream out = getDataOutput();
    DataSerializer.writeObject(file, out);
    out.flush();

    DataInput in = getDataInput();
    File file2 = DataSerializer.readObject(in);
    assertEquals(file, file2);
  }

  /**
   * Tests data serializing a {@link InetAddress}
   */
  @Test
  public void testInetAddress() throws Exception {
    InetAddress address = InetAddress.getLocalHost();

    DataOutputStream out = getDataOutput();
    DataSerializer.writeInetAddress(address, out);
    out.flush();

    DataInput in = getDataInput();
    InetAddress address2 = DataSerializer.readInetAddress(in);
    assertEquals(address, address2);
  }

  /**
   * Tests data serializing a {@link InetAddress} using {@link DataSerializer#writeObject}.
   */
  @Test
  public void testInetAddressObject() throws Exception {
    InetAddress address = InetAddress.getLocalHost();

    DataOutputStream out = getDataOutput();
    DataSerializer.writeObject(address, out);
    out.flush();

    DataInput in = getDataInput();
    InetAddress address2 = DataSerializer.readObject(in);
    assertEquals(address, address2);
  }

  /**
   * Tests data serializing <code>null</code> using {@link DataSerializer#writeObject}.
   */
  @Test
  public void testNullObject() throws Exception {
    Object value = null;

    DataOutputStream out = getDataOutput();
    DataSerializer.writeObject(value, out);
    out.flush();

    DataInput in = getDataInput();
    Object value2 = DataSerializer.readObject(in);
    assertEquals(value, value2);
  }

  /**
   * Tests data serializing a non-<code>null</code> {@link String}
   */
  @Test
  public void testString() throws Exception {
    String value = "Hello";

    DataOutputStream out = getDataOutput();
    DataSerializer.writeString(value, out);
    DataSerializer.writeObject(value, out);
    out.flush();

    DataInput in = getDataInput();
    String value2 = DataSerializer.readString(in);
    assertEquals(value, value2);
    value2 = DataSerializer.readObject(in);
    assertEquals(value, value2);
  }

  @Test
  public void testUtfString() throws Exception {
    String value = "Hello" + Character.MIN_VALUE + Character.MAX_VALUE;

    DataOutputStream out = getDataOutput();
    DataSerializer.writeString(value, out);
    DataSerializer.writeObject(value, out);
    out.flush();

    DataInput in = getDataInput();
    String value2 = DataSerializer.readString(in);
    assertEquals(value, value2);
    value2 = DataSerializer.readObject(in);
    assertEquals(value, value2);
  }

  @Test
  public void testBigString() throws Exception {
    StringBuilder sb = new StringBuilder(100000);
    for (int i = 0; i < 100000; i++) {
      sb.append("a");
    }
    String value = sb.toString();

    DataOutputStream out = getDataOutput();
    DataSerializer.writeString(value, out);
    DataSerializer.writeObject(value, out);
    out.flush();

    DataInput in = getDataInput();
    String value2 = DataSerializer.readString(in);
    assertEquals(value, value2);
    value2 = DataSerializer.readObject(in);
    assertEquals(value, value2);
  }

  /**
   * Tests data serializing a non-<code>null</code> {@link String} longer than 64k.
   */
  @Test
  public void testBigUtfString() throws Exception {
    StringBuilder sb = new StringBuilder(100000);
    for (int i = 0; i < 100000; i++) {
      if ((i % 2) == 0) {
        sb.append(Character.MAX_VALUE);
      } else {
        sb.append(Character.MIN_VALUE);
      }
    }
    String value = sb.toString();

    DataOutputStream out = getDataOutput();
    DataSerializer.writeString(value, out);
    DataSerializer.writeObject(value, out);
    out.flush();

    DataInput in = getDataInput();
    String value2 = DataSerializer.readString(in);
    assertEquals(value, value2);
    value2 = DataSerializer.readObject(in);
    assertEquals(value, value2);
  }

  /**
   * Tests data serializing a non-ascii {@link String}
   */
  @Test
  public void testNonAsciiString() throws Exception {
    basicTestString("Hello1" + '\u0000');
    setUp();
    basicTestString("Hello2" + '\u0080');
    setUp();
    basicTestString("Hello3" + '\uFFFF');
  }

  private void basicTestString(String value) throws IOException {
    DataOutputStream out = getDataOutput();
    DataSerializer.writeString(value, out);
    out.flush();

    DataInput in = getDataInput();
    String value2 = DataSerializer.readString(in);
    assertEquals(value, value2);

  }

  /**
   * Tests data serializing a <code>null</code> {@link String}
   */
  @Ignore("for unknown reason")
  @Test
  public void testNullString() throws Exception {
    basicTestString(null);
  }

  /**
   * Tests data serializing a {@link Boolean}
   */
  @Test
  public void testBoolean() throws Exception {
    Boolean value = getRandom().nextInt() % 2 == 0;

    DataOutputStream out = getDataOutput();
    DataSerializer.writeBoolean(value, out);
    out.flush();

    DataInput in = getDataInput();
    Boolean value2 = DataSerializer.readBoolean(in);
    assertEquals(value, value2);
  }

  /**
   * Tests data serializing a {@link Boolean} using {@link DataSerializer#writeObject}.
   */
  @Test
  public void testBooleanObject() throws Exception {

    Boolean value = getRandom().nextInt() % 2 == 0;

    DataOutputStream out = getDataOutput();
    DataSerializer.writeObject(value, out);
    out.flush();

    DataInput in = getDataInput();
    Boolean value2 = DataSerializer.readObject(in);
    assertEquals(value, value2);
  }

  @Test
  public void testWriteObjectAsByteArray() throws Exception {
    // make sure recursive calls to WriteObjectAsByteArray work to test bug 38194
    Object v = new WOABA();
    DataOutputStream out = getDataOutput();
    DataSerializer.writeObjectAsByteArray(v, out);
    out.flush();

    DataInput in = getDataInput();
    byte[] b2 = DataSerializer.readByteArray(in);
    // todo should we deserislize the byte[] and make sure it is equal to v?
    ByteArrayInputStream bais = new ByteArrayInputStream(b2);
    DataInputStream dis = new DataInputStream(bais);
    Object v2 = DataSerializer.readObject(dis);
    if (!(v2 instanceof WOABA)) {
      fail("expected instance of WOABA but found " + v2.getClass());
    }
  }

  private static class WOABA implements DataSerializable {
    private byte[] deserialized;
    private WOABA2 f = new WOABA2();

    public WOABA() {}

    public void validate() throws Exception {
      ByteArrayInputStream bais = new ByteArrayInputStream(deserialized);
      DataInputStream dis = new DataInputStream(bais);
      Object v = DataSerializer.readObject(dis);
      if (!(v instanceof WOABA2)) {
        fail("expected instance of WOABA2 but found " + v.getClass());
      }
      f = (WOABA2) v;
    }

    @Override
    public void toData(DataOutput out) throws IOException {
      DataSerializer.writeObjectAsByteArray(f, out);
    }

    @Override
    public void fromData(DataInput in) throws IOException, ClassNotFoundException {
      deserialized = DataSerializer.readByteArray(in);
    }
  }

  private static class WOABA2 implements DataSerializable {

    public WOABA2() {}

    @Override
    public void toData(DataOutput out) throws IOException {
      String f = "foobar";
      DataSerializer.writeObjectAsByteArray(f, out);
    }

    @Override
    public void fromData(DataInput in) throws IOException, ClassNotFoundException {
      DataSerializer.readByteArray(in);
    }
  }

  /**
   * Tests data serializing a {@link Character}
   */
  @Test
  public void testCharacter() throws Exception {
    Character value = (char) ('A' + getRandom().nextInt('Z' - 'A'));

    DataOutputStream out = getDataOutput();
    DataSerializer.writeCharacter(value, out);
    out.flush();

    DataInput in = getDataInput();
    Character value2 = DataSerializer.readCharacter(in);
    assertEquals(value, value2);
  }

  /**
   * Tests data serializing a {@link Character} using {@link DataSerializer#writeObject}.
   */
  @Test
  public void testCharacterObject() throws Exception {
    Character value = (char) ('A' + getRandom().nextInt('Z' - 'A'));

    DataOutputStream out = getDataOutput();
    DataSerializer.writeObject(value, out);
    out.flush();

    DataInput in = getDataInput();
    Character value2 = DataSerializer.readObject(in);
    assertEquals(value, value2);
  }

  /**
   * Tests data serializing a {@link Byte}
   */
  @Test
  public void testByte() throws Exception {
    Byte value = (byte) getRandom().nextInt();

    DataOutputStream out = getDataOutput();
    DataSerializer.writeByte(value, out);
    out.flush();

    DataInput in = getDataInput();
    Byte value2 = DataSerializer.readByte(in);
    assertEquals(value, value2);
  }

  /**
   * Tests data serializing a {@link Byte} using {@link DataSerializer#writeObject}.
   */
  @Test
  public void testByteObject() throws Exception {
    Byte value = (byte) getRandom().nextInt();

    DataOutputStream out = getDataOutput();
    DataSerializer.writeObject(value, out);
    out.flush();

    DataInput in = getDataInput();
    Byte value2 = DataSerializer.readObject(in);
    assertEquals(value, value2);
  }

  /**
   * Tests data serializing a {@link Short}
   */
  @Test
  public void testShort() throws IOException {
    Short value = (short) getRandom().nextInt();

    DataOutputStream out = getDataOutput();
    DataSerializer.writeShort(value, out);
    out.flush();

    DataInput in = getDataInput();
    Short value2 = DataSerializer.readShort(in);
    assertEquals(value, value2);
  }

  /**
   * Tests data serializing a {@link Short} using {@link DataSerializer#writeObject}.
   */
  @Test
  public void testShortObject() throws IOException, ClassNotFoundException {

    Short value = (short) getRandom().nextInt();

    DataOutputStream out = getDataOutput();
    DataSerializer.writeObject(value, out);
    out.flush();

    DataInput in = getDataInput();
    Short value2 = DataSerializer.readObject(in);
    assertEquals(value, value2);
  }

  /**
   * Tests data serializing a {@link Integer}
   */
  @Test
  public void testInteger() throws Exception {
    Integer value = getRandom().nextInt();

    DataOutputStream out = getDataOutput();
    DataSerializer.writeInteger(value, out);
    out.flush();

    DataInput in = getDataInput();
    Integer value2 = DataSerializer.readInteger(in);
    assertEquals(value, value2);
  }

  /**
   * Tests data serializing a {@link Integer} using {@link DataSerializer#writeObject}.
   */
  @Test
  public void testIntegerObject() throws Exception {
    Integer value = getRandom().nextInt();

    DataOutputStream out = getDataOutput();
    DataSerializer.writeObject(value, out);
    out.flush();

    DataInput in = getDataInput();
    Integer value2 = DataSerializer.readObject(in);
    assertEquals(value, value2);
  }

  /**
   * Tests data serializing a {@link Long}
   */
  @Test
  public void testLong() throws Exception {
    Long value = getRandom().nextLong();

    DataOutputStream out = getDataOutput();
    DataSerializer.writeLong(value, out);
    out.flush();

    DataInput in = getDataInput();
    Long value2 = DataSerializer.readLong(in);
    assertEquals(value, value2);
  }

  /**
   * Tests data serializing a {@link Long} using {@link DataSerializer#writeObject}.
   */
  @Test
  public void testLongObject() throws Exception {
    Long value = getRandom().nextLong();

    DataOutputStream out = getDataOutput();
    DataSerializer.writeObject(value, out);
    out.flush();

    DataInput in = getDataInput();
    Long value2 = DataSerializer.readObject(in);
    assertEquals(value, value2);
  }

  /**
   * Tests data serializing a {@link Float}
   */
  @Test
  public void testFloat() throws Exception {
    Float value = getRandom().nextFloat();

    DataOutputStream out = getDataOutput();
    DataSerializer.writeFloat(value, out);
    out.flush();

    DataInput in = getDataInput();
    Float value2 = DataSerializer.readFloat(in);
    assertEquals(value, value2);
  }

  /**
   * Tests data serializing a {@link Float} using {@link DataSerializer#writeObject}.
   */
  @Test
  public void testFloatObject() throws Exception {
    Float value = getRandom().nextFloat();

    DataOutputStream out = getDataOutput();
    DataSerializer.writeObject(value, out);
    out.flush();

    DataInput in = getDataInput();
    Float value2 = DataSerializer.readObject(in);
    assertEquals(value, value2);
  }

  /**
   * Tests data serializing a {@link Double}
   */
  @Test
  public void testDouble() throws Exception {
    Double value = getRandom().nextDouble();

    DataOutputStream out = getDataOutput();
    DataSerializer.writeDouble(value, out);
    out.flush();

    DataInput in = getDataInput();
    Double value2 = DataSerializer.readDouble(in);
    assertEquals(value, value2);
  }

  /**
   * Tests data serializing a {@link Double} using {@link DataSerializer#writeObject}.
   */
  @Test
  public void testDoubleObject() throws Exception {
    Double value = getRandom().nextDouble();

    DataOutputStream out = getDataOutput();
    DataSerializer.writeObject(value, out);
    out.flush();

    DataInput in = getDataInput();
    Double value2 = DataSerializer.readObject(in);
    assertEquals(value, value2);
  }

  /**
   * Tests data serializing a <code>byte</code> array
   */
  @Test
  public void testByteArray() throws Exception {
    byte[] array = new byte[] {(byte) 4, (byte) 5, (byte) 6};

    DataOutputStream out = getDataOutput();
    DataSerializer.writeByteArray(array, out);
    DataSerializer.writeObject(array, out);
    out.flush();

    DataInput in = getDataInput();
    for (int idx = 0; idx < 2; idx++) {
      byte[] array2 =
          (idx == 0) ? DataSerializer.readByteArray(in) : (byte[]) DataSerializer.readObject(in);
      assertEquals(array.length, array2.length);
      for (int i = 0; i < array.length; i++) {
        assertEquals(array[i], array2[i]);
      }
    }
  }

  /**
   * Tests data serializing a <code>byte</code> array using {@link DataSerializer#writeObject}.
   */
  @Test
  public void testByteArrayObject() throws Exception {
    byte[] array = new byte[] {(byte) 4, (byte) 5, (byte) 6};

    DataOutputStream out = getDataOutput();
    DataSerializer.writeObject(array, out);
    out.flush();

    DataInput in = getDataInput();
    byte[] array2 = DataSerializer.readObject(in);

    assertEquals(array.length, array2.length);
    for (int i = 0; i < array.length; i++) {
      assertEquals(array[i], array2[i]);
    }
  }

  /**
   * Tests data serializing a <code>short</code> array
   */
  @Test
  public void testShortArray() throws Exception {
    short[] array = new short[] {(short) 4, (short) 5, (short) 6};

    DataOutputStream out = getDataOutput();
    DataSerializer.writeShortArray(array, out);
    out.flush();

    DataInput in = getDataInput();
    short[] array2 = DataSerializer.readShortArray(in);

    assertEquals(array.length, array2.length);
    for (int i = 0; i < array.length; i++) {
      assertEquals(array[i], array2[i]);
    }
  }

  /**
   * Tests data serializing a <code>short</code> array using {@link DataSerializer#writeObject}.
   */
  @Test
  public void testShortArrayObject() throws Exception {
    short[] array = new short[] {(short) 4, (short) 5, (short) 6};

    DataOutputStream out = getDataOutput();
    DataSerializer.writeObject(array, out);
    out.flush();

    DataInput in = getDataInput();
    short[] array2 = DataSerializer.readObject(in);

    assertEquals(array.length, array2.length);
    for (int i = 0; i < array.length; i++) {
      assertEquals(array[i], array2[i]);
    }
  }

  /**
   * Tests data serializing a <code>String</code> array
   */
  @Test
  public void testStringArray() throws Exception {
    Random random = getRandom();

    String[] array = new String[] {String.valueOf(random.nextLong()),
        String.valueOf(random.nextLong()), String.valueOf(random.nextLong())};

    DataOutputStream out = getDataOutput();
    DataSerializer.writeStringArray(array, out);
    out.flush();

    DataInput in = getDataInput();
    String[] array2 = DataSerializer.readStringArray(in);

    assertEquals(array.length, array2.length);
    for (int i = 0; i < array.length; i++) {
      assertEquals(array[i], array2[i]);
    }
  }

  /**
   * Tests data serializing a <code>String</code> array that contains a <code>null</code>
   * <code>String</code>.
   */
  @Test
  public void testStringArrayWithNull() throws Exception {
    Random random = getRandom();

    String[] array =
        new String[] {String.valueOf(random.nextLong()), null, String.valueOf(random.nextLong())};

    DataOutputStream out = getDataOutput();
    DataSerializer.writeStringArray(array, out);
    out.flush();

    DataInput in = getDataInput();
    String[] array2 = DataSerializer.readStringArray(in);

    assertEquals(array.length, array2.length);
    for (int i = 0; i < array.length; i++) {
      assertEquals(array[i], array2[i]);
    }
  }

  /**
   * Tests data serializing a <code>String</code> array using {@link DataSerializer#writeObject}.
   */
  @Test
  public void testStringArrayObject() throws Exception {
    Random random = getRandom();

    String[] array = new String[] {String.valueOf(random.nextLong()),
        String.valueOf(random.nextLong()), String.valueOf(random.nextLong())};

    DataOutputStream out = getDataOutput();
    DataSerializer.writeObject(array, out);
    out.flush();

    DataInput in = getDataInput();
    String[] array2 = DataSerializer.readObject(in);

    assertEquals(array.length, array2.length);
    for (int i = 0; i < array.length; i++) {
      assertEquals(array[i], array2[i]);
    }
  }

  /**
   * Tests data serializing a <code>int</code> array
   */
  @Test
  public void testIntArray() throws Exception {
    int[] array = new int[] {4, 5, 6};

    DataOutputStream out = getDataOutput();
    DataSerializer.writeIntArray(array, out);
    out.flush();

    DataInput in = getDataInput();
    int[] array2 = DataSerializer.readIntArray(in);

    assertEquals(array.length, array2.length);
    for (int i = 0; i < array.length; i++) {
      assertEquals(array[i], array2[i]);
    }
  }

  /**
   * Tests data serializing a <code>int</code> array using {@link DataSerializer#writeObject}.
   */
  @Test
  public void testIntArrayObject() throws Exception {
    int[] array = new int[] {4, 5, 6};

    DataOutputStream out = getDataOutput();
    DataSerializer.writeObject(array, out);
    out.flush();

    DataInput in = getDataInput();
    int[] array2 = DataSerializer.readObject(in);

    assertEquals(array.length, array2.length);
    for (int i = 0; i < array.length; i++) {
      assertEquals(array[i], array2[i]);
    }
  }

  /**
   * Tests data serializing a <code>long</code> array
   */
  @Test
  public void testLongArray() throws Exception {
    long[] array = new long[] {4, 5, 6};

    DataOutputStream out = getDataOutput();
    DataSerializer.writeLongArray(array, out);
    out.flush();

    DataInput in = getDataInput();
    long[] array2 = DataSerializer.readLongArray(in);

    assertEquals(array.length, array2.length);
    for (int i = 0; i < array.length; i++) {
      assertEquals(array[i], array2[i]);
    }
  }

  /**
   * Tests data serializing a <code>long</code> array using {@link DataSerializer#writeObject}.
   */
  @Test
  public void testLongArrayObject() throws Exception {
    long[] array = new long[] {4, 5, 6};

    DataOutputStream out = getDataOutput();
    DataSerializer.writeObject(array, out);
    out.flush();

    DataInput in = getDataInput();
    long[] array2 = DataSerializer.readObject(in);

    assertEquals(array.length, array2.length);
    for (int i = 0; i < array.length; i++) {
      assertEquals(array[i], array2[i]);
    }
  }

  /**
   * Tests data serializing a <code>float</code> array
   */
  @Test
  public void testFloatArray() throws Exception {
    float[] array = new float[] {(float) 4.0, (float) 5.0, (float) 6.0};

    DataOutputStream out = getDataOutput();
    DataSerializer.writeFloatArray(array, out);
    out.flush();

    DataInput in = getDataInput();
    float[] array2 = DataSerializer.readFloatArray(in);

    assertEquals(array.length, array2.length);
    for (int i = 0; i < array.length; i++) {
      assertEquals(array[i], array2[i], 0.0f);
    }
  }

  /**
   * Tests data serializing a <code>float</code> array using {@link DataSerializer#writeObject}.
   */
  @Test
  public void testFloatArrayObject() throws Exception {
    float[] array = new float[] {(float) 4.0, (float) 5.0, (float) 6.0};

    DataOutputStream out = getDataOutput();
    DataSerializer.writeObject(array, out);
    out.flush();

    DataInput in = getDataInput();
    float[] array2 = DataSerializer.readObject(in);

    assertEquals(array.length, array2.length);
    for (int i = 0; i < array.length; i++) {
      assertEquals(array[i], array2[i], 0.0f);
    }
  }

  /**
   * Tests data serializing a <code>double</code> array
   */
  @Test
  public void testDoubleArray() throws Exception {
    double[] array = new double[] {4.0, 5.0, 6.0};

    DataOutputStream out = getDataOutput();
    DataSerializer.writeDoubleArray(array, out);
    out.flush();

    DataInput in = getDataInput();
    double[] array2 = DataSerializer.readDoubleArray(in);

    assertEquals(array.length, array2.length);
    for (int i = 0; i < array.length; i++) {
      assertEquals(array[i], array2[i], 0.0f);
    }
  }

  /**
   * Tests data serializing a <code>double</code> array using {@link DataSerializer#writeObject}.
   */
  @Test
  public void testDoubleArrayObject() throws Exception {
    double[] array = new double[] {4.0, 5.0, 6.0};

    DataOutputStream out = getDataOutput();
    DataSerializer.writeObject(array, out);
    out.flush();

    DataInput in = getDataInput();
    double[] array2 = DataSerializer.readObject(in);

    assertEquals(array.length, array2.length);
    for (int i = 0; i < array.length; i++) {
      assertEquals(array[i], array2[i], 0.0f);
    }
  }

  /**
   * Tests data serializing a <code>Object</code> array
   */
  @Test
  public void testObjectArray() throws Exception {
    Random random = getRandom();
    SerializableImpl[] array = new SerializableImpl[] {new SerializableImpl(random),
        new SerializableImpl(random), new SerializableImpl(random)};

    DataOutputStream out = getDataOutput();
    DataSerializer.writeObjectArray(array, out);
    out.flush();

    DataInput in = getDataInput();
    SerializableImpl[] array2 = (SerializableImpl[]) DataSerializer.readObjectArray(in);

    assertEquals(array.length, array2.length);
    for (int i = 0; i < array.length; i++) {
      assertEquals(array[i], array2[i]);
    }
  }

  /**
   * Tests data serializing a <code>Object</code> array using {@link DataSerializer#writeObject}.
   */
  @Test
  public void testObjectArrayObject() throws Exception {
    Random random = getRandom();
    SerializableImpl[] array = new SerializableImpl[] {new SerializableImpl(random),
        new SerializableImpl(random), new SerializableImpl(random)};

    DataOutputStream out = getDataOutput();
    DataSerializer.writeObject(array, out);
    out.flush();

    DataInput in = getDataInput();
    SerializableImpl[] array2 = DataSerializer.readObject(in);

    assertEquals(array.length, array2.length);
    for (int i = 0; i < array.length; i++) {
      assertEquals(array[i], array2[i]);
    }
  }

  /**
   * Tests serializing an object that is {@link DataSerializableJUnitTest.SerializableImpl not
   * specially cased}.
   */
  @Test
  public void testUnspecialObject() throws Exception {
    Object o = new SerializableImpl(getRandom());

    DataOutputStream out = getDataOutput();
    DataSerializer.writeObject(o, out);
    out.flush();

    DataInput in = getDataInput();
    Object o2 = DataSerializer.readObject(in);
    assertEquals(o, o2);
  }

  /**
   * Tests serializing an object that implements {@link DataSerializable}
   */
  @Test
  public void testDataSerializable() throws Exception {
    DataSerializable ds = new DataSerializableImpl(getRandom());

    DataOutputStream out = getDataOutput();
    ds.toData(out);
    out.flush();

    DataSerializable ds2 = new DataSerializableImpl();
    ds2.fromData(getDataInput());

    assertEquals(ds, ds2);
  }

  /**
   * Tests serializing an object that implements {@link DataSerializable}
   */
  @Test
  public void testVersionedDataSerializable() throws Exception {
    VersionedDataSerializableImpl ds = new VersionedDataSerializableImpl(getRandom());

    VersionedDataOutputStream v =
        new VersionedDataOutputStream(baos, KnownVersion.GEODE_1_11_0);
    DataSerializer.writeObject(ds, v);
    v.flush();

    ByteBuffer bb = ByteBuffer.wrap(baos.toByteArray());
