Handle AMQP Value in message body

Fixes #70

Author: acogoluegnes

src/main/java/com/rabbitmq/stream/codec/QpidProtonCodec.java

@@ -24,6 +24,7 @@
 import java.util.UUID;
 import java.util.function.Function;
 import org.apache.qpid.proton.amqp.*;
+import org.apache.qpid.proton.amqp.messaging.AmqpValue;
 import org.apache.qpid.proton.amqp.messaging.ApplicationProperties;
 import org.apache.qpid.proton.amqp.messaging.Data;
 import org.apache.qpid.proton.amqp.messaging.MessageAnnotations;
@@ -464,7 +465,20 @@ public long getPublishingId() {
 
     @Override
     public byte[] getBodyAsBinary() {
-      return message.getBody() == null ? null : ((Data) message.getBody()).getValue().getArray();
+      if (message.getBody() == null) {
+        return null;
+      } else if (message.getBody() instanceof Data) {
+        return ((Data) message.getBody()).getValue().getArray();
+      } else if (message.getBody() instanceof AmqpValue) {
+        AmqpValue value = (AmqpValue) message.getBody();
+        if (value.getValue() instanceof byte[]) {
+          return (byte[]) value.getValue();
+        }
+      }
+      throw new IllegalStateException(
+          "Body cannot by returned as array of bytes: "
+              + message.getBody()
+              + ". Use #getBody() to get native representation.");
     }
 
     @Override

src/main/java/com/rabbitmq/stream/codec/SwiftMqCodec.java

@@ -395,12 +395,45 @@ public long getPublishingId() {
 
     @Override
     public byte[] getBodyAsBinary() {
-      return amqpMessage.getData().get(0).getValue();
+      if (amqpMessage.getData() != null) {
+        return amqpMessage.getData().get(0).getValue();
+      } else if (amqpMessage.getAmqpValue() != null) {
+        AMQPType value = amqpMessage.getAmqpValue().getValue();
+        if (value instanceof AMQPBinary) {
+          return ((AMQPBinary) value).getValue();
+        } else if (value instanceof AMQPArray) {
+          try {
+            AMQPType[] array = ((AMQPArray) value).getValue();
+            if (array.length > 0) {
+              // far-fetch
+              if (array[0] instanceof AMQPByte) {
+                byte[] result = new byte[array.length];
+                for (int i = 0; i < array.length; i++) {
+                  result[i] = ((AMQPByte) array[i]).getValue();
+                }
+                return result;
+              }
+            }
+          } catch (IOException e) {
+            throw new RuntimeException(e);
+          }
+        }
+      } else if (amqpMessage.getData() == null && amqpMessage.getAmqpValue() == null) {
+        return null;
+      }
+      throw new IllegalStateException(
+          "Body cannot by returned as array of bytes. Use #getBody() to get native representation.");
     }
 
     @Override
     public Object getBody() {
-      return body;
+      if (amqpMessage.getData() != null) {
+        return amqpMessage.getData();
+      } else if (amqpMessage.getAmqpValue() != null) {
+        return amqpMessage.getAmqpValue();
+      } else {
+        return null;
+      }
     }
 
     @Override

src/test/java/com/rabbitmq/stream/codec/CodecsTest.java

@@ -19,21 +19,28 @@
 import static org.mockito.Mockito.when;
 
 import com.rabbitmq.stream.Codec;
+import com.rabbitmq.stream.Codec.EncodedMessage;
 import com.rabbitmq.stream.Message;
 import com.rabbitmq.stream.MessageBuilder;
 import com.rabbitmq.stream.amqp.UnsignedByte;
 import com.rabbitmq.stream.amqp.UnsignedInteger;
 import com.rabbitmq.stream.amqp.UnsignedLong;
 import com.rabbitmq.stream.amqp.UnsignedShort;
+import com.rabbitmq.stream.codec.QpidProtonCodec.QpidProtonAmqpMessageWrapper;
 import java.math.BigDecimal;
 import java.math.BigInteger;
 import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Date;
+import java.util.List;
+import java.util.UUID;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.stream.Stream;
+import org.apache.qpid.proton.amqp.messaging.AmqpValue;
 import org.assertj.core.api.InstanceOfAssertFactories;
 import org.assertj.core.api.ThrowableAssert;
 import org.junit.jupiter.params.ParameterizedTest;
@@ -71,6 +78,18 @@ static Iterable<Codec> readCreatedMessage() {
         new SwiftMqCodec());
   }
 
+  static Stream<Codec> codecs() {
+    return Stream.of(new QpidProtonCodec(), new SwiftMqCodec());
+  }
+
+  static Stream<MessageBuilder> messageBuilders() {
+    return Stream.of(
+        new QpidProtonMessageBuilder(),
+        new SwiftMqMessageBuilder(),
+        new WrapperMessageBuilder(),
+        new SimpleCodec().messageBuilder());
+  }
+
   @ParameterizedTest
   @MethodSource("codecsCouples")
   void codecs(CodecCouple codecCouple) {
@@ -493,12 +512,28 @@ void notSupportedTypes(Supplier<MessageBuilder> messageBuilderSupplier) {
             action -> assertThatThrownBy(action).isInstanceOf(UnsupportedOperationException.class));
   }
 
-  static Stream<MessageBuilder> messageBuilders() {
-    return Stream.of(
-        new QpidProtonMessageBuilder(),
-        new SwiftMqMessageBuilder(),
-        new WrapperMessageBuilder(),
-        new SimpleCodec().messageBuilder());
+  @ParameterizedTest
+  @MethodSource("codecs")
+  void supportAmqpValueBody(Codec codec) {
+    Function<Object, Message> encodeDecode =
+        content -> {
+          org.apache.qpid.proton.message.Message nativeMessage =
+              org.apache.qpid.proton.message.Message.Factory.create();
+          nativeMessage.setBody(new AmqpValue(content));
+          QpidProtonAmqpMessageWrapper wrapper =
+              new QpidProtonAmqpMessageWrapper(true, 1L, nativeMessage);
+          EncodedMessage encoded = new QpidProtonCodec().encode(wrapper);
+          byte[] encodedData = new byte[encoded.getSize()];
+          System.arraycopy(encoded.getData(), 0, encodedData, 0, encoded.getSize());
+          Message decodedMessage = codec.decode(encodedData);
+          return decodedMessage;
+        };
+
+    Message m1 = encodeDecode.apply("hello".getBytes(StandardCharsets.UTF_8));
+    assertThat(m1.getBodyAsBinary()).asString(StandardCharsets.UTF_8).isEqualTo("hello");
+
+    Message m2 = encodeDecode.apply("a string is not an array of byte");
+    assertThatThrownBy(() -> m2.getBodyAsBinary()).isInstanceOf(IllegalStateException.class);
   }
 
   @ParameterizedTest

src/test/java/com/rabbitmq/stream/perf/StreamPerfTestTest.java

@@ -318,12 +318,7 @@ void compressionWithoutSubEntriesShouldNotStart() throws Exception {
   void monitoringShouldReturnValidEndpoint() throws Exception {
     int monitoringPort = randomNetworkPort();
     Future<?> run =
-        run(
-            builder()
-                .deleteStreams()
-                .monitoring()
-                .monitoringPort(monitoringPort)
-                .prometheus());
+        run(builder().deleteStreams().monitoring().monitoringPort(monitoringPort).prometheus());
     waitUntilStreamExists(s);
     waitOneSecond();