Always send a body section

qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/message/AmqpCodec.java

@@ -76,6 +76,12 @@ private static class EncoderDecoderContext {
         Map<Integer, ReadableBuffer> messageAnnotationsCache = new HashMap<>();
     }
 
+    /**
+     * The body section written for a message whose facade carries no body, see
+     * {@link #encodeMessage(AmqpJmsMessageFacade)}.
+     */
+    private static final AmqpValue NULL_BODY_SECTION = new AmqpValue(null);
+
     private static final ThreadLocal<EncoderDecoderContext> TLS_CODEC = new ThreadLocal<EncoderDecoderContext>() {
         @Override
         protected EncoderDecoderContext initialValue() {
@@ -203,6 +209,13 @@ public static ByteBuf encodeMessage(AmqpJmsMessageFacade message) {
         }
         if (body != null) {
             encoder.writeObject(body);
+        } else {
+            // AMQP requires every message to carry a body section (AMQP 1.0 section 3.2 lists the
+            // other sections as "zero or one" but the body as one of three mandatory choices), and
+            // the JMS mapping encodes a JMS Message that has no body as "a single amqp-value
+            // section containing null" (AMQP JMS Mapping section 3.2.4.7). The facade keeps a null
+            // body to represent "this message has no body", so supply that section here.
+            encoder.writeObject(NULL_BODY_SECTION);
         }
         if (footer != null) {
             encoder.writeObject(footer);

qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ProducerIntegrationTest.java

@@ -160,6 +160,40 @@ public void testCloseSenderTimesOut() throws Exception {
         }
     }
 
+    @Test
+    @Timeout(20)
+    public void testSentMessageWithNoBodyIsEncodedAsAmqpValueNull() throws Exception {
+        try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
+            Connection connection = testFixture.establishConnecton(testPeer);
+            testPeer.expectBegin();
+            testPeer.expectSenderAttach();
+
+            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+            Queue queue = session.createQueue("myQueue");
+            MessageProducer producer = session.createProducer(queue);
+
+            MessageHeaderSectionMatcher headersMatcher = new MessageHeaderSectionMatcher(true);
+            MessageAnnotationsSectionMatcher msgAnnotationsMatcher = new MessageAnnotationsSectionMatcher(true);
+            msgAnnotationsMatcher.withEntry(AmqpMessageSupport.JMS_MSG_TYPE, equalTo(AmqpMessageSupport.JMS_MESSAGE));
+            MessagePropertiesSectionMatcher propsMatcher = new MessagePropertiesSectionMatcher(true);
+            TransferPayloadCompositeMatcher messageMatcher = new TransferPayloadCompositeMatcher();
+            messageMatcher.setHeadersMatcher(headersMatcher);
+            messageMatcher.setMessageAnnotationsMatcher(msgAnnotationsMatcher);
+            messageMatcher.setPropertiesMatcher(propsMatcher);
+            // AMQP requires a body section on every message, and the JMS mapping encodes a JMS
+            // Message that has no body as a single amqp-value section containing null.
+            messageMatcher.setMessageContentMatcher(new EncodedAmqpValueMatcher(null));
+            testPeer.expectTransfer(messageMatcher);
+            testPeer.expectClose();
+
+            producer.send(session.createMessage());
+
+            connection.close();
+
+            testPeer.waitForAllHandlersToComplete(1000);
+        }
+    }
+
     @Test
     @Timeout(20)
     public void testSentTextMessageCanBeModified() throws Exception {

qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/amqp/message/AmqpJmsMessageFacadeTest.java

@@ -47,6 +47,7 @@
 import org.apache.qpid.jms.JmsQueue;
 import org.apache.qpid.jms.JmsTemporaryQueue;
 import org.apache.qpid.jms.JmsTopic;
+import org.apache.qpid.jms.message.JmsMessage;
 import org.apache.qpid.jms.message.facade.JmsMessageFacade;
 import org.apache.qpid.jms.provider.amqp.AmqpConsumer;
 import org.apache.qpid.jms.test.testpeer.describedtypes.sections.PropertiesDescribedType;
@@ -63,8 +64,11 @@
 import org.apache.qpid.proton.amqp.messaging.Header;
 import org.apache.qpid.proton.amqp.messaging.MessageAnnotations;
 import org.apache.qpid.proton.amqp.messaging.Properties;
+import org.apache.qpid.proton.amqp.messaging.Section;
 import org.apache.qpid.proton.codec.Data;
 import org.apache.qpid.proton.message.Message;
+
+import io.netty.buffer.ByteBuf;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
 
@@ -2280,6 +2284,38 @@ public void testMessageHasBodyDetectsPayload() throws Exception {
         assertTrue(amqpMessageFacade.hasBody());
     }
 
+    @Test
+    public void testEncodeMessageWithNoBodyWritesAmqpValueNullBodySection() throws Exception {
+        AmqpJmsMessageFacade amqpMessageFacade = createNewMessageFacade();
+        assertFalse(amqpMessageFacade.hasBody());
+
+        ByteBuf encoded = amqpMessageFacade.encodeMessage();
+        byte[] bytes = new byte[encoded.readableBytes()];
+        encoded.readBytes(bytes);
+
+        Message protonMessage = Message.Factory.create();
+        protonMessage.decode(bytes, 0, bytes.length);
+
+        // AMQP requires a body section on every message, and the JMS mapping encodes a JMS
+        // Message that has no body as a single amqp-value section containing null.
+        Section body = protonMessage.getBody();
+        assertTrue(body instanceof AmqpValue, "Expected an amqp-value body section but got: " + body);
+        assertNull(((AmqpValue) body).getValue());
+    }
+
+    @Test
+    public void testMessageWithNoBodyRoundTripsAsGenericMessage() throws Exception {
+        AmqpJmsMessageFacade amqpMessageFacade = createNewMessageFacade();
+
+        AmqpJmsMessageFacade decoded = AmqpCodec.decodeMessage(
+            createMockAmqpConsumer(), new AmqpReadableBuffer(amqpMessageFacade.encodeMessage()));
+
+        // The x-opt-jms-msg-type annotation has to keep this a generic Message: an amqp-value
+        // section containing null would otherwise be read back as a TextMessage.
+        assertEquals(JmsMessageFacade.JMS_MESSAGE, decoded.getJmsMsgType());
+        assertEquals(JmsMessage.class, decoded.asJmsMessage().getClass());
+    }
+
     @Test
     public void testClearBodyRemoveMessageBody() {
         Message message = Message.Factory.create();