diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/message/AmqpCodec.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/message/AmqpCodec.java index 0750335af..399dac2a3 100644 --- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/message/AmqpCodec.java +++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/message/AmqpCodec.java @@ -76,6 +76,12 @@ private static class EncoderDecoderContext { Map 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 TLS_CODEC = new ThreadLocal() { @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); diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ProducerIntegrationTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ProducerIntegrationTest.java index 0680941c7..2cb528025 100644 --- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/ProducerIntegrationTest.java +++ b/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 { diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/amqp/message/AmqpJmsMessageFacadeTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/amqp/message/AmqpJmsMessageFacadeTest.java index 5dc9312ac..1bdef06bb 100644 --- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/provider/amqp/message/AmqpJmsMessageFacadeTest.java +++ b/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();