This is an automated email from the ASF dual-hosted git repository.
tabish pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/qpid-jms.git
The following commit(s) were added to refs/heads/main by this push:
new de04f375 QPIDJMS-622 receiveBody must throw if type is Message or
StreamMessage
de04f375 is described below
commit de04f3757d8762cb987bba27c1f6d635a1a0de53
Author: Timothy Bish <[email protected]>
AuthorDate: Thu Mar 12 15:00:22 2026 -0400
QPIDJMS-622 receiveBody must throw if type is Message or StreamMessage
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.
---
.../org/apache/qpid/jms/JmsMessageConsumer.java | 28 ++--
.../qpid/jms/message/facade/JmsMessageFacade.java | 27 +++-
.../provider/amqp/message/AmqpMessageSupport.java | 13 +-
.../integration/JMSConsumerIntegrationTest.java | 146 +++++++++++++++++++--
.../message/facade/test/JmsTestMessageFacade.java | 9 +-
5 files changed, 195 insertions(+), 28 deletions(-)
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
index 3aeff646..0923f975 100644
--- 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
@@ -23,18 +23,11 @@ import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
-import jakarta.jms.IllegalStateException;
-import jakarta.jms.JMSException;
-import jakarta.jms.Message;
-import jakarta.jms.MessageConsumer;
-import jakarta.jms.MessageFormatException;
-import jakarta.jms.MessageListener;
-import jakarta.jms.Session;
-
import org.apache.qpid.jms.exceptions.JmsConnectionFailedException;
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;
@@ -55,6 +48,14 @@ import org.apache.qpid.jms.util.PriorityMessageQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import jakarta.jms.IllegalStateException;
+import jakarta.jms.JMSException;
+import jakarta.jms.Message;
+import jakarta.jms.MessageConsumer;
+import jakarta.jms.MessageFormatException;
+import jakarta.jms.MessageListener;
+import jakarta.jms.Session;
+
/**
* implementation of a JMS Message Consumer
*/
@@ -251,7 +252,16 @@ public class JmsMessageConsumer implements AutoCloseable,
MessageConsumer, JmsMe
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
index 80a6e445..7674ae30 100644
---
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
@@ -18,11 +18,11 @@ package org.apache.qpid.jms.message.facade;
import java.util.Set;
-import jakarta.jms.JMSException;
-
import org.apache.qpid.jms.JmsDestination;
import org.apache.qpid.jms.tracing.TraceableMessage;
+import jakarta.jms.JMSException;
+
/**
* The Message Facade interface defines the required mapping between a
Provider's
* own Message type and the JMS Message types. A Provider can implement the
Facade
@@ -31,6 +31,29 @@ import org.apache.qpid.jms.tracing.TraceableMessage;
*/
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
index f448e723..f28c6acf 100644
---
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 @@ package org.apache.qpid.jms.provider.amqp.message;
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
index c41f0f4b..ba503208 100644
---
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,19 +27,12 @@ import static org.junit.jupiter.api.Assertions.fail;
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 jakarta.jms.IllegalStateRuntimeException;
-import jakarta.jms.JMSConsumer;
-import jakarta.jms.JMSContext;
-import jakarta.jms.JMSRuntimeException;
-import jakarta.jms.Message;
-import jakarta.jms.MessageFormatRuntimeException;
-import jakarta.jms.MessageListener;
-import jakarta.jms.Queue;
-
import org.apache.qpid.jms.provider.amqp.message.AmqpMessageSupport;
import org.apache.qpid.jms.test.QpidJmsTestCase;
import org.apache.qpid.jms.test.Wait;
@@ -57,6 +50,16 @@ import org.junit.jupiter.api.Timeout;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import jakarta.jms.IllegalStateRuntimeException;
+import jakarta.jms.JMSConsumer;
+import jakarta.jms.JMSContext;
+import jakarta.jms.JMSRuntimeException;
+import jakarta.jms.Message;
+import jakarta.jms.MessageFormatRuntimeException;
+import jakarta.jms.MessageListener;
+import jakarta.jms.Queue;
+import jakarta.jms.StreamMessage;
+
public class JMSConsumerIntegrationTest extends QpidJmsTestCase {
private static final Logger LOG =
LoggerFactory.getLogger(JMSConsumerIntegrationTest.class);
@@ -635,4 +638,129 @@ public class JMSConsumerIntegrationTest extends
QpidJmsTestCase {
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
index c85932d8..b80dc916 100644
---
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
@@ -23,11 +23,11 @@ import java.util.Map;
import java.util.Set;
import java.util.function.BiConsumer;
-import jakarta.jms.JMSException;
-
import org.apache.qpid.jms.JmsDestination;
import org.apache.qpid.jms.message.facade.JmsMessageFacade;
+import jakarta.jms.JMSException;
+
/**
* A test implementation of the JmsMessageFaceade that provides a generic
* message instance which can be used instead of implemented in Provider
specific
@@ -76,6 +76,11 @@ public class JmsTestMessageFacade implements
JmsMessageFacade {
return JmsMsgType.MESSAGE;
}
+ @Override
+ public byte getJmsMsgType() {
+ return (byte) getMsgType().ordinal();
+ }
+
@Override
public JmsTestMessageFacade copy() {
JmsTestMessageFacade copy = new JmsTestMessageFacade();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
