Accomodate updated openmessaging api

shutian.lzh · shutian.lzh · commit c508cb33f644 · 2018-05-08T13:22:12.000+08:00
diff --git a/openmessaging/src/main/java/io/openmessaging/rocketmq/domain/BytesMessageImpl.java b/openmessaging/src/main/java/io/openmessaging/rocketmq/domain/BytesMessageImpl.java
@@ -20,6 +20,7 @@
 import io.openmessaging.KeyValue;
 import io.openmessaging.Message;
 import io.openmessaging.OMS;
+import io.openmessaging.exception.OMSMessageFormatException;
 import org.apache.commons.lang3.builder.ToStringBuilder;
 
 public class BytesMessageImpl implements BytesMessage {
@@ -33,8 +34,12 @@ public BytesMessageImpl() {
     }
 
     @Override
-    public byte[] getBody() {
-        return body;
+    public <T> T getBody(Class<T> type) throws OMSMessageFormatException {
+        if (type == byte[].class) {
+            return (T)body;
+        }
+
+        throw new OMSMessageFormatException("", "Cannot assign byte[] to " + type.getName());
     }
 
     @Override
diff --git a/openmessaging/src/main/java/io/openmessaging/rocketmq/utils/OMSUtil.java b/openmessaging/src/main/java/io/openmessaging/rocketmq/utils/OMSUtil.java
@@ -46,7 +46,7 @@ public static String buildInstanceName() {
 
     public static org.apache.rocketmq.common.message.Message msgConvert(BytesMessage omsMessage) {
         org.apache.rocketmq.common.message.Message rmqMessage = new org.apache.rocketmq.common.message.Message();
-        rmqMessage.setBody(omsMessage.getBody());
+        rmqMessage.setBody(omsMessage.getBody(byte[].class));
 
         KeyValue sysHeaders = omsMessage.sysHeaders();
         KeyValue userHeaders = omsMessage.userHeaders();
diff --git a/openmessaging/src/test/java/io/openmessaging/rocketmq/consumer/PullConsumerImplTest.java b/openmessaging/src/test/java/io/openmessaging/rocketmq/consumer/PullConsumerImplTest.java
@@ -83,7 +83,7 @@ public void testPoll() {
 
         Message message = consumer.receive();
         assertThat(message.sysHeaders().getString(Message.BuiltinKeys.MESSAGE_ID)).isEqualTo("NewMsgId");
-        assertThat(((BytesMessage) message).getBody()).isEqualTo(testBody);
+        assertThat(((BytesMessage) message).getBody(byte[].class)).isEqualTo(testBody);
     }
 
     @Test
diff --git a/openmessaging/src/test/java/io/openmessaging/rocketmq/consumer/PushConsumerImplTest.java b/openmessaging/src/test/java/io/openmessaging/rocketmq/consumer/PushConsumerImplTest.java
@@ -75,7 +75,7 @@ public void testConsumeMessage() {
             @Override
             public void onReceived(Message message, Context context) {
                 assertThat(message.sysHeaders().getString(Message.BuiltinKeys.MESSAGE_ID)).isEqualTo("NewMsgId");
-                assertThat(((BytesMessage) message).getBody()).isEqualTo(testBody);
+                assertThat(((BytesMessage) message).getBody(byte[].class)).isEqualTo(testBody);
                 context.ack();
             }
         });
diff --git a/pom.xml b/pom.xml
@@ -592,7 +592,7 @@
             <dependency>
                 <groupId>io.openmessaging</groupId>
                 <artifactId>openmessaging-api</artifactId>
-                <version>0.3.0-alpha</version>
+                <version>0.3.1-alpha-SNAPSHOT</version>
             </dependency>
             <dependency>
                 <groupId>log4j</groupId>

Original file line number	Diff line number	Diff line change
`@@ -83,7 +83,7 @@ public void testPoll() {`
`83`	`83`
`84`	`84`	`Message message = consumer.receive();`
`85`	`85`	`assertThat(message.sysHeaders().getString(Message.BuiltinKeys.MESSAGE_ID)).isEqualTo("NewMsgId");`
`86`		`- assertThat(((BytesMessage) message).getBody()).isEqualTo(testBody);`
	`86`	`+ assertThat(((BytesMessage) message).getBody(byte[].class)).isEqualTo(testBody);`
`87`	`87`	`}`
`88`	`88`
`89`	`89`	`@Test`
Original file line number	Diff line number	Diff line change
`@@ -75,7 +75,7 @@ public void testConsumeMessage() {`
`75`	`75`	`@Override`
`76`	`76`	`public void onReceived(Message message, Context context) {`
`77`	`77`	`assertThat(message.sysHeaders().getString(Message.BuiltinKeys.MESSAGE_ID)).isEqualTo("NewMsgId");`
`78`		`- assertThat(((BytesMessage) message).getBody()).isEqualTo(testBody);`
	`78`	`+ assertThat(((BytesMessage) message).getBody(byte[].class)).isEqualTo(testBody);`
`79`	`79`	`context.ack();`
`80`	`80`	`}`
`81`	`81`	`});`