feat(#1716): Implement receiveBody()

jbonofre · jbonofre · commit 8a9ed7680944 · 2026-03-26T06:07:07.000+01:00
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;
 
@@ -713,6 +714,144 @@ 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 and the
+     * message is not acknowledged. It may be delivered again when a subsequent
+     * {@code receive} or {@code receiveBody} call is made.
+     *
+     * @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
+     * @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 and the
+     * message is not acknowledged. It may be delivered again when a subsequent
+     * {@code receive} or {@code receiveBody} call is made.
+     *
+     * @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
+     * @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 and the
+     * message is not acknowledged. It may be delivered again when a subsequent
+     * {@code receive} or {@code receiveBody} call is made.
+     *
+     * @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
+     * @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 message is re-enqueued without acknowledgement so that it
+     * remains available for a subsequent {@code receive} or
+     * {@code receiveBody} call.
+     */
+    private <T> T doReceiveBody(MessageDispatch md, Class<T> c) throws JMSException {
+        ActiveMQMessage message = createActiveMQMessage(md);
+        if (!message.isBodyAssignableTo(c)) {
+            // spec: message is not acknowledged and left for redelivery
+            unconsumedMessages.enqueueFirst(md);
+            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-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2ReceiveBodyTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2ReceiveBodyTest.java
