/*
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License. You may obtain a
 * copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */
package org.apache.geode.pdx;

import static org.apache.geode.cache.Region.SEPARATOR;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInput;
import java.io.DataInputStream;
import java.io.File;
import java.io.IOException;
import java.io.NotSerializableException;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.regex.Pattern;

import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.experimental.categories.Category;

import org.apache.geode.CopyHelper;
import org.apache.geode.DataSerializable;
import org.apache.geode.DataSerializer;
import org.apache.geode.DeltaTestImpl;
import org.apache.geode.ToDataException;
import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.DiskStoreFactory;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.internal.HeapDataOutputStream;
import org.apache.geode.internal.PdxSerializerObject;
import org.apache.geode.internal.SystemAdmin;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.serialization.DSCODE;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.internal.tcp.ByteBufferInputStream.ByteSourceFactory;
import org.apache.geode.internal.util.ArrayUtils;
import org.apache.geode.pdx.internal.DataSize;
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.PeerTypeRegistration;
import org.apache.geode.pdx.internal.TypeRegistry;
import org.apache.geode.test.junit.categories.SerializationTest;

@Category({SerializationTest.class})
public class PdxSerializableJUnitTest {

  private GemFireCacheImpl cache;

  @Before
  public void setUp() {
    // make it a loner
    cache = (GemFireCacheImpl) new CacheFactory().set(MCAST_PORT, "0").create();
  }

  @After
  public void tearDown() {
    cache.close();
  }

  private int getPdxTypeIdForClass(Class c) {
    // here we are assuming Dsid == 0
    return cache.getPdxRegistry().getExistingTypeForClass(c).hashCode()
        & PeerTypeRegistration.PLACE_HOLDER_FOR_TYPE_ID;
  }

  private int getNumPdxTypes() {
    return cache.getPdxRegistry().typeMap().size();
  }

  @Test
  public void testNoDiskStore() throws Exception {
    cache.close();
    cache = (GemFireCacheImpl) new CacheFactory().set(MCAST_PORT, "0").setPdxPersistent(true)
        .setPdxDiskStore("doesNotExist").create();
    HeapDataOutputStream out = new HeapDataOutputStream(KnownVersion.CURRENT);
    PdxSerializable object = new SimpleClass(1, (byte) 5, null);
    try {
      DataSerializer.writeObject(object, out);
      throw new Exception("expected PdxInitializationException");
    } catch (PdxInitializationException ignored) {
    }
  }

  // for bugs 44271 and 44914
  @Test
  public void testPdxPersistentKeys() throws Exception {
    cache.close();
    cache = (GemFireCacheImpl) new CacheFactory().set(MCAST_PORT, "0").setPdxPersistent(true)
        .setPdxDiskStore("pdxDS").create();
    try {
      DiskStoreFactory dsf = cache.createDiskStoreFactory();
      dsf.create("pdxDS");
      cache.createDiskStoreFactory().create("r2DS");
      Region r1 = cache.createRegionFactory(RegionShortcut.LOCAL_PERSISTENT).create("r1");
      r1.put(new SimpleClass(1, (byte) 1), "1");
      r1.put(new SimpleClass(2, (byte) 2), "2");
      r1.put(new SimpleClass(1, (byte) 1), "1.2"); // so we have something to compact offline
      Region r2 = cache.createRegionFactory(RegionShortcut.LOCAL_PERSISTENT)
          .setDiskStoreName("r2DS").create("r2");
      r2.put(new SimpleClass(1, (byte) 1), new SimpleClass(1, (byte) 1));
      r2.put(new SimpleClass(2, (byte) 2), new SimpleClass(2, (byte) 2));
      cache.close();
      cache = (GemFireCacheImpl) new CacheFactory().set(MCAST_PORT, "0").setPdxPersistent(true)
          .setPdxDiskStore("pdxDS").create();
      dsf = cache.createDiskStoreFactory();
      dsf.create("pdxDS");
      cache.createDiskStoreFactory().create("r2DS");
      r1 = cache.createRegionFactory(RegionShortcut.LOCAL_PERSISTENT).create("r1");
      r2 = cache.createRegionFactory(RegionShortcut.LOCAL_PERSISTENT).setDiskStoreName("r2DS")
          .create("r2");
      assertEquals(true, r1.containsKey(new SimpleClass(1, (byte) 1)));
      assertEquals(true, r1.containsKey(new SimpleClass(2, (byte) 2)));
      assertEquals(true, r2.containsKey(new SimpleClass(1, (byte) 1)));
      assertEquals(true, r2.containsKey(new SimpleClass(2, (byte) 2)));
      assertEquals(new SimpleClass(1, (byte) 1), r2.get(new SimpleClass(1, (byte) 1)));
      assertEquals(new SimpleClass(2, (byte) 2), r2.get(new SimpleClass(2, (byte) 2)));
      cache.close();
      // use a cache.xml to recover
      cache = (GemFireCacheImpl) new CacheFactory().set(MCAST_PORT, "0").create();
      ByteArrayOutputStream baos = new ByteArrayOutputStream();
      PrintWriter pw = new PrintWriter(new OutputStreamWriter(baos), true);
      pw.println("<?xml version=\"1.0\"?>");
      pw.println("<!DOCTYPE cache PUBLIC");
      pw.println("  \"-//GemStone Systems, Inc.//GemFire Declarative Caching 7.0//EN\"");
      pw.println("  \"http://www.gemstone.com/dtd/cache7_0.dtd\">");
      pw.println("<cache>");
      pw.println("  <disk-store name=\"r2DS\"/>");
      pw.println("  <disk-store name=\"pdxDS\"/>");
      pw.println("  <pdx persistent=\"true\" disk-store-name=\"pdxDS\"/>");
      pw.println("  <region name=\"r1\" refid=\"LOCAL_PERSISTENT\"/>");
      pw.println("  <region name=\"r2\" refid=\"LOCAL_PERSISTENT\">");
      pw.println("    <region-attributes disk-store-name=\"r2DS\"/>");
      pw.println("  </region>");
      pw.println("</cache>");
      pw.close();
      byte[] bytes = baos.toByteArray();
      cache.loadCacheXml(new ByteArrayInputStream(bytes));
      r1 = cache.getRegion(SEPARATOR + "r1");
      r2 = cache.getRegion(SEPARATOR + "r2");
      assertEquals(true, r1.containsKey(new SimpleClass(1, (byte) 1)));
      assertEquals(true, r1.containsKey(new SimpleClass(2, (byte) 2)));
      assertEquals(true, r2.containsKey(new SimpleClass(1, (byte) 1)));
      assertEquals(true, r2.containsKey(new SimpleClass(2, (byte) 2)));
      assertEquals(new SimpleClass(1, (byte) 1), r2.get(new SimpleClass(1, (byte) 1)));
      assertEquals(new SimpleClass(2, (byte) 2), r2.get(new SimpleClass(2, (byte) 2)));
      cache.close();
      // make sure offlines tools work with disk store that has pdx keys
      SystemAdmin.validateDiskStore("DEFAULT", ".");
      SystemAdmin.compactDiskStore("DEFAULT", ".");
      SystemAdmin.modifyDiskStore("DEFAULT", ".");
      SystemAdmin.validateDiskStore("r2DS", ".");
      SystemAdmin.compactDiskStore("r2DS", ".");
      SystemAdmin.modifyDiskStore("r2DS", ".");
      SystemAdmin.validateDiskStore("pdxDS", ".");
      SystemAdmin.compactDiskStore("pdxDS", ".");
      SystemAdmin.modifyDiskStore("pdxDS", ".");
    } finally {
      try {
        cache.close();
      } finally {
        Pattern pattern = Pattern.compile("BACKUP(DEFAULT|pdxDS|r2DS).*");
        File[] files = new File(".").listFiles((dir1, name) -> pattern.matcher(name).matches());
        if (files != null) {
          for (File file : files) {
            Files.delete(file.toPath());
          }
        }
      }
    }
  }

  @Test
  public void testPdxPersistentKeysDefDS() throws Exception {
    cache.close();
    cache =
        (GemFireCacheImpl) new CacheFactory().set(MCAST_PORT, "0").setPdxPersistent(true).create();
    try {
      cache.createDiskStoreFactory().create("r2DS");
      Region r1 = cache.createRegionFactory(RegionShortcut.LOCAL_PERSISTENT)
          .setDiskStoreName("r2DS").create("r1");
      r1.put(new SimpleClass(1, (byte) 1), "1");
      r1.put(new SimpleClass(2, (byte) 2), "2");
      r1.put(new SimpleClass(1, (byte) 1), "1.2"); // so we have something to compact offline
      Region r2 = cache.createRegionFactory(RegionShortcut.LOCAL_PERSISTENT)
          .setDiskStoreName("r2DS").create("r2");
      r2.put(new SimpleClass(1, (byte) 1), new SimpleClass(1, (byte) 1));
      r2.put(new SimpleClass(2, (byte) 2), new SimpleClass(2, (byte) 2));
      cache.close();
      cache = (GemFireCacheImpl) new CacheFactory().set(MCAST_PORT, "0").setPdxPersistent(true)
          .create();
      cache.createDiskStoreFactory().create("r2DS");
      r1 = cache.createRegionFactory(RegionShortcut.LOCAL_PERSISTENT).setDiskStoreName("r2DS")
          .create("r1");
      r2 = cache.createRegionFactory(RegionShortcut.LOCAL_PERSISTENT).setDiskStoreName("r2DS")
          .create("r2");
      assertEquals(true, r1.containsKey(new SimpleClass(1, (byte) 1)));
      assertEquals(true, r1.containsKey(new SimpleClass(2, (byte) 2)));
      assertEquals(true, r2.containsKey(new SimpleClass(1, (byte) 1)));
      assertEquals(true, r2.containsKey(new SimpleClass(2, (byte) 2)));
      assertEquals(new SimpleClass(1, (byte) 1), r2.get(new SimpleClass(1, (byte) 1)));
      assertEquals(new SimpleClass(2, (byte) 2), r2.get(new SimpleClass(2, (byte) 2)));
      cache.close();
      // use a cache.xml to recover
      cache = (GemFireCacheImpl) new CacheFactory().set(MCAST_PORT, "0").create();
      ByteArrayOutputStream baos = new ByteArrayOutputStream();
      PrintWriter pw = new PrintWriter(new OutputStreamWriter(baos), true);
      pw.println("<?xml version=\"1.0\"?>");
      pw.println("<!DOCTYPE cache PUBLIC");
      pw.println("  \"-//GemStone Systems, Inc.//GemFire Declarative Caching 7.0//EN\"");
      pw.println("  \"http://www.gemstone.com/dtd/cache7_0.dtd\">");
      pw.println("<cache>");
      pw.println("  <disk-store name=\"r2DS\"/>");
      pw.println("  <pdx persistent=\"true\"/>");
      pw.println("  <region name=\"r1\" refid=\"LOCAL_PERSISTENT\">");
      pw.println("    <region-attributes disk-store-name=\"r2DS\"/>");
      pw.println("  </region>");
      pw.println("  <region name=\"r2\" refid=\"LOCAL_PERSISTENT\">");
      pw.println("    <region-attributes disk-store-name=\"r2DS\"/>");
      pw.println("  </region>");
      pw.println("</cache>");
      pw.close();
      byte[] bytes = baos.toByteArray();
      cache.loadCacheXml(new ByteArrayInputStream(bytes));
      r1 = cache.getRegion(SEPARATOR + "r1");
      r2 = cache.getRegion(SEPARATOR + "r2");
      assertEquals(true, r1.containsKey(new SimpleClass(1, (byte) 1)));
      assertEquals(true, r1.containsKey(new SimpleClass(2, (byte) 2)));
      assertEquals(true, r2.containsKey(new SimpleClass(1, (byte) 1)));
      assertEquals(true, r2.containsKey(new SimpleClass(2, (byte) 2)));
      assertEquals(new SimpleClass(1, (byte) 1), r2.get(new SimpleClass(1, (byte) 1)));
      assertEquals(new SimpleClass(2, (byte) 2), r2.get(new SimpleClass(2, (byte) 2)));
      cache.close();
      // make sure offlines tools work with disk store that has pdx keys
      SystemAdmin.validateDiskStore("DEFAULT", ".");
      SystemAdmin.compactDiskStore("DEFAULT", ".");
      SystemAdmin.modifyDiskStore("DEFAULT", ".");
      SystemAdmin.validateDiskStore("r2DS", ".");
      SystemAdmin.compactDiskStore("r2DS", ".");
      SystemAdmin.modifyDiskStore("r2DS", ".");
    } finally {
      try {
        cache.close();
      } finally {
        Pattern pattern = Pattern.compile("BACKUP(DEFAULT|r2DS).*");
        File[] files = new File(".").listFiles((dir1, name) -> pattern.matcher(name).matches());
        if (files != null) {
          for (File file : files) {
            Files.delete(file.toPath());
          }
        }
      }
    }
  }

  @Test
  public void testByteFormatForSimpleClass() throws Exception {
    HeapDataOutputStream out = new HeapDataOutputStream(KnownVersion.CURRENT);
    PdxSerializable object = new SimpleClass(1, (byte) 5, null);
    DataSerializer.writeObject(object, out);
    int typeId = getPdxTypeIdForClass(SimpleClass.class);
    byte[] actual = out.toByteArray();
    byte[] expected = new byte[] {DSCODE.PDX.toByte(), // byte
        0, 0, 0, 4 + 1 + 1, // int - length of byte stream = 4(myInt) + 1(myByte) + 1 (myEnum)
        (byte) (typeId >> 24), (byte) (typeId >> 16), (byte) (typeId >> 8), (byte) typeId, // int -
                                                                                           // typeId
        0, 0, 0, 1, // int - myInt = 1
        5, // byte - myByte = 5
        DSCODE.NULL.toByte()};

    StringBuilder msg = new StringBuilder("Actual output: ");
    for (byte val : actual) {
      msg.append(val + ", ");
    }
    msg.append("\nExpected output: ");
    for (byte val : expected) {
      msg.append(val + ", ");
    }
    assertTrue("Mismatch in length, actual.length: " + actual.length + " and expected length: "
        + expected.length, actual.length == expected.length);
    for (int i = 0; i < actual.length; i++) {
      if (actual[i] != expected[i]) {
        System.out.println(msg);
      }
      assertTrue("Mismatch at index " + i, actual[i] == expected[i]);
    }
    System.out.println("\n");

    DataInput in = new DataInputStream(new ByteArrayInputStream(actual));
    SimpleClass actualVal = DataSerializer.readObject(in);
    // System.out.println("actualVal..."+actualVal);
    assertTrue(
        "Mismatch in write and read value: Value Write..." + object + " Value Read..." + actualVal,
        object.equals(actualVal));

  }

  @Test
  public void testByteFormatForStrings() throws Exception {
    boolean myFlag = true;
    short myShort = 25;
    String myString1 = "Class4_myString1";
    long myLong = 15654;
    String myString2 = "Class4_myString2";
    String myString3 = "Class4_myString3";
    int myInt = 1420;
    float myFloat = 123.023f;

    HeapDataOutputStream out = new HeapDataOutputStream(KnownVersion.CURRENT);
    SimpleClass1 pdx =
        new SimpleClass1(myFlag, myShort, myString1, myLong, myString2, myString3, myInt, myFloat);
    DataSerializer.writeObject(pdx, out);
    int typeId = getPdxTypeIdForClass(SimpleClass1.class);

    HeapDataOutputStream hdos1 = new HeapDataOutputStream(KnownVersion.CURRENT);
    DataSerializer.writeString(myString1, hdos1);
    byte[] str1Bytes = hdos1.toByteArray();
    HeapDataOutputStream hdos2 = new HeapDataOutputStream(KnownVersion.CURRENT);
    DataSerializer.writeString(myString2, hdos2);
    byte[] str2Bytes = hdos2.toByteArray();
    HeapDataOutputStream hdos3 = new HeapDataOutputStream(KnownVersion.CURRENT);
    DataSerializer.writeString(myString3, hdos3);
    byte[] str3Bytes = hdos3.toByteArray();

    int length = 1 /* myFlag */ + 2 /* myShort */
        + 8 /* myLong */ + 4 /* myInt */ + 4 /* myFloat */ + str1Bytes.length + str2Bytes.length
        + str3Bytes.length + 2 /* byte offset for 3 strings */;

    int offset1 = 1 + 2;
    int offset2 = offset1 + 8 + str1Bytes.length;
    int offset3 = offset1 + 8 + str1Bytes.length + str2Bytes.length;
    byte[] actual = out.toByteArray();
    int floatBytes = Float.floatToRawIntBits(myFloat);
    Byte[] expected = new Byte[] {DSCODE.PDX.toByte(), // byte
        (byte) (length >> 24), (byte) (length >> 16), (byte) (length >> 8), (byte) length, // int -
                                                                                           // length
                                                                                           // of
                                                                                           // byte
                                                                                           // stream
        (byte) (typeId >> 24), (byte) (typeId >> 16), (byte) (typeId >> 8), (byte) typeId, // int -
                                                                                           // typeId
        1, // boolean - myFlag = true
        (byte) (myShort >> 8), (byte) myShort, // short - myShort
        (byte) (myLong >> 56), (byte) (myLong >> 48), (byte) (myLong >> 40), (byte) (myLong >> 32),
        (byte) (myLong >> 24), (byte) (myLong >> 16), (byte) (myLong >> 8), (byte) myLong, // long -
                                                                                           // myLong
        (byte) (myInt >> 24), (byte) (myInt >> 16), (byte) (myInt >> 8), (byte) myInt, // int -
                                                                                       // myInt
        (byte) (floatBytes >> 24), (byte) (floatBytes >> 16), (byte) (floatBytes >> 8),
        (byte) floatBytes, // float - myFloat
        (byte) offset3, // offset of myString3
        (byte) offset2, // offset of myString2
    };

    for (int i = (str1Bytes.length - 1); i >= 0; i--) {
      expected =
          (Byte[]) ArrayUtils.insert(expected, offset1 + PdxWriterImpl.HEADER_SIZE, str1Bytes[i]); // +
                                                                                                   // 5
                                                                                                   // for:
                                                                                                   // 1
                                                                                                   // for
                                                                                                   // DSCODE.PDX.toByte()
                                                                                                   // and
                                                                                                   // 4
                                                                                                   // for
                                                                                                   // byte
                                                                                                   // stream
                                                                                                   // length
    }
    for (int i = (str2Bytes.length - 1); i >= 0; i--) {
      expected =
          (Byte[]) ArrayUtils.insert(expected, offset2 + PdxWriterImpl.HEADER_SIZE, str2Bytes[i]);
    }
    for (int i = (str3Bytes.length - 1); i >= 0; i--) {
      expected =
          (Byte[]) ArrayUtils.insert(expected, offset3 + PdxWriterImpl.HEADER_SIZE, str3Bytes[i]);
    }

    StringBuilder msg = new StringBuilder("Actual output: ");
    for (byte val : actual) {
      msg.append(val + ", ");
    }
    msg.append("\nExpected output: ");
    for (byte val : expected) {
      msg.append(val + ", ");
    }
    if (actual.length != expected.length) {
      System.out.println(msg);
    }
    assertTrue("Mismatch in length, actual.length: " + actual.length + " and expected length: "
        + expected.length, actual.length == expected.length);
    for (int i = 0; i < actual.length; i++) {
      if (actual[i] != expected[i]) {
        System.out.println(msg);
      }
      assertTrue("Mismatch at index " + i, actual[i] == expected[i]);
    }
    System.out.println("\n");

    DataInput in = new DataInputStream(new ByteArrayInputStream(actual));
    SimpleClass1 actualVal = DataSerializer.readObject(in);
    // System.out.println("actualVal..."+actualVal);
    assertTrue(
        "Mismatch in write and read value: Value Write..." + pdx + " Value Read..." + actualVal,
        pdx.equals(actualVal));

    cache.setReadSerializedForTest(true);
    try {
      in = new DataInputStream(new ByteArrayInputStream(actual));
      PdxInstance pi = DataSerializer.readObject(in);
      actualVal = (SimpleClass1) pi.getObject();
      assertTrue(
          "Mismatch in write and read value: Value Write..." + pdx + " Value Read..." + actualVal,
          pdx.equals(actualVal));
      assertTrue(pi.hasField("myFlag"));
      assertTrue(pi.hasField("myShort"));
      assertTrue(pi.hasField("myString1"));
      assertTrue(pi.hasField("myLong"));
      assertTrue(pi.hasField("myString2"));
      assertTrue(pi.hasField("myString3"));
      assertTrue(pi.hasField("myInt"));
      assertTrue(pi.hasField("myFloat"));
      assertEquals(pdx.isMyFlag(), pi.getField("myFlag"));
      assertEquals(pdx.getMyShort(), pi.getField("myShort"));
      assertEquals(pdx.getMyString1(), pi.getField("myString1"));
      assertEquals(pdx.getMyLong(), pi.getField("myLong"));
      assertEquals(pdx.getMyString2(), pi.getField("myString2"));
      assertEquals(pdx.getMyString3(), pi.getField("myString3"));
      assertEquals(pdx.getMyInt(), pi.getField("myInt"));
      assertEquals(pdx.getMyFloat(), pi.getField("myFloat"));
      PdxReaderImpl reader = (PdxReaderImpl) pi;
      PdxType type = reader.getPdxType();
      assertEquals(SimpleClass1.class.getName(), type.getClassName());
      assertEquals(8, type.getFieldCount());
      assertEquals(2, type.getVariableLengthFieldCount());

      assertEquals(0, type.getPdxField("myFlag").getFieldIndex());
      assertEquals(1, type.getPdxField("myShort").getFieldIndex());
      assertEquals(2, type.getPdxField("myString1").getFieldIndex());
      assertEquals(3, type.getPdxField("myLong").getFieldIndex());
      assertEquals(4, type.getPdxField("myString2").getFieldIndex());
      assertEquals(5, type.getPdxField("myString3").getFieldIndex());
      assertEquals(6, type.getPdxField("myInt").getFieldIndex());
      assertEquals(7, type.getPdxField("myFloat").getFieldIndex());

      assertEquals(FieldType.BOOLEAN, type.getPdxField("myFlag").getFieldType());
      assertEquals(FieldType.SHORT, type.getPdxField("myShort").getFieldType());
      assertEquals(FieldType.STRING, type.getPdxField("myString1").getFieldType());
      assertEquals(FieldType.LONG, type.getPdxField("myLong").getFieldType());
      assertEquals(FieldType.STRING, type.getPdxField("myString2").getFieldType());
      assertEquals(FieldType.STRING, type.getPdxField("myString3").getFieldType());
      assertEquals(FieldType.INT, type.getPdxField("myInt").getFieldType());
      assertEquals(FieldType.FLOAT, type.getPdxField("myFloat").getFieldType());

      assertEquals("myFlag", type.getPdxField("myFlag").getFieldName());
      assertEquals("myShort", type.getPdxField("myShort").getFieldName());
      assertEquals("myString1", type.getPdxField("myString1").getFieldName());
      assertEquals("myLong", type.getPdxField("myLong").getFieldName());
      assertEquals("myString2", type.getPdxField("myString2").getFieldName());
      assertEquals("myString3", type.getPdxField("myString3").getFieldName());
      assertEquals("myInt", type.getPdxField("myInt").getFieldName());
      assertEquals("myFloat", type.getPdxField("myFloat").getFieldName());

      assertEquals(0, type.getPdxField("myFlag").getVarLenFieldSeqId());
      assertEquals(0, type.getPdxField("myShort").getVarLenFieldSeqId());
      assertEquals(0, type.getPdxField("myString1").getVarLenFieldSeqId());
      assertEquals(0, type.getPdxField("myLong").getVarLenFieldSeqId());
      assertEquals(1, type.getPdxField("myString2").getVarLenFieldSeqId());
      assertEquals(2, type.getPdxField("myString3").getVarLenFieldSeqId());
      assertEquals(2, type.getPdxField("myInt").getVarLenFieldSeqId());
      assertEquals(2, type.getPdxField("myFloat").getVarLenFieldSeqId());

      assertEquals(false, type.getPdxField("myFlag").isVariableLengthType());
      assertEquals(false, type.getPdxField("myShort").isVariableLengthType());
      assertEquals(true, type.getPdxField("myString1").isVariableLengthType());
      assertEquals(false, type.getPdxField("myLong").isVariableLengthType());
      assertEquals(true, type.getPdxField("myString2").isVariableLengthType());
      assertEquals(true, type.getPdxField("myString3").isVariableLengthType());
      assertEquals(false, type.getPdxField("myInt").isVariableLengthType());
      assertEquals(false, type.getPdxField("myFloat").isVariableLengthType());

      assertEquals(ByteSourceFactory.wrap(new byte[] {(byte) (pdx.isMyFlag() ? 1 : 0)}),
          reader.getRaw(0));
      assertEquals(
          ByteSourceFactory
              .wrap(new byte[] {(byte) (pdx.getMyShort() >> 8), (byte) pdx.getMyShort()}),
          reader.getRaw(1));
      assertEquals(ByteSourceFactory.wrap(str1Bytes), reader.getRaw(2));
      assertEquals(ByteSourceFactory.wrap(new byte[] {(byte) (pdx.getMyLong() >> 56),
          (byte) (pdx.getMyLong() >> 48), (byte) (pdx.getMyLong() >> 40),
          (byte) (pdx.getMyLong() >> 32), (byte) (pdx.getMyLong() >> 24),
          (byte) (pdx.getMyLong() >> 16), (byte) (pdx.getMyLong() >> 8), (byte) pdx.getMyLong(),}),
          reader.getRaw(3));
      assertEquals(ByteSourceFactory.wrap(str2Bytes), reader.getRaw(4));
      assertEquals(ByteSourceFactory.wrap(str3Bytes), reader.getRaw(5));
      assertEquals(
          ByteSourceFactory.wrap(new byte[] {(byte) (pdx.getMyInt() >> 24),
              (byte) (pdx.getMyInt() >> 16), (byte) (pdx.getMyInt() >> 8), (byte) pdx.getMyInt()}),
          reader.getRaw(6));
      assertEquals(ByteSourceFactory.wrap(new byte[] {(byte) (floatBytes >> 24),
          (byte) (floatBytes >> 16), (byte) (floatBytes >> 8), (byte) floatBytes}),
          reader.getRaw(7));
    } finally {
      cache.setReadSerializedForTest(false);
    }
  }


  @Test
  public void testByteFormatForLongStrings() throws Exception {
    boolean myFlag = true;
    short myShort = 25;
    String myString1 =
        "A very long string1.A very long string1.A very long string1.A very long string1.A very long string1.A very long string1.A very long string1."
            + "A very long string1.A very long string1.A very long string1.A very long string1.A very long string1.A very long string1.A very long string1.A very long string1."
            + "A very long string1.A very long string1.A very long string1.A very long string1.A very long string1.A very long string1.A very long string1.A very long string1."
            + "A very long string1.A very long string1.A very long string1.A very long string1.A very long string1.A very long string1.A very long string1.A very long string1."
            + "A very long string1.A very long string1.A very long string1.A very long string1.A very long string1.A very long string1.A very long string1.A very long string1."
            + "A very long string1.A very long string1.A very long string1.A very long string1.A very long string1.A very long string1.A very long string1.A very long string1."
            + "A very long string1.A very long string1.A very long string1.A very long string1.A very long string1.A very long string1.A very long string1.A very long string1."
            + "A very long string1.A very long string1.A very long string1.A very long string1.A very long string1.A very long string1.A very long string1.A very long string1."
            + "A very long string1.A very long string1.A very long string1.A very long string1.A very long string1.A very long string1.A very long string1.A very long string1."
            + "A very long string1.A very long string1.A very long string1.A very long string1.A very long string1.A very long string1.A very long string1.";
    long myLong = 15654;
    String myString2 = "Class4_myString2";
    String myString3 =
        "Even longer string3. Even longer string3. Even longer string3. Even longer string3. Even longer string3. Even longer string3. "
            + "Even longer string3. Even longer string3. Even longer string3. Even longer string3. Even longer string3. Even longer string3. Even longer string3. "
            + "Even longer string3. Even longer string3. Even longer string3. Even longer string3. Even longer string3. Even longer string3. Even longer string3. "
            + "Even longer string3. Even longer string3. Even longer string3. Even longer string3. Even longer string3. Even longer string3. Even longer string3. "
            + "Even longer string3. Even longer string3. Even longer string3. Even longer string3. Even longer string3. Even longer string3. Even longer string3. "
            + "Even longer string3. Even longer string3. Even longer string3. Even longer string3. Even longer string3. ";
    int myInt = 1420;
    float myFloat = 123.023f;

    HeapDataOutputStream out = new HeapDataOutputStream(KnownVersion.CURRENT);
    SimpleClass1 pdx =
        new SimpleClass1(myFlag, myShort, myString1, myLong, myString2, myString3, myInt, myFloat);
    DataSerializer.writeObject(pdx, out);
    int typeId = getPdxTypeIdForClass(SimpleClass1.class);

    HeapDataOutputStream hdos1 = new HeapDataOutputStream(KnownVersion.CURRENT);
    DataSerializer.writeString(myString1, hdos1);
    byte[] str1Bytes = hdos1.toByteArray();
    HeapDataOutputStream hdos2 = new HeapDataOutputStream(KnownVersion.CURRENT);
    DataSerializer.writeString(myString2, hdos2);
    byte[] str2Bytes = hdos2.toByteArray();
    HeapDataOutputStream hdos3 = new HeapDataOutputStream(KnownVersion.CURRENT);
    DataSerializer.writeString(myString3, hdos3);
    byte[] str3Bytes = hdos3.toByteArray();

    int length = 1 /* myFlag */ + 2 /* myShort */
        + 8 /* myLong */ + 4 /* myInt */ + 4 /* myFloat */ + str1Bytes.length + str2Bytes.length
        + str3Bytes.length + (2 * 2) /* short offset for 3 strings */;

    int offset1 = 1 + 2;
    int offset2 = offset1 + 8 + str1Bytes.length;
    int offset3 = offset1 + 8 + str1Bytes.length + str2Bytes.length;
    byte[] actual = out.toByteArray();
    int floatBytes = Float.floatToRawIntBits(myFloat);
    Byte[] expected = new Byte[] {DSCODE.PDX.toByte(), // byte
        (byte) (length >> 24), (byte) (length >> 16), (byte) (length >> 8), (byte) length, // int -
                                                                                           // length
                                                                                           // of
                                                                                           // byte
                                                                                           // stream
        (byte) (typeId >> 24), (byte) (typeId >> 16), (byte) (typeId >> 8), (byte) typeId, // int -
                                                                                           // typeId
        1, // boolean - myFlag = true
        (byte) (myShort >> 8), (byte) myShort, // short - myShort
        (byte) (myLong >> 56), (byte) (myLong >> 48), (byte) (myLong >> 40), (byte) (myLong >> 32),
        (byte) (myLong >> 24), (byte) (myLong >> 16), (byte) (myLong >> 8), (byte) myLong, // long -
                                                                                           // myLong
        (byte) (myInt >> 24), (byte) (myInt >> 16), (byte) (myInt >> 8), (byte) myInt, // int -
                                                                                       // myInt
        (byte) (floatBytes >> 24), (byte) (floatBytes >> 16), (byte) (floatBytes >> 8),
        (byte) floatBytes, // float - myFloat
        (byte) (offset3 >> 8), (byte) offset3, // offset of myString3
        (byte) (offset2 >> 8), (byte) offset2, // offset of myString2
    };

    for (int i = (str1Bytes.length - 1); i >= 0; i--) {
      expected =
          (Byte[]) ArrayUtils.insert(expected, offset1 + PdxWriterImpl.HEADER_SIZE, str1Bytes[i]); // +
                                                                                                   // 5
