feat(#1716): Implement receiveBody()

jbonofre · jbonofre · commit d2fde0e4fca8 · 2026-03-26T20:38:01.000+01:00
Implement JMS 2.0 / Jakarta Messaging 3.1 receiveBody() methods on
ActiveMQMessageConsumer and wire them through ActiveMQConsumer (JMSConsumer).

The doReceiveBody() helper handles ack-mode-dependent failure behavior
per section 8.6 of the Jakarta Messaging 3.1 specification:
- AUTO_ACK / DUPS_OK: re-enqueue without acknowledgement
- CLIENT_ACK / TRANSACTED: treat as delivered (credit window expansion)

Plain Message and StreamMessage types are always rejected with
MessageFormatException, per spec (even though Message.isBodyAssignableTo
returns true for any type).

ActiveMQObjectMessage.isBodyAssignableTo now returns false on
deserialization failure instead of propagating the exception.
diff --git a/activemq-client/src/main/java/org/apache/activemq/ActiveMQConsumer.java b/activemq-client/src/main/java/org/apache/activemq/ActiveMQConsumer.java
@@ -19,26 +19,26 @@
 import jakarta.jms.JMSConsumer;
 import jakarta.jms.JMSException;
 import jakarta.jms.JMSRuntimeException;
-import jakarta.jms.Message;
 import jakarta.jms.MessageConsumer;
 import jakarta.jms.MessageListener;
+import jakarta.jms.Message;
 
 import org.apache.activemq.util.JMSExceptionSupport;
 
 public class ActiveMQConsumer implements JMSConsumer {
-    
+
     private final ActiveMQContext activemqContext;
-    private final MessageConsumer activemqMessageConsumer;
+    private final ActiveMQMessageConsumer consumer;
 
-    ActiveMQConsumer(ActiveMQContext activemqContext, MessageConsumer activemqMessageConsumer) {
+    ActiveMQConsumer(ActiveMQContext activemqContext, MessageConsumer consumer) {
         this.activemqContext = activemqContext;
-        this.activemqMessageConsumer = activemqMessageConsumer;
+        this.consumer = (ActiveMQMessageConsumer) consumer;
     }
 
     @Override
     public String getMessageSelector() {
         try {
-            return activemqMessageConsumer.getMessageSelector();
+            return consumer.getMessageSelector();
         } catch (JMSException e) {
             throw JMSExceptionSupport.convertToJMSRuntimeException(e);
         }
@@ -47,7 +47,7 @@ public String getMessageSelector() {
     @Override
     public MessageListener getMessageListener() throws JMSRuntimeException {
         try {
-            return activemqMessageConsumer.getMessageListener();
+            return consumer.getMessageListener();
         } catch (JMSException e) {
             throw JMSExceptionSupport.convertToJMSRuntimeException(e);
         }
@@ -56,7 +56,7 @@ public MessageListener getMessageListener() throws JMSRuntimeException {
     @Override
     public void setMessageListener(MessageListener listener) throws JMSRuntimeException {
         try {
-            activemqMessageConsumer.setMessageListener(listener);
+            consumer.setMessageListener(listener);
         } catch (JMSException e) {
             throw JMSExceptionSupport.convertToJMSRuntimeException(e);
         }
@@ -65,7 +65,7 @@ public void setMessageListener(MessageListener listener) throws JMSRuntimeExcept
     @Override
     public Message receive() {
         try {
-            return activemqMessageConsumer.receive();
+            return consumer.receive();
         } catch (JMSException e) {
             throw JMSExceptionSupport.convertToJMSRuntimeException(e);
         }
@@ -74,7 +74,7 @@ public Message receive() {
     @Override
     public Message receive(long timeout) {
         try {
-            return activemqMessageConsumer.receive(timeout);
+            return consumer.receive(timeout);
         } catch (JMSException e) {
             throw JMSExceptionSupport.convertToJMSRuntimeException(e);
         }
@@ -83,7 +83,7 @@ public Message receive(long timeout) {
     @Override
     public Message receiveNoWait() {
         try {
-            return activemqMessageConsumer.receiveNoWait();
+            return consumer.receiveNoWait();
         } catch (JMSException e) {
             throw JMSExceptionSupport.convertToJMSRuntimeException(e);
         }
@@ -92,25 +92,37 @@ public Message receiveNoWait() {
     @Override
     public void close() {
         try {
-            activemqMessageConsumer.close();
+            consumer.close();
         } catch (JMSException e) {
             throw JMSExceptionSupport.convertToJMSRuntimeException(e);
-        }        
+        }
     }
 
     @Override
     public <T> T receiveBody(Class<T> c) {
-        throw new UnsupportedOperationException("receiveBody(Class<T>) is not supported");
+        try {
+            return consumer.receiveBody(c);
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
     }
 
     @Override
     public <T> T receiveBody(Class<T> c, long timeout) {
-        throw new UnsupportedOperationException("receiveBody(Class<T>, long) is not supported");
+        try {
+            return consumer.receiveBody(c, timeout);
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
     }
 
     @Override
     public <T> T receiveBodyNoWait(Class<T> c) {
-        throw new UnsupportedOperationException("receiveBodyNoWait(Class<T>) is not supported");
+        try {
+            return consumer.receiveBodyNoWait(c);
+        } catch (JMSException e) {
+            throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+        }
     }
 
 }
diff --git a/activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java b/activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java
@@ -35,6 +35,7 @@
 import jakarta.jms.JMSException;
 import jakarta.jms.Message;
 import jakarta.jms.MessageConsumer;
+import jakarta.jms.MessageFormatException;
 import jakarta.jms.MessageListener;
 import jakarta.jms.TransactionRolledBackException;
 
@@ -43,6 +44,7 @@
 import org.apache.activemq.command.ActiveMQDestination;
 import org.apache.activemq.command.ActiveMQMessage;
 import org.apache.activemq.command.ActiveMQObjectMessage;
+import org.apache.activemq.command.ActiveMQStreamMessage;
 import org.apache.activemq.command.ActiveMQTempDestination;
 import org.apache.activemq.command.CommandTypes;
 import org.apache.activemq.command.ConsumerId;
@@ -713,6 +715,221 @@ public Message receiveNoWait() throws JMSException {
         return createActiveMQMessage(md);
     }
 
+    /**
+     * Receives the next message produced for this message consumer and returns
+     * its body as an object of the specified type. This call blocks
+     * indefinitely until a message is produced or until this message consumer
+     * is closed.
+     * <p>
+     * If the message is not of a type for which the body can be assigned to
+     * the specified type, a {@code MessageFormatException} is thrown. The
+     * subsequent behaviour depends on the session's acknowledge mode:
+     * <ul>
+     *   <li>{@code AUTO_ACKNOWLEDGE} / {@code DUPS_OK_ACKNOWLEDGE}: the
+     *       message is not acknowledged and will be delivered again before any
+     *       subsequent messages. This is not considered redelivery and does not
+     *       cause the {@code JMSRedelivered} header or
+     *       {@code JMSXDeliveryCount} property to be updated.</li>
+     *   <li>{@code CLIENT_ACKNOWLEDGE}: the message is treated as delivered.
+     *       The application must call {@code session.recover()} to have it
+     *       redelivered.</li>
+     *   <li>Transacted session: the message is treated as delivered within the
+     *       transaction. The application must call {@code session.rollback()}
+     *       to have it redelivered.</li>
+     * </ul>
+     * <p>
+     * This method cannot be used to receive {@code Message} or
+     * {@code StreamMessage} objects; a {@code MessageFormatException} will
+     * always be thrown for these types.
+     *
+     * @param c the type to which the body of the next message should be
+     *          assigned
+     * @return the body of the next message, or null if this message consumer
+     *         is concurrently closed
+     * @throws MessageFormatException if the message body cannot be assigned to
+     *         the specified type, or if the message is a {@code Message} or
+     *         {@code StreamMessage}
+     * @throws JMSException if the JMS provider fails to receive the next
+     *         message due to some internal error
+     */
+    public <T> T receiveBody(Class<T> c) throws JMSException {
+        checkClosed();
+        checkMessageListener();
+
+        sendPullCommand(0);
+        MessageDispatch md = dequeue(-1);
+        if (md == null) {
+            return null;
+        }
+
+        return doReceiveBody(md, c);
+    }
+
+    /**
+     * Receives the next message produced for this message consumer and returns
+     * its body as an object of the specified type, blocking up to the
+     * specified timeout. A {@code timeout} of zero never expires and the call
+     * blocks indefinitely.
+     * <p>
+     * If the message is not of a type for which the body can be assigned to
+     * the specified type, a {@code MessageFormatException} is thrown. The
+     * subsequent behaviour depends on the session's acknowledge mode:
+     * <ul>
+     *   <li>{@code AUTO_ACKNOWLEDGE} / {@code DUPS_OK_ACKNOWLEDGE}: the
+     *       message is not acknowledged and will be delivered again before any
+     *       subsequent messages. This is not considered redelivery and does not
+     *       cause the {@code JMSRedelivered} header or
+     *       {@code JMSXDeliveryCount} property to be updated.</li>
+     *   <li>{@code CLIENT_ACKNOWLEDGE}: the message is treated as delivered.
+     *       The application must call {@code session.recover()} to have it
+     *       redelivered.</li>
+     *   <li>Transacted session: the message is treated as delivered within the
+     *       transaction. The application must call {@code session.rollback()}
+     *       to have it redelivered.</li>
+     * </ul>
+     * <p>
+     * This method cannot be used to receive {@code Message} or
+     * {@code StreamMessage} objects; a {@code MessageFormatException} will
+     * always be thrown for these types.
+     *
+     * @param c       the type to which the body of the next message should be
+     *                assigned
+     * @param timeout the timeout value (in milliseconds), a timeout of zero
+     *                never expires
+     * @return the body of the next message, or null if the timeout expires or
+     *         this message consumer is concurrently closed
+     * @throws MessageFormatException if the message body cannot be assigned to
+     *         the specified type, or if the message is a {@code Message} or
+     *         {@code StreamMessage}
+     * @throws JMSException if the JMS provider fails to receive the next
+     *         message due to some internal error
+     */
+    public <T> T receiveBody(Class<T> c, long timeout) throws JMSException {
+        checkClosed();
+        checkMessageListener();
+        if (timeout == 0) {
+            return this.receiveBody(c);
+        }
+
+        sendPullCommand(timeout);
+        while (timeout > 0) {
+            MessageDispatch md;
+            if (info.getPrefetchSize() == 0) {
+                md = dequeue(-1);
+            } else {
+                md = dequeue(timeout);
+            }
+
+            if (md == null) {
+                return null;
+            }
+
+            return doReceiveBody(md, c);
+        }
+        return null;
+    }
+
+    /**
+     * Receives the next message produced for this message consumer and returns
+     * its body as an object of the specified type if one is immediately
+     * available.
+     * <p>
+     * If the message is not of a type for which the body can be assigned to
+     * the specified type, a {@code MessageFormatException} is thrown. The
+     * subsequent behaviour depends on the session's acknowledge mode:
+     * <ul>
+     *   <li>{@code AUTO_ACKNOWLEDGE} / {@code DUPS_OK_ACKNOWLEDGE}: the
+     *       message is not acknowledged and will be delivered again before any
+     *       subsequent messages. This is not considered redelivery and does not
+     *       cause the {@code JMSRedelivered} header or
+     *       {@code JMSXDeliveryCount} property to be updated.</li>
+     *   <li>{@code CLIENT_ACKNOWLEDGE}: the message is treated as delivered.
+     *       The application must call {@code session.recover()} to have it
+     *       redelivered.</li>
+     *   <li>Transacted session: the message is treated as delivered within the
+     *       transaction. The application must call {@code session.rollback()}
+     *       to have it redelivered.</li>
+     * </ul>
+     * <p>
+     * This method cannot be used to receive {@code Message} or
+     * {@code StreamMessage} objects; a {@code MessageFormatException} will
+     * always be thrown for these types.
+     *
+     * @param c the type to which the body of the next message should be
+     *          assigned
+     * @return the body of the next message, or null if one is not immediately
+     *         available
+     * @throws MessageFormatException if the message body cannot be assigned to
+     *         the specified type, or if the message is a {@code Message} or
+     *         {@code StreamMessage}
+     * @throws JMSException if the JMS provider fails to receive the next
+     *         message due to some internal error
+     */
+    public <T> T receiveBodyNoWait(Class<T> c) throws JMSException {
+        checkClosed();
+        checkMessageListener();
+        sendPullCommand(-1);
+
+        MessageDispatch md;
+        if (info.getPrefetchSize() == 0) {
+            md = dequeue(-1);
+        } else {
+            md = dequeue(0);
+        }
+
+        if (md == null) {
+            return null;
+        }
+
+        return doReceiveBody(md, c);
+    }
+
+    /**
+     * Checks that the message body can be assigned to the requested type,
+     * acknowledges the message, and returns its body.  If the body cannot be
+     * assigned, the handling depends on the session's acknowledge mode:
+     * <ul>
+     *   <li>AUTO_ACKNOWLEDGE / DUPS_OK_ACKNOWLEDGE: the message is re-enqueued
+     *       without acknowledgement so that it remains available for a subsequent
+     *       {@code receive} or {@code receiveBody} call.</li>
+     *   <li>CLIENT_ACKNOWLEDGE / TRANSACTED: the message is treated as delivered
+     *       (not re-enqueued). The application may call {@code session.recover()}
+     *       or {@code session.rollback()} respectively to redeliver.</li>
+     * </ul>
+     * <p>
+     * Per Jakarta Messaging 3.1, {@code receiveBody} must always throw
+     * {@code MessageFormatException} for plain {@code Message} and
+     * {@code StreamMessage} types, regardless of what
+     * {@code isBodyAssignableTo} returns.
+     */
+    private <T> T doReceiveBody(MessageDispatch md, Class<T> c) throws JMSException {
+        ActiveMQMessage message = createActiveMQMessage(md);
+
+        // Jakarta Messaging 3.1: receiveBody must always fail for Message and StreamMessage.
+        // Note: Message.isBodyAssignableTo() returns true for any type per spec,
+        // which conflicts with receiveBody's requirement, so we must check explicitly.
+        boolean bodyNotAssignable = message.getClass() == ActiveMQMessage.class
+                || message instanceof ActiveMQStreamMessage
+                || !message.isBodyAssignableTo(c);
+
+        if (bodyNotAssignable) {
+            if (session.isAutoAcknowledge() || session.isDupsOkAcknowledge()) {
+                // re-enqueue for redelivery on next receive/receiveBody call
+                unconsumedMessages.enqueueFirst(md);
+            } else {
+                // CLIENT_ACKNOWLEDGE or TRANSACTED: message is considered delivered,
+                // application must use session.recover() or session.rollback()
+                beforeMessageIsConsumed(md);
+                afterMessageIsConsumed(md, false);
+            }
+            throw new MessageFormatException("Message body cannot be read as type: " + c);
+        }
+
+        beforeMessageIsConsumed(md);
+        afterMessageIsConsumed(md, false);
+        return message.getBody(c);
+    }
+
     /**
      * Closes the message consumer.
      * <P>
diff --git a/activemq-client/src/main/java/org/apache/activemq/command/ActiveMQObjectMessage.java b/activemq-client/src/main/java/org/apache/activemq/command/ActiveMQObjectMessage.java
@@ -288,8 +288,15 @@ public void initTransients() {
     }
 
     @Override
-    public boolean isBodyAssignableTo(Class c) throws JMSException {
-        final Serializable object = getObject();
+    public boolean isBodyAssignableTo(Class c) {
+        final Serializable object;
+        try {
+            object = getObject();
+        } catch (JMSException e) {
+            // Per Jakarta Messaging 3.1: if the message is an ObjectMessage
+            // and object deserialization fails then false is returned.
+            return false;
+        }
         if (object == null) {
             return true;
         }
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2ReceiveBodyTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2ReceiveBodyTest.java
