Send empty message body instead of null message body

Null message body is forbidden in the AMQP 1.0 specification.
We set a data section with an empty binary when the message
body has not been set.

Stream messages are not checked on the server side, but
messages from other protocols will be checked for a non-null
body in RabbitMQ 4.0, and letting null body stream messages
through will crash when those messages are read with other protocols.

References rabbitmq/rabbitmq-server#6835
Fixes #544

Author: acogoluegnes

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

@@ -25,15 +25,14 @@
 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;
+import org.apache.qpid.proton.amqp.messaging.*;
 import org.apache.qpid.proton.codec.ReadableBuffer;
 import org.apache.qpid.proton.codec.WritableBuffer;
 
 public class QpidProtonCodec implements Codec {
 
+  static final Section EMPTY_BODY = new Data(new Binary(new byte[0]));
+
   private static final Function<String, String> MESSAGE_ANNOTATIONS_STRING_KEY_EXTRACTOR = k -> k;
   private static final Function<Symbol, String> MESSAGE_ANNOTATIONS_SYMBOL_KEY_EXTRACTOR =
       Symbol::toString;
@@ -233,13 +232,16 @@ public EncodedMessage encode(Message message) {
         qpidMessage.setMessageAnnotations(new MessageAnnotations(messageAnnotations));
       }
 
-      if (message.getBodyAsBinary() != null) {
+      if (message.getBodyAsBinary() == null) {
+        qpidMessage.setBody(EMPTY_BODY);
+      } else {
         qpidMessage.setBody(new Data(new Binary(message.getBodyAsBinary())));
       }
     }
     int bufferSize;
     if (qpidMessage.getBody() instanceof Data) {
       bufferSize = (int) (((Data) qpidMessage.getBody()).getValue().getLength() * 1.5);
+      bufferSize = bufferSize == 0 ? 128 : bufferSize;
     } else {
       bufferSize = 8192;
     }

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

@@ -14,6 +14,8 @@
 // info@rabbitmq.com.
 package com.rabbitmq.stream.codec;
 
+import static com.rabbitmq.stream.codec.QpidProtonCodec.EMPTY_BODY;
+
 import com.rabbitmq.stream.Message;
 import com.rabbitmq.stream.MessageBuilder;
 import java.math.BigDecimal;
@@ -57,6 +59,9 @@ public Message build() {
         message.setMessageAnnotations(
             new MessageAnnotations(messageAnnotationsBuilder.messageAnnotations));
       }
+      if (message.getBody() == null) {
+        message.setBody(EMPTY_BODY);
+      }
       return new QpidProtonCodec.QpidProtonAmqpMessageWrapper(
           hasPublishingId, publishingId, message);
     } else {

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

@@ -23,9 +23,12 @@
 
 public class SimpleCodec implements Codec {
 
+  static final byte[] EMPTY_BODY = new byte[0];
+
   @Override
   public EncodedMessage encode(Message message) {
-    return new EncodedMessage(message.getBodyAsBinary().length, message.getBodyAsBinary());
+    byte[] body = message.getBodyAsBinary() == null ? EMPTY_BODY : message.getBodyAsBinary();
+    return new EncodedMessage(body.length, body);
   }
 
   @Override

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

@@ -32,6 +32,8 @@
 
 public class SwiftMqCodec implements Codec {
 
+  static final Data EMPTY_BODY = new Data(new byte[0]);
+
   private static Object convertAmqpMapValue(AMQPType value) {
     if (value instanceof AMQPBoolean) {
       return ((AMQPBoolean) value).getValue() ? Boolean.TRUE : Boolean.FALSE;
@@ -260,7 +262,9 @@ public EncodedMessage encode(Message message) {
         }
       }
 
-      if (message.getBodyAsBinary() != null) {
+      if (message.getBodyAsBinary() == null) {
+        outboundMessage.addData(EMPTY_BODY);
+      } else {
         outboundMessage.addData(new Data(message.getBodyAsBinary()));
       }
     }

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

@@ -66,6 +66,9 @@ public Message build() {
           throw new StreamException("Error while setting application properties", e);
         }
       }
+      if (outboundMessage.getData() == null) {
+        outboundMessage.addData(SwiftMqCodec.EMPTY_BODY);
+      }
       return new SwiftMqCodec.SwiftMqAmqpMessageWrapper(
           hasPublishingId, publishingId, outboundMessage);
     } else {

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

@@ -42,7 +42,7 @@ public Message build() {
       return new SimpleMessage(
           this.hasPublishingId,
           this.publishingId,
-          body,
+          body == null ? SimpleCodec.EMPTY_BODY : body,
           this.messageAnnotationsBuilder == null
               ? null
               : this.messageAnnotationsBuilder.messageAnnotations,

src/test/java/com/rabbitmq/stream/impl/AmqpInteroperabilityTest.java

@@ -31,8 +31,6 @@
 import com.rabbitmq.stream.codec.SwiftMqCodec;
 import com.rabbitmq.stream.impl.Client.ClientParameters;
 import com.rabbitmq.stream.impl.Client.Response;
-import com.rabbitmq.stream.impl.TestUtils.BrokerVersion;
-import com.rabbitmq.stream.impl.TestUtils.BrokerVersionAtLeast;
 import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
 import java.util.*;