Added an option to support deferred decoding of messages.

messages now implement equals() and hashCode()



git-svn-id: https://svn.apache.org/repos/asf/activemq/sandbox/activemq-protobuf@703019 13f79535-47bb-0310-9956-ffa450edef68

activemq-protobuf-test/src/main/proto/deferred_decode.proto

@@ -0,0 +1,38 @@
+//
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+package org.apache.activemq.protobuf;
+option java_outer_classname = "DeferredUnmarshal";
+option deferred_decode = true;
+
+message Foo {
+
+  optional int32 field1 =  1;
+  optional int64 field2 =  2;
+
+}
+
+
+message Bar {
+
+  optional int32 field1 =  1;
+  optional int64 field2 =  2;
+
+  optional Foo field3 =  3;
+
+}
+

activemq-protobuf-test/src/test/java/org/apache/activemq/protobuf/DeferredUnmarshalTest.java

@@ -0,0 +1,75 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.protobuf;
+
+import org.apache.activemq.protobuf.DeferredUnmarshal.Foo;
+import org.apache.activemq.protobuf.DeferredUnmarshal.Bar;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+
+import junit.framework.TestCase;
+
+public class DeferredUnmarshalTest extends TestCase {
+
+	public void testDeferredDecoding() throws InvalidProtocolBufferException {
+
+		Foo foo = new Foo();
+		foo.setField1(5);
+		foo.setField2(20);
+
+		Bar bar = new Bar();
+
+		// There is no decoding pending so its' considered decoded.
+		assertTrue(bar.isDecoded());
+
+		bar.setField1(25);
+		bar.setField2(220);
+		bar.setField3(foo);
+
+		// The message should not be encoded yet.
+		assertFalse(bar.isEncoded());
+
+		// The message should be encoded now..
+		byte[] encodedForm = bar.toUnframedByteArray();
+		assertTrue(bar.isEncoded());
+
+		// Repeated encoding operations should just give back the same byte[]
+		assertTrue(encodedForm == bar.toUnframedByteArray());
+
+		// Decoding does not occur until a field is accessed.  The new message should still be considered encoded.
+		Bar bar2 = Bar.parseUnframed(encodedForm);
+		assertTrue(bar2.isEncoded());
+		assertFalse(bar2.isDecoded());
+
+		// This should now decode the message.
+		assertEquals(25, bar2.getField1());
+		assertTrue(bar2.isDecoded());
+
+		// Since bar2 still has not been modified it should still spit out the same byte[]
+		assertTrue(encodedForm == bar2.toUnframedByteArray());
+
+		// Nested messages should remain un-decoded.
+		assertFalse( bar2.getField3().isDecoded() );
+
+		// Changing a field should remove the encoding.
+		bar2.setField1(35);
+		assertFalse(bar2.isEncoded());
+		assertTrue(bar2.isDecoded());
+
+	}
+
+}

activemq-protobuf-test/src/test/java/org/apache/activemq/protobuf/EqualsTest.java

@@ -0,0 +1,58 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.protobuf;
+
+import org.apache.activemq.protobuf.DeferredUnmarshal.Bar;
+import org.apache.activemq.protobuf.DeferredUnmarshal.Foo;
+
+import junit.framework.TestCase;
+
+public class EqualsTest extends TestCase {
+
+	public void testDeferredUnmarshal() {
+
+		Bar bar1 = createBar();
+		Bar bar2 = createBar();
+
+		// They should have the same hash and equal the same value.
+		assertTrue(bar1.hashCode()==bar2.hashCode());
+		assertTrue(bar1.equals(bar2));
+
+		// Change bar2 a little.
+
+		bar2.setField2(35);
+
+		assertFalse(bar1.hashCode()==bar2.hashCode());
+		assertFalse(bar1.equals(bar2));
+
+
+	}
+
+	private Bar createBar() {
+		Bar bar;
+		Foo foo = new Foo();
+		foo.setField1(5);
+		foo.setField2(20);
+
+		bar = new Bar();
+		bar.setField1(25);
+		bar.setField2(220);
+		bar.setField3(foo);
+		return bar;
+	}
+
+}

activemq-protobuf/src/main/java/org/apache/activemq/protobuf/BaseMessage.java

@@ -39,6 +39,10 @@
 
     abstract public T clone() throws CloneNotSupportedException;
 
+	public void clear() {
+		memoizedSerializedSize = -1;
+	}
+
     ///////////////////////////////////////////////////////////////////
     // Write related helpers.
     ///////////////////////////////////////////////////////////////////
@@ -225,6 +229,7 @@ protected List<String> prefix(List<String> missingFields, String prefix) {
         return rc;
     }
 
+    abstract protected T checktInitialized() throws InvalidProtocolBufferException;
 
     /**
      * Read a raw Varint from the stream.  If larger than 32 bits, discard the
@@ -274,4 +279,8 @@ static protected byte readRawByte(InputStream is) throws IOException {
     	}
     	return (byte) rc;
     }
+
+	protected void loadAndClear() {
+		memoizedSerializedSize=-1;
+	}
 }

activemq-protobuf/src/main/java/org/apache/activemq/protobuf/DeferredDecodeMessage.java

@@ -0,0 +1,91 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.protobuf;
+
+import java.io.IOException;
+
+import com.google.protobuf.CodedInputStream;
+import com.google.protobuf.InvalidProtocolBufferException;
+
+abstract public class DeferredDecodeMessage<T> extends BaseMessage<T>{
+
+	protected byte[] encodedForm;
+	protected boolean decoded=true;
+
+	@Override
+    public T mergeFramed(CodedInputStream input) throws IOException {
+        int length = input.readRawVarint32();
+        int oldLimit = input.pushLimit(length);
+        T rc=  mergeUnframed(input.readRawBytes(length));
+        input.popLimit(oldLimit);
+        return rc;
+    }
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public T mergeUnframed(byte[] data) throws InvalidProtocolBufferException {
+		encodedForm = data;
+		decoded=false;
+		return (T)this;
+	}
+
+	@Override
+	public byte[] toUnframedByteArray() {
+		if( encodedForm==null ) {
+			encodedForm = super.toUnframedByteArray();
+		}
+		return encodedForm;
+	}
+
+	protected void load() {
+		if (!decoded) {
+			decoded = true;
+			try {
+				byte[] originalForm = encodedForm;
+				CodedInputStream input = CodedInputStream.newInstance(originalForm);
+				mergeUnframed(input);
+				input.checkLastTagWas(0);
+				// We need to reset the encoded form because the mergeUnframed from a stream clears it out.
+				encodedForm = originalForm;
+				checktInitialized();
+			} catch (Throwable e) {
+				throw new RuntimeException("Deferred message decoding failed: "+e.getMessage(), e);
+			}
+		}
+	}
+
+	protected void loadAndClear() {
+		super.loadAndClear();
+		load();
+		encodedForm = null;
+	}
+
+	public void clear() {
+		super.clear();
+		encodedForm = null;
+		decoded = true;
+	}
+
+	public boolean isDecoded() {
+		return decoded;
+	}
+
+	public boolean isEncoded() {
+		return encodedForm!=null;
+	}
+
+}