Always send a message with body

As per section 3.2 of the AMQP 1.0 specification,
a message must have a body, even if it is empty.

We use an empty array of bytes for a default, non-null body.

See https://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-messaging-v1.0-os.html#section-message-format

Author: acogoluegnes

src/main/java/com/rabbitmq/model/amqp/AmqpMessage.java

@@ -29,10 +29,12 @@
 
 class AmqpMessage implements Message {
 
+  private static final byte[] EMPTY_BODY = new byte[0];
+
   private final org.apache.qpid.protonj2.client.Message<byte[]> delegate;
 
   AmqpMessage() {
-    this(org.apache.qpid.protonj2.client.Message.create());
+    this(org.apache.qpid.protonj2.client.Message.create(EMPTY_BODY));
   }
 
   AmqpMessage(byte[] body) {

src/test/java/com/rabbitmq/model/amqp/ClientTest.java

@@ -292,26 +292,27 @@ void exchangeDeletionImpactOnSender(TestInfo info) throws Exception {
         Client client = client()) {
       connection.management().exchange().name(exchange).type(FANOUT).declare();
 
+      byte[] body = new byte[0];
       org.apache.qpid.protonj2.client.Connection protonConnection = connection(client);
       Session session = protonConnection.openSession();
       Sender sender =
           session.openSender(
               "/exchange/" + exchange, new SenderOptions().deliveryMode(AT_LEAST_ONCE));
-      Tracker tracker = sender.send(Message.create());
+      Tracker tracker = sender.send(Message.create(body));
       tracker.awaitSettlement(10, SECONDS);
       assertThat(tracker.remoteState()).isEqualTo(released());
 
       connection.management().binding().sourceExchange(exchange).destinationQueue(q).bind();
 
-      tracker = sender.send(Message.create());
+      tracker = sender.send(Message.create(body));
       tracker.awaitSettlement(10, SECONDS);
       assertThat(tracker.remoteState()).isEqualTo(DeliveryState.accepted());
 
       connection.management().exchangeDeletion().delete(exchange);
       try {
         int count = 0;
         while (count++ < 10) {
-          tracker = sender.send(Message.create());
+          tracker = sender.send(Message.create(body));
           tracker.awaitSettlement(10, SECONDS);
           assertThat(tracker.remoteState()).isEqualTo(released());
           Thread.sleep(100);