QPIDJMS-622 receiveBody must throw if type is Message or StreamMessage

tabish121 · tabish121 · commit e0c5367780e8 · 2026-03-12T15:50:53.000-04:00
When the next message is of type Message or StreamMessage the specification requires we throw an exception vs calling into getBody which for Message at least the specification has conflicting rules which simply returns null for all types requested. (cherry picked from commit de04f37)
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageConsumer.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsMessageConsumer.java
@@ -35,6 +35,7 @@
 import org.apache.qpid.jms.exceptions.JmsExceptionSupport;
 import org.apache.qpid.jms.message.JmsInboundMessageDispatch;
 import org.apache.qpid.jms.message.JmsMessage;
+import org.apache.qpid.jms.message.facade.JmsMessageFacade;
 import org.apache.qpid.jms.meta.JmsConsumerId;
 import org.apache.qpid.jms.meta.JmsConsumerInfo;
 import org.apache.qpid.jms.meta.JmsResource.ResourceState;
@@ -251,7 +252,16 @@ public <T> T receiveBody(Class<T> desired, long timeout) throws JMSException {
         try {
             envelope = dequeue(timeout, connection.isReceiveLocalOnly());
             if (envelope != null) {
-                messageBody = envelope.getMessage().getBody(desired);
+                final JmsMessage message = envelope.getMessage();
+                final int messageType = message.getFacade().getJmsMsgType();
+
+                if (messageType == JmsMessageFacade.JMS_MESSAGE) {
+                    throw new MessageFormatException("receiveBody cannot be called on a bare Message instance.");
+                } else if (messageType == JmsMessageFacade.JMS_STREAM_MESSAGE) {
+                    throw new MessageFormatException("receiveBody cannot be called on a StreamMessage instance.");
+                }
+
+                messageBody = message.getBody(desired);
             }
         } catch (MessageFormatException mfe) {
             // Should behave as if receiveBody never happened in these modes.
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/message/facade/JmsMessageFacade.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/message/facade/JmsMessageFacade.java
@@ -31,6 +31,29 @@
  */
 public interface JmsMessageFacade extends TraceableMessage {
 
+    public static final byte JMS_MESSAGE = 0;
+    public static final byte JMS_OBJECT_MESSAGE = 1;
+    public static final byte JMS_MAP_MESSAGE = 2;
+    public static final byte JMS_BYTES_MESSAGE = 3;
+    public static final byte JMS_STREAM_MESSAGE = 4;
+    public static final byte JMS_TEXT_MESSAGE = 5;
+
+    /**
+     * Returns a byte value that represents the message type from a know set of vales.
+     *
+     * <ul>
+     *   <li>jakarta.jms.Message = 0</li>
+     *   <li>jakarta.jms.ObjectMessage = 1</li>
+     *   <li>jakarta.jms.MapMessage = 2</li>
+     *   <li>jakarta.jms.BytesMessage = 3</li>
+     *   <li>jakarta.jms.StreamMessage = 4</li>
+     *   <li>jakarta.jms.TextMessage = 5</li>
+     * </ul>
+     *
+     * @return a byte value that represents the message type.
+     */
+    public byte getJmsMsgType();
+
     /**
      * Returns the property names for this Message instance. The Set returned may be
      * manipulated by the receiver without impacting the facade, and an empty set
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/message/AmqpMessageSupport.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/message/AmqpMessageSupport.java
@@ -19,6 +19,7 @@
 import java.nio.ByteBuffer;
 import java.util.Map;
 
+import org.apache.qpid.jms.message.facade.JmsMessageFacade;
 import org.apache.qpid.proton.amqp.Symbol;
 import org.apache.qpid.proton.amqp.messaging.MessageAnnotations;
 import org.apache.qpid.proton.codec.ReadableBuffer;
@@ -47,37 +48,37 @@ public final class AmqpMessageSupport {
      * Value mapping for JMS_MSG_TYPE which indicates the message is a generic JMS Message
      * which has no body.
      */
-    public static final byte JMS_MESSAGE = 0;
+    public static final byte JMS_MESSAGE = JmsMessageFacade.JMS_MESSAGE;
 
     /**
      * Value mapping for JMS_MSG_TYPE which indicates the message is a JMS ObjectMessage
      * which has an Object value serialized in its message body.
      */
-    public static final byte JMS_OBJECT_MESSAGE = 1;
+    public static final byte JMS_OBJECT_MESSAGE = JmsMessageFacade.JMS_OBJECT_MESSAGE;
 
     /**
      * Value mapping for JMS_MSG_TYPE which indicates the message is a JMS MapMessage
      * which has an Map instance serialized in its message body.
      */
-    public static final byte JMS_MAP_MESSAGE = 2;
+    public static final byte JMS_MAP_MESSAGE = JmsMessageFacade.JMS_MAP_MESSAGE;
 
     /**
      * Value mapping for JMS_MSG_TYPE which indicates the message is a JMS BytesMessage
      * which has a body that consists of raw bytes.
      */
-    public static final byte JMS_BYTES_MESSAGE = 3;
+    public static final byte JMS_BYTES_MESSAGE = JmsMessageFacade.JMS_BYTES_MESSAGE;
 
     /**
      * Value mapping for JMS_MSG_TYPE which indicates the message is a JMS StreamMessage
      * which has a body that is a structured collection of primitives values.
      */
-    public static final byte JMS_STREAM_MESSAGE = 4;
+    public static final byte JMS_STREAM_MESSAGE = JmsMessageFacade.JMS_STREAM_MESSAGE;
 
     /**
      * Value mapping for JMS_MSG_TYPE which indicates the message is a JMS TextMessage
      * which has a body that contains a UTF-8 encoded String.
      */
-    public static final byte JMS_TEXT_MESSAGE = 5;
+    public static final byte JMS_TEXT_MESSAGE = JmsMessageFacade.JMS_TEXT_MESSAGE;
 
     public static final String JMS_AMQP_TTL = "JMS_AMQP_TTL";
     public static final String JMS_AMQP_REPLY_TO_GROUP_ID = "JMS_AMQP_REPLY_TO_GROUP_ID";
diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/JMSConsumerIntegrationTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/JMSConsumerIntegrationTest.java
@@ -27,8 +27,10 @@
 
 import java.io.ByteArrayOutputStream;
 import java.io.ObjectOutputStream;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.LinkedHashMap;
+import java.util.List;
 import java.util.Map;
 
 import javax.jms.IllegalStateRuntimeException;
@@ -39,6 +41,7 @@
 import javax.jms.MessageFormatRuntimeException;
 import javax.jms.MessageListener;
 import javax.jms.Queue;
+import javax.jms.StreamMessage;
 
 import org.apache.qpid.jms.provider.amqp.message.AmqpMessageSupport;
 import org.apache.qpid.jms.test.QpidJmsTestCase;
@@ -635,4 +638,129 @@ public void testReceiveBodyBytesMessageFailsWhenWrongTypeRequested() throws Exce
             testPeer.waitForAllHandlersToComplete(3000);
         }
     }
+
+    @Test
+    @Timeout(20)
+    public void testReceiveBodyOnMessageFailsWhenAnyTypeRequested() throws Exception {
+        try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
+            JMSContext context = testFixture.createJMSContext(testPeer);
+
+            testPeer.expectBegin();
+
+            Queue queue = context.createQueue("myQueue");
+
+            PropertiesDescribedType properties = new PropertiesDescribedType();
+
+            MessageAnnotationsDescribedType msgAnnotations = null;
+            msgAnnotations = new MessageAnnotationsDescribedType();
+            msgAnnotations.setSymbolKeyedAnnotation(AmqpMessageSupport.JMS_MSG_TYPE.toString(), AmqpMessageSupport.JMS_MESSAGE);
+
+            testPeer.expectReceiverAttach();
+            testPeer.expectLinkFlowRespondWithTransfer(null, msgAnnotations, properties, null, null, 1);
+            testPeer.expectDispositionThatIsAcceptedAndSettled();
+
+            JMSConsumer messageConsumer = context.createConsumer(queue);
+            try {
+                messageConsumer.receiveBody(String.class, 3000);
+                fail("Should not read as String type");
+            } catch (MessageFormatRuntimeException mfre) {
+            }
+
+            try {
+                messageConsumer.receiveBody(byte[].class, 3000);
+                fail("Should not read as String type");
+            } catch (MessageFormatRuntimeException mfre) {
+            }
+
+            try {
+                messageConsumer.receiveBody(Object.class, 3000);
+                fail("Should not read as String type");
+            } catch (MessageFormatRuntimeException mfre) {
+            }
+
+            try {
+                messageConsumer.receiveBody(Map.class, 3000);
+                fail("Should not read as Map type");
+            } catch (MessageFormatRuntimeException mfre) {
+            }
+
+            final Message received = messageConsumer.receive(1000);
+
+            assertNotNull(received);
+
+            testPeer.waitForAllHandlersToComplete(3000);
+            testPeer.expectEnd();
+            testPeer.expectClose();
+
+            context.close();
+
+            testPeer.waitForAllHandlersToComplete(3000);
+        }
+    }
+
+    @Test
+    @Timeout(20)
+    public void testReceiveBodyOnStreamMessageFailsWhenAnyTypeRequested() throws Exception {
+        try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
+            JMSContext context = testFixture.createJMSContext(testPeer);
+
+            testPeer.expectBegin();
+
+            Queue queue = context.createQueue("myQueue");
+
+            List<Object> list = new ArrayList<Object>();
+            list.add(true);
+            list.add(false);
+            list.add("test");
+
+            PropertiesDescribedType properties = new PropertiesDescribedType();
+
+            MessageAnnotationsDescribedType msgAnnotations = null;
+            msgAnnotations = new MessageAnnotationsDescribedType();
+            msgAnnotations.setSymbolKeyedAnnotation(AmqpMessageSupport.JMS_MSG_TYPE.toString(), AmqpMessageSupport.JMS_STREAM_MESSAGE);
+
+            DescribedType amqpValueSectionContent = new AmqpValueDescribedType(list);
+
+            testPeer.expectReceiverAttach();
+            testPeer.expectLinkFlowRespondWithTransfer(null, msgAnnotations, properties, null, amqpValueSectionContent, 1);
+            testPeer.expectDispositionThatIsAcceptedAndSettled();
+
+            JMSConsumer messageConsumer = context.createConsumer(queue);
+            try {
+                messageConsumer.receiveBody(String.class, 3000);
+                fail("Should not read as String type");
+            } catch (MessageFormatRuntimeException mfre) {
+            }
+
+            try {
+                messageConsumer.receiveBody(byte[].class, 3000);
+                fail("Should not read as String type");
+            } catch (MessageFormatRuntimeException mfre) {
+            }
+
+            try {
+                messageConsumer.receiveBody(Object.class, 3000);
+                fail("Should not read as String type");
+            } catch (MessageFormatRuntimeException mfre) {
+            }
+
+            try {
+                messageConsumer.receiveBody(Map.class, 3000);
+                fail("Should not read as Map type");
+            } catch (MessageFormatRuntimeException mfre) {
+            }
+
+            final StreamMessage received = (StreamMessage) messageConsumer.receive(1000);
+
+            assertNotNull(received);
+
+            testPeer.waitForAllHandlersToComplete(3000);
+            testPeer.expectEnd();
+            testPeer.expectClose();
+
+            context.close();
+
+            testPeer.waitForAllHandlersToComplete(3000);
+        }
+    }
 }
diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/message/facade/test/JmsTestMessageFacade.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/message/facade/test/JmsTestMessageFacade.java
@@ -28,6 +28,7 @@
 import org.apache.qpid.jms.JmsDestination;
 import org.apache.qpid.jms.message.facade.JmsMessageFacade;
 
+
 /**
  * A test implementation of the JmsMessageFaceade that provides a generic
  * message instance which can be used instead of implemented in Provider specific
@@ -76,6 +77,11 @@ public JmsMsgType getMsgType() {
         return JmsMsgType.MESSAGE;
     }
 
+    @Override
+    public byte getJmsMsgType() {
+        return (byte) getMsgType().ordinal();
+    }
+
     @Override
     public JmsTestMessageFacade copy() {
         JmsTestMessageFacade copy = new JmsTestMessageFacade();
