clean up patch and test from Sandeep Tata fixing #21

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@760006 13f79535-47bb-0310-9956-ffa450edef68

src/org/apache/cassandra/db/ReadMessage.java

@@ -25,6 +25,9 @@ import java.io.IOException;
 import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.List;
+
+import org.apache.commons.lang.StringUtils;
+
 import org.apache.cassandra.continuations.Suspendable;
 import org.apache.cassandra.io.ICompactSerializer;
 import org.apache.cassandra.net.Message;
@@ -154,6 +157,20 @@ public class ReadMessage implements Serializable
     {
     	return columns_;
     }
+
+    public String toString()
+    {
+        return "ReadMessage(" +
+               "table='" + table_ + '\'' +
+               ", key='" + key_ + '\'' +
+               ", columnFamily_column='" + columnFamily_column_ + '\'' +
+               ", start=" + start_ +
+               ", count=" + count_ +
+               ", sinceTimestamp=" + sinceTimestamp_ +
+               ", columns=[" + StringUtils.join(columns_, ", ") + "]" +
+               ", isDigestQuery=" + isDigestQuery_ +
+               ')';
+    }
 }
 
 class ReadMessageSerializer implements ICompactSerializer<ReadMessage>
@@ -202,7 +219,7 @@ class ReadMessageSerializer implements ICompactSerializer<ReadMessage>
 		{
 			rm = new ReadMessage(table, key, columnFamily_column, columns);
 		}
-		if( sinceTimestamp > 0 )
+		else if( sinceTimestamp > 0 )
 		{
 			rm = new ReadMessage(table, key, columnFamily_column, sinceTimestamp);
 		}

test/org/apache/cassandra/db/ReadMessageTest.java

@@ -0,0 +1,44 @@
+package org.apache.cassandra.db;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+import org.apache.cassandra.io.DataInputBuffer;
+import org.apache.cassandra.io.DataOutputBuffer;
+import org.testng.annotations.Test;
+
+public class ReadMessageTest
+{
+    @Test
+    public void testMakeReadMessage()
+    {
+        ArrayList<String> colList = new ArrayList<String>();
+        colList.add("col1");
+        colList.add("col2");
+
+        ReadMessage rm = new ReadMessage("Table1", "row1", "foo", colList);
+        ReadMessage rm2 = serializeAndDeserializeReadMessage(rm);
+
+        assert rm2.toString().equals(rm.toString());
+    }
+
+    private ReadMessage serializeAndDeserializeReadMessage(ReadMessage rm)
+    {
+        ReadMessage rm2 = null;
+        ReadMessageSerializer rms = (ReadMessageSerializer) ReadMessage.serializer();
+        DataOutputBuffer dos = new DataOutputBuffer();
+        DataInputBuffer dis = new DataInputBuffer();
+
+        try
+        {
+            rms.serialize(rm, dos);
+            dis.reset(dos.getData(), dos.getLength());
+            rm2 = rms.deserialize(dis);
+        }
+        catch (IOException e)
+        {
+            throw new RuntimeException(e);
+        }
+        return rm2;
+    }
+}