package com.twitter.elephantbird.mapreduce.io;

import java.util.List;

import com.google.protobuf.ByteString;
import com.google.protobuf.DescriptorProtos;
import com.google.protobuf.DescriptorProtos.FieldDescriptorProto.Type;
import com.google.protobuf.DescriptorProtos.FieldDescriptorProto.Label;
import com.google.protobuf.Descriptors;
import com.google.protobuf.DynamicMessage;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.Message;

import com.twitter.elephantbird.util.Protobufs;

/**
 * This is a {@link DynamicMessage} equivalent of following protobuf : <pre>
 *
 * message SerializedBlock {
 *  optional int32 version              = 1;
 *  optional string proto_class_name    = 2;
 *  repeated bytes proto_blobs          = 3;
 * };       </pre>
 *
 * The primary purpose is to avoid any dependence on pre-generated protobuf
 * classes. This makes EB libraries compatible with both protobuf 2.4 and 2.5.
 * <p>
 *
 * Developer Note: More documentation on protobuf fields and file format
 * is included at the bottom of this file.
 */
public class SerializedBlock {

  private final Message message;

  // use newInstance() to create a new message
  private SerializedBlock(Message message) {
    this.message = message;
  }

  public Message getMessage() {
    return message;
  }

  public int getVersion() {
    return (Integer) message.getField(versionDesc);
  }

  public String getProtoClassName() {
    return (String) message.getField(protoClassNameDesc);
  }

  public List<ByteString> getProtoBlobs() {
    return (List<ByteString>) message.getField(protoBlobsDesc);
  }

  public static SerializedBlock newInstance(String protoClassName, List<ByteString> protoBlobs) {
    return new SerializedBlock(
        DynamicMessage.newBuilder(messageDescriptor)
            .setField(versionDesc, Integer.valueOf(1))
            .setField(protoClassNameDesc, protoClassName)
            .setField(protoBlobsDesc, protoBlobs)
            .build());
  }

  public static SerializedBlock parseFrom(byte[] messageBuffer)
                                          throws InvalidProtocolBufferException {
    return new SerializedBlock(
        DynamicMessage.newBuilder(messageDescriptor)
            .mergeFrom(messageBuffer)
            .build());
  }

  private static final Descriptors.Descriptor messageDescriptor;
  private static final Descriptors.FieldDescriptor versionDesc;
  private static final Descriptors.FieldDescriptor protoClassNameDesc;
  private static final Descriptors.FieldDescriptor protoBlobsDesc;

  static {
    // initialize messageDescriptor and the three field descriptors

    DescriptorProtos.FieldDescriptorProto version =
        DescriptorProtos.FieldDescriptorProto.newBuilder()
        .setName("version")
        .setNumber(1)
        .setType(Type.TYPE_INT32)
        .setLabel(Label.LABEL_OPTIONAL)
        .build();

    DescriptorProtos.FieldDescriptorProto protoClassName =
        DescriptorProtos.FieldDescriptorProto.newBuilder()
        .setName("proto_class_name")
        .setNumber(2)
        .setType(Type.TYPE_STRING)
        .setLabel(Label.LABEL_OPTIONAL)
        .build();

    DescriptorProtos.FieldDescriptorProto protoBlobs =
        DescriptorProtos.FieldDescriptorProto.newBuilder()
        .setName("proto_blobs")
        .setNumber(3)
        .setType(Type.TYPE_BYTES)
        .setLabel(Label.LABEL_REPEATED)
        .build();

    try {
      messageDescriptor = Protobufs.makeMessageDescriptor(
              DescriptorProtos.DescriptorProto.newBuilder()
                .setName("SerializedBlock")
                .addField(version)
                .addField(protoClassName)
                .addField(protoBlobs)
                .build());
    } catch (Descriptors.DescriptorValidationException e) {
      throw new RuntimeException(e);
    }

    versionDesc        = messageDescriptor.findFieldByName("version");
    protoClassNameDesc = messageDescriptor.findFieldByName("proto_class_name");
    protoBlobsDesc     = messageDescriptor.findFieldByName("proto_blobs");
  }
}

/*
  Contents of old protobuf spec file block_storage.proto
    which included detailed documentation :

  package com.twitter.data.proto;

  message SerializedBlock {
    // The version of the block format we are writing. Always set to 1.
    optional int32 version              = 1;
    // The class_name of the message, e.g. "com.twitter.data.proto.Tables.Status"
    optional string proto_class_name    = 2;
    // A list of serialized byte blobs of the contained protocol buffers.
    // Generally there should be no more than 1000 or so blobs per block,
    // because all of them get parsed into memory at once during analysis.
    // The number of blobs per block can vary arbitrarily, and can even be
    // just 1 if necessary (somewhat space-inefficient).
    repeated bytes proto_blobs          = 3;
  };

  // The serialization format for all protobuf data consists of repeated sequences that look like

  // 16-byte GUID | 4-byte size, little-endian | that many bytes of a serialized SerializedBlock data structure
  // 16-byte GUID | 4-byte size, little-endian | that many bytes of a serialized SerializedBlock data structure
  // ...
  // 16-byte GUID | 4-byte size, little-endian | that many bytes of a serialized SerializedBlock data structure

  // The 16-byte GUID is ALWAYS the following:
  // 0x29, 0xd8, 0xd5, 0x06, 0x58, 0xcd, 0x4c, 0x29, 0xb2, 0xbc, 0x57, 0x99, 0x21, 0x71, 0xbd, 0xff
  //

  // Pseudocode for serializing the block is the following (approximately in Java):
  // SerializedBlock block = SerializedBlock.newBuilder().setVersion(1)
  //                                                     .setProtoClassName(Status.class.getName())
  //                                                     .addProtoBlobs(status1.toByteString())
  //                                                     .addProtoBlobs(status2.toByteString())
  //                                                     .build();
  // Now write to disk, with os being an OutputStream:
  // os.write(<bytes of the guid above>)
  // os.write(<raw little endian of block.getSerializedSize())
  // os.write(block.toByteArray());
  //
  // and repeat.
  // Note that the description above implies that all files start with the GUID.
*/

@@ -799,24 +799,322 @@
 * Th
-e primary purpos
+is protobuf is required for BinaryBlock format used to store to binary records.%0A * See %7B@link BinaryBlockReader%7D and %7B@link BinaryBlockWriter%7D. The file layout%0A * is described in a comment at the bottom of this file. The format is an alternative%0A * to a SequenceFile. %3Cp%3E%0A *%0A * The primary purpose of DynamicMessag
 e is
@@ -1140,16 +1140,19 @@
 dence on
+%0A *
  pre-gen
@@ -1170,19 +1170,16 @@
 obuf
-%0A *
  classes
 . Th
@@ -1178,67 +1178,86 @@
 sses
-. This makes EB libraries compatible with both protobuf 
+, since generated files are not compatible between%0A * protobuf versions (
 2.4
+.1
  and
@@ -1261,20 +1261,80 @@
 and 2.5.
-%0A * 
+0).This class works with either of these%0A * versions at runtime.
 %3Cp%3E%0A *%0A 

Update javadoc for SerializedBlock