package com.twitter.elephantbird.pig.util;

import java.io.DataInput;
import java.io.DataInputStream;
import java.io.IOException;
import java.util.List;

import com.google.protobuf.Message;
import com.google.protobuf.Descriptors.Descriptor;
import com.google.protobuf.Descriptors.FieldDescriptor;
import com.google.protobuf.Message.Builder;

@SuppressWarnings("serial")
/**
 * This class wraps a protocol buffer message and attempts to delay parsing until individual
 * fields are requested.
 */
public class ProtobufTuple extends AbstractLazyTuple {

  private final Message msg_;
  private final Descriptor descriptor_;
  private final List<FieldDescriptor> fieldDescriptors_;
  private final ProtobufToPig protoConv_;
  private final int protoSize_;

  public ProtobufTuple(Message msg) {
    msg_ = msg;
    descriptor_ = msg.getDescriptorForType();
    fieldDescriptors_ = descriptor_.getFields();
    protoSize_ = fieldDescriptors_.size();
    protoConv_ = new ProtobufToPig();
    initRealTuple(protoSize_);
  }

  protected Object getObjectAt(int idx) {
    FieldDescriptor fieldDescriptor = fieldDescriptors_.get(idx);
    Object fieldValue = msg_.getField(fieldDescriptor);
    if (fieldDescriptor.getType() == FieldDescriptor.Type.MESSAGE) {
      return protoConv_.messageToTuple(fieldDescriptor, fieldValue);
    } else {
      return protoConv_.singleFieldToTuple(fieldDescriptor, fieldValue);
    }
  }

  @Override
  public long getMemorySize() {
    // The protobuf estimate is obviously inaccurate.
    return msg_.getSerializedSize() + realTuple.getMemorySize();
  }

  @Override
  public void readFields(DataInput inp) throws IOException {
    Builder builder = msg_.newBuilderForType();
    try {
      builder.mergeDelimitedFrom((DataInputStream) inp);
    } catch (ClassCastException e) {
      throw new IOException("Provided DataInput not instance of DataInputStream.", e);
    }
    Message msg = builder.build();
    reference(new ProtobufTuple(msg));
  }
}

@@ -41,94 +41,8 @@
 l;%0A%0A
-import java.io.DataInput;%0Aimport java.io.DataInputStream;%0Aimport java.io.IOException;%0A
 impo
@@ -207,52 +207,8 @@
 tor;
-%0Aimport com.google.protobuf.Message.Builder;
 %0A%0A@S
@@ -1452,404 +1452,7 @@
 %0A  %7D
-%0A%0A  @Override%0A  public void readFields(DataInput inp) throws IOException %7B%0A    Builder builder = msg_.newBuilderForType();%0A    try %7B%0A      builder.mergeDelimitedFrom((DataInputStream) inp);%0A    %7D catch (ClassCastException e) %7B%0A      throw new IOException(%22Provided DataInput not instance of DataInputStream.%22, e);%0A    %7D%0A    Message msg = builder.build();%0A    reference(new ProtobufTuple(msg));%0A  %7D
 %0A%7D%0A

remove ProtobufTuple's readFields. defaults to Tuple.readFields().