
package com.twitter.elephantbird.mapreduce.io;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.WritableComparable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * A Hadoop Writable wrapper around a serialized messages like Protocol buffers.
 */
public abstract class BinaryWritable<M> implements WritableComparable<BinaryWritable<M>> {
  private static final Logger LOG = LoggerFactory.getLogger(BinaryWritable.class);

  private M message;
  private BinaryConverter<M> converter;

  protected BinaryWritable(M message, BinaryConverter<M> converter) {
    this.message = message;
    this.converter = converter;
  }

  public M get() {
    return message;
  }

  public void clear() {
    message = null;
  }

  public void set(M message) {
    this.message = message;
  }

  public void write(DataOutput out) throws IOException {
    byte[] bytes = null;
    if (message != null) {
      bytes = converter.toBytes(message);
      if (bytes == null) {
        // should we throw an IOException instead?
        LOG.warn("Could not serialize " + message.getClass());
      }
    }
    if (bytes != null) {
      out.writeInt(bytes.length);
      out.write(bytes, 0, bytes.length);
    } else {
      out.writeInt(0);
    }
  }

  public void readFields(DataInput in) throws IOException {
    int size = in.readInt();
    if (size > 0) {
      byte[] messageBytes = new byte[size];
      in.readFully(messageBytes, 0, size);
      message = converter.fromBytes(messageBytes);
    }
  }

  @Override
  public int compareTo(BinaryWritable<M> other) {
    byte[] bytes = converter.toBytes(message);
    byte[] otherBytes = converter.toBytes(other.get());
    return BytesWritable.Comparator.compareBytes(bytes, 0, bytes.length, otherBytes, 0, otherBytes.length);
  }

  @Override
  public boolean equals(Object obj) {
    if (obj == null)
      return false;

    BinaryWritable<?> other;
    try {
      other = (BinaryWritable<?>)obj;
    } catch (ClassCastException e) {
      return false;
    }
    if (message != null)
      return message.equals(other.message);
    if (other.message == null) // contained objects in both writables are null.
      return converter.equals(other.converter);

    return false;
  }
}

@@ -569,16 +569,22 @@
 private 
+final 
 BinaryCo
@@ -605,16 +605,16 @@
 verter;%0A
-
 %0A  prote
@@ -890,32 +890,44 @@
 = message;%0A  %7D%0A%0A
+  @Override%0A
   public void wr
@@ -1359,24 +1359,36 @@
 %0A    %7D%0A  %7D%0A%0A
+  @Override%0A
   public voi
@@ -1977,24 +1977,26 @@
 obj == null)
+ %7B
 %0A      retur
@@ -2003,16 +2003,22 @@
 n false;
+%0A    %7D
 %0A%0A    Bi
@@ -2175,16 +2175,18 @@
 != null)
+ %7B
 %0A      r
@@ -2218,24 +2218,30 @@
 r.message);%0A
+    %7D%0A
     if (othe
@@ -2263,56 +2263,9 @@
 ll) 
-// contained objects in both writables are null.
+%7B
 %0A   
@@ -2308,16 +2308,22 @@
 verter);
+%0A    %7D
 %0A%0A    re
@@ -2334,14 +2334,120 @@
  false;%0A
-
   %7D%0A
+%0A  @Override%0A  public int hashCode() %7B%0A    return 31 + ((message == null) ? 0 : message.hashCode());%0A  %7D%0A%0A
 %7D%0A

add hashCode() to binaryWritables