/*
 *  Copyright (c) 2012-2013 Malhar, Inc.
 *  All Rights Reserved.
 */
package com.malhartech.bufferserver.packet;

import com.malhartech.bufferserver.util.Codec;

/**
 *
 * @author Chetan Narsude <chetan@malhar-inc.com>
 */
public class PayloadTuple extends Tuple
{
  public PayloadTuple(byte[] array, int offset, int length)
  {
    super(array, offset, length);
  }

  @Override
  public MessageType getType()
  {
    return MessageType.PAYLOAD;
  }

  @Override
  public int getPartition()
  {
    return readVarInt(offset + 1, offset + length);
  }

  @Override
  public int getWindowId()
  {
    throw new UnsupportedOperationException("Not supported yet."); //To change body of generated methods, choose Tools | Templates.
  }

  @Override
  public int getDataOffset()
  {
    int dataOffset = this.offset + 1;
    while (buffer[dataOffset++] < 0) {
    }
    return dataOffset;
  }

  @Override
  public int getBaseSeconds()
  {
    throw new UnsupportedOperationException("Not supported yet."); //To change body of generated methods, choose Tools | Templates.
  }

  @Override
  public int getWindowWidth()
  {
    throw new UnsupportedOperationException("Not supported yet."); //To change body of generated methods, choose Tools | Templates.
  }

  public static byte[] getSerializedTuple(int partition, int size)
  {
    int bits = 32 - Integer.numberOfLeadingZeros(partition);
    do {
      size++;
    }
    while ((bits -= 7) > 0);

    byte[] array = new byte[size];
    Codec.writeRawVarint32(size, array, 0);
    return array;
  }

}

@@ -1479,11 +1479,57 @@
 size
+ + 1
 %5D;%0A
+    array%5B0%5D = MessageType.PAYLOAD_VALUE;%0A
     
@@ -1551,20 +1551,25 @@
 arint32(
-size
+partition
 , array,
@@ -1569,17 +1569,17 @@
  array, 
-0
+1
 );%0A    r

fix the serialization of the payload tuples (the type of the tuple was missing).