From d2fde0e4fca8421ce7fa50f5821e33526b065616 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?JB=20Onofr=C3=A9?= Date: Sat, 7 Mar 2026 18:12:08 +0100 Subject: [PATCH] feat(#1716): Implement receiveBody() 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. --- .../org/apache/activemq/ActiveMQConsumer.java | 44 +- .../activemq/ActiveMQMessageConsumer.java | 217 ++++++++ .../command/ActiveMQObjectMessage.java | 11 +- .../jms2/ActiveMQJMS2ReceiveBodyTest.java | 500 ++++++++++++++++++ 4 files changed, 754 insertions(+), 18 deletions(-) create mode 100644 activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2ReceiveBodyTest.java diff --git a/activemq-client/src/main/java/org/apache/activemq/ActiveMQConsumer.java b/activemq-client/src/main/java/org/apache/activemq/ActiveMQConsumer.java index b104edb8dcb..2da3adf8227 100644 --- 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 receiveBody(Class c) { - throw new UnsupportedOperationException("receiveBody(Class) is not supported"); + try { + return consumer.receiveBody(c); + } catch (JMSException e) { + throw JMSExceptionSupport.convertToJMSRuntimeException(e); + } } @Override public T receiveBody(Class c, long timeout) { - throw new UnsupportedOperationException("receiveBody(Class, long) is not supported"); + try { + return consumer.receiveBody(c, timeout); + } catch (JMSException e) { + throw JMSExceptionSupport.convertToJMSRuntimeException(e); + } } @Override public T receiveBodyNoWait(Class c) { - throw new UnsupportedOperationException("receiveBodyNoWait(Class) 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 index a6bf1b20952..d7188b5151a 100644 --- 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. + *

+ * 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: + *

    + *
  • {@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.
    • + *
  • {@code CLIENT_ACKNOWLEDGE}: the message is treated as delivered. + * The application must call {@code session.recover()} to have it + * redelivered.
    • + *
  • Transacted session: the message is treated as delivered within the + * transaction. The application must call {@code session.rollback()} + * to have it redelivered.
    • + *
+ *

+ * 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 receiveBody(Class 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. + *

+ * 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: + *

    + *
  • {@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.
    • + *
  • {@code CLIENT_ACKNOWLEDGE}: the message is treated as delivered. + * The application must call {@code session.recover()} to have it + * redelivered.
    • + *
  • Transacted session: the message is treated as delivered within the + * transaction. The application must call {@code session.rollback()} + * to have it redelivered.
    • + *
+ *

+ * 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 receiveBody(Class 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. + *

+ * 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: + *

    + *
  • {@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.
    • + *
  • {@code CLIENT_ACKNOWLEDGE}: the message is treated as delivered. + * The application must call {@code session.recover()} to have it + * redelivered.
    • + *
  • Transacted session: the message is treated as delivered within the + * transaction. The application must call {@code session.rollback()} + * to have it redelivered.
    • + *
+ *

+ * 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 receiveBodyNoWait(Class 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: + *

    + *
  • 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.
    • + *
  • 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.
    • + *
+ *

+ * 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 doReceiveBody(MessageDispatch md, Class 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. *

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 index 79cbf4c0d5e..07a6171d4cc 100644 --- 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 new file mode 100644 index 00000000000..a9bfb8f5e0d --- /dev/null +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2ReceiveBodyTest.java @@ -0,0 +1,500 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.activemq.jms2; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.io.Serializable; +import java.util.Map; + +import jakarta.jms.Destination; +import jakarta.jms.JMSConsumer; +import jakarta.jms.JMSContext; +import jakarta.jms.JMSException; +import jakarta.jms.Message; +import jakarta.jms.MessageFormatRuntimeException; +import jakarta.jms.Session; +import jakarta.jms.StreamMessage; + +import org.junit.Test; + +public class ActiveMQJMS2ReceiveBodyTest extends ActiveMQJMS2TestBase { + + @Test + public void testReceiveBodyTextMessage() { + try (JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) { + Destination destination = jmsContext.createQueue(methodNameDestinationName); + jmsContext.createProducer().send(destination, "hello"); + + try (JMSConsumer jmsConsumer = jmsContext.createConsumer(destination)) { + jmsContext.start(); + String body = jmsConsumer.receiveBody(String.class, 5000); + assertEquals("hello", body); + } + } + } + + @Test + public void testReceiveBodyBytesMessage() { + try (JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) { + Destination destination = jmsContext.createQueue(methodNameDestinationName); + byte[] payload = new byte[] {1, 2, 3, 4, 5}; + jmsContext.createProducer().send(destination, payload); + + try (JMSConsumer jmsConsumer = jmsContext.createConsumer(destination)) { + jmsContext.start(); + byte[] body = jmsConsumer.receiveBody(byte[].class, 5000); + assertArrayEquals(payload, body); + } + } + } + + @Test + public void testReceiveBodyObjectMessage() { + try (JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) { + Destination destination = jmsContext.createQueue(methodNameDestinationName); + jmsContext.createProducer().send(destination, (Serializable) "testObject"); + + try (JMSConsumer jmsConsumer = jmsContext.createConsumer(destination)) { + jmsContext.start(); + Serializable body = jmsConsumer.receiveBody(Serializable.class, 5000); + assertEquals("testObject", body); + } + } + } + + @Test + public void testReceiveBodyMapMessage() { + try (JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) { + Destination destination = jmsContext.createQueue(methodNameDestinationName); + jmsContext.createProducer().send(destination, Map.of("key", "value")); + + try (JMSConsumer jmsConsumer = jmsContext.createConsumer(destination)) { + jmsContext.start(); + @SuppressWarnings("unchecked") + Map body = jmsConsumer.receiveBody(Map.class, 5000); + assertNotNull(body); + assertEquals("value", body.get("key")); + } + } + } + + @Test + public void testReceiveBodyBlockingTextMessage() { + try (JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) { + Destination destination = jmsContext.createQueue(methodNameDestinationName); + jmsContext.createProducer().send(destination, "blocking-test"); + + try (JMSConsumer jmsConsumer = jmsContext.createConsumer(destination)) { + jmsContext.start(); + String body = jmsConsumer.receiveBody(String.class); + assertEquals("blocking-test", body); + } + } + } + + @Test + public void testReceiveBodyNoWaitTextMessage() { + try (JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) { + Destination destination = jmsContext.createQueue(methodNameDestinationName); + jmsContext.createProducer().send(destination, "hello"); + + try (JMSConsumer jmsConsumer = jmsContext.createConsumer(destination)) { + jmsContext.start(); + // Give time for message to arrive in prefetch + Thread.sleep(500); + String body = jmsConsumer.receiveBodyNoWait(String.class); + assertEquals("hello", body); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + fail(e.getMessage()); + } + } + + @Test + public void testReceiveBodyNoWaitReturnsNullWhenEmpty() { + try (JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) { + Destination destination = jmsContext.createQueue(methodNameDestinationName); + + try (JMSConsumer jmsConsumer = jmsContext.createConsumer(destination)) { + jmsContext.start(); + String body = jmsConsumer.receiveBodyNoWait(String.class); + assertNull(body); + } + } + } + + @Test + public void testReceiveBodyWithTimeoutReturnsNullOnExpiry() { + try (JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) { + Destination destination = jmsContext.createQueue(methodNameDestinationName); + + try (JMSConsumer jmsConsumer = jmsContext.createConsumer(destination)) { + jmsContext.start(); + String body = jmsConsumer.receiveBody(String.class, 100); + assertNull(body); + } + } + } + + @Test(expected = MessageFormatRuntimeException.class) + public void testReceiveBodyThrowsMessageFormatRuntimeExceptionOnTypeMismatch() { + try (JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) { + Destination destination = jmsContext.createQueue(methodNameDestinationName); + jmsContext.createProducer().send(destination, new byte[] {1, 2, 3}); + + try (JMSConsumer jmsConsumer = jmsContext.createConsumer(destination)) { + jmsContext.start(); + jmsConsumer.receiveBody(String.class, 5000); + } + } + } + + @Test(expected = MessageFormatRuntimeException.class) + public void testReceiveBodyBlockingThrowsMessageFormatRuntimeExceptionOnTypeMismatch() { + try (JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) { + Destination destination = jmsContext.createQueue(methodNameDestinationName); + jmsContext.createProducer().send(destination, new byte[] {1, 2, 3}); + + try (JMSConsumer jmsConsumer = jmsContext.createConsumer(destination)) { + jmsContext.start(); + jmsConsumer.receiveBody(String.class); + } + } + } + + @Test(expected = MessageFormatRuntimeException.class) + public void testReceiveBodyNoWaitThrowsMessageFormatRuntimeExceptionOnTypeMismatch() { + try (JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) { + Destination destination = jmsContext.createQueue(methodNameDestinationName); + jmsContext.createProducer().send(destination, new byte[] {1, 2, 3}); + + try (JMSConsumer jmsConsumer = jmsContext.createConsumer(destination)) { + jmsContext.start(); + // Give time for message to arrive in prefetch + try { Thread.sleep(500); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } + jmsConsumer.receiveBodyNoWait(String.class); + } + } + } + + @Test + public void testReceiveBodyMessageNotAcknowledgedOnTypeMismatch() { + try (JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) { + Destination destination = jmsContext.createQueue(methodNameDestinationName); + jmsContext.createProducer().send(destination, new byte[] {1, 2, 3}); + + try (JMSConsumer jmsConsumer = jmsContext.createConsumer(destination)) { + jmsContext.start(); + + // Attempt with wrong type — message should not be acknowledged + try { + jmsConsumer.receiveBody(String.class, 5000); + fail("Should have thrown MessageFormatRuntimeException"); + } catch (MessageFormatRuntimeException e) { + // expected + } + + // Message should still be available via receive() + assertNotNull("Message should be redelivered after type mismatch", + jmsConsumer.receive(5000)); + } + } + } + + @Test + public void testReceiveBodyCorrectTypeAfterMismatch() { + try (JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) { + Destination destination = jmsContext.createQueue(methodNameDestinationName); + byte[] payload = new byte[] {1, 2, 3}; + jmsContext.createProducer().send(destination, payload); + + try (JMSConsumer jmsConsumer = jmsContext.createConsumer(destination)) { + jmsContext.start(); + + // Attempt with wrong type + try { + jmsConsumer.receiveBody(String.class, 5000); + fail("Should have thrown MessageFormatRuntimeException"); + } catch (MessageFormatRuntimeException e) { + // expected + } + + // Retry with the correct type — should succeed + byte[] body = jmsConsumer.receiveBody(byte[].class, 5000); + assertArrayEquals(payload, body); + } + } + } + + // ---- Plain Message type: receiveBody must always throw ---- + + @Test + public void testReceiveBodyPlainMessageThrows() { + try (JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) { + Destination destination = jmsContext.createQueue(methodNameDestinationName); + Message plainMessage = jmsContext.createMessage(); + jmsContext.createProducer().send(destination, plainMessage); + + try (JMSConsumer jmsConsumer = jmsContext.createConsumer(destination)) { + jmsContext.start(); + try { + jmsConsumer.receiveBody(String.class, 5000); + fail("Should have thrown MessageFormatRuntimeException for plain Message"); + } catch (MessageFormatRuntimeException e) { + // expected per spec + } + } + } + } + + @Test + public void testReceiveBodyPlainMessageAutoAckRedelivered() { + try (JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) { + Destination destination = jmsContext.createQueue(methodNameDestinationName); + Message plainMessage = jmsContext.createMessage(); + jmsContext.createProducer().send(destination, plainMessage); + + try (JMSConsumer jmsConsumer = jmsContext.createConsumer(destination)) { + jmsContext.start(); + try { + jmsConsumer.receiveBody(String.class, 5000); + fail("Should have thrown MessageFormatRuntimeException"); + } catch (MessageFormatRuntimeException e) { + // expected + } + + // AUTO_ACK: message should be re-enqueued and available via receive() + Message redelivered = jmsConsumer.receive(5000); + assertNotNull("Plain Message should be redelivered in AUTO_ACK mode", redelivered); + } + } + } + + // ---- StreamMessage type: receiveBody must always throw ---- + + @Test + public void testReceiveBodyStreamMessageThrows() throws JMSException { + try (JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) { + Destination destination = jmsContext.createQueue(methodNameDestinationName); + StreamMessage streamMessage = jmsContext.createStreamMessage(); + streamMessage.writeString("payload"); + jmsContext.createProducer().send(destination, streamMessage); + + try (JMSConsumer jmsConsumer = jmsContext.createConsumer(destination)) { + jmsContext.start(); + try { + jmsConsumer.receiveBody(String.class, 5000); + fail("Should have thrown MessageFormatRuntimeException for StreamMessage"); + } catch (MessageFormatRuntimeException e) { + // expected per spec + } + } + } + } + + @Test + public void testReceiveBodyStreamMessageAutoAckRedelivered() throws JMSException { + try (JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.AUTO_ACKNOWLEDGE)) { + Destination destination = jmsContext.createQueue(methodNameDestinationName); + StreamMessage streamMessage = jmsContext.createStreamMessage(); + streamMessage.writeString("payload"); + jmsContext.createProducer().send(destination, streamMessage); + + try (JMSConsumer jmsConsumer = jmsContext.createConsumer(destination)) { + jmsContext.start(); + try { + jmsConsumer.receiveBody(String.class, 5000); + fail("Should have thrown MessageFormatRuntimeException"); + } catch (MessageFormatRuntimeException e) { + // expected + } + + // AUTO_ACK: message should be re-enqueued and available via receive() + Message redelivered = jmsConsumer.receive(5000); + assertNotNull("StreamMessage should be redelivered in AUTO_ACK mode", redelivered); + } + } + } + + // ---- CLIENT_ACKNOWLEDGE: message treated as delivered on failure ---- + + @Test + public void testReceiveBodyClientAckMessageTreatedAsDelivered() { + try (JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.CLIENT_ACKNOWLEDGE)) { + Destination destination = jmsContext.createQueue(methodNameDestinationName); + jmsContext.createProducer().send(destination, new byte[] {1, 2, 3}); + + try (JMSConsumer jmsConsumer = jmsContext.createConsumer(destination)) { + jmsContext.start(); + + // receiveBody with wrong type should throw + try { + jmsConsumer.receiveBody(String.class, 5000); + fail("Should have thrown MessageFormatRuntimeException"); + } catch (MessageFormatRuntimeException e) { + // expected + } + + // CLIENT_ACK: message is treated as delivered, not re-enqueued. + // A subsequent receive without recover should not return the same message. + Message msg = jmsConsumer.receiveNoWait(); + assertNull("Message should not be re-enqueued in CLIENT_ACK mode", msg); + } + } + } + + @Test + public void testReceiveBodyClientAckRecoverRedelivers() throws JMSException { + try (JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.CLIENT_ACKNOWLEDGE)) { + Destination destination = jmsContext.createQueue(methodNameDestinationName); + jmsContext.createProducer().send(destination, new byte[] {1, 2, 3}); + + try (JMSConsumer jmsConsumer = jmsContext.createConsumer(destination)) { + jmsContext.start(); + + try { + jmsConsumer.receiveBody(String.class, 5000); + fail("Should have thrown MessageFormatRuntimeException"); + } catch (MessageFormatRuntimeException e) { + // expected + } + + // After recover(), the message should be redelivered + jmsContext.recover(); + Message redelivered = jmsConsumer.receive(5000); + assertNotNull("Message should be redelivered after session.recover()", redelivered); + assertTrue("Redelivered flag should be set", redelivered.getJMSRedelivered()); + } + } + } + + // ---- TRANSACTED: message treated as delivered within transaction ---- + + @Test + public void testReceiveBodyTransactedMessageTreatedAsDelivered() { + try (JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.SESSION_TRANSACTED)) { + Destination destination = jmsContext.createQueue(methodNameDestinationName); + jmsContext.createProducer().send(destination, new byte[] {1, 2, 3}); + jmsContext.commit(); + + try (JMSConsumer jmsConsumer = jmsContext.createConsumer(destination)) { + jmsContext.start(); + + // receiveBody with wrong type should throw + try { + jmsConsumer.receiveBody(String.class, 5000); + fail("Should have thrown MessageFormatRuntimeException"); + } catch (MessageFormatRuntimeException e) { + // expected + } + + // TRANSACTED: message is treated as delivered, not re-enqueued. + // A subsequent receive within the same transaction should not return it. + Message msg = jmsConsumer.receiveNoWait(); + assertNull("Message should not be re-enqueued in TRANSACTED mode", msg); + } + } + } + + @Test + public void testReceiveBodyTransactedRollbackRedelivers() throws JMSException { + try (JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.SESSION_TRANSACTED)) { + Destination destination = jmsContext.createQueue(methodNameDestinationName); + jmsContext.createProducer().send(destination, new byte[] {1, 2, 3}); + jmsContext.commit(); + + try (JMSConsumer jmsConsumer = jmsContext.createConsumer(destination)) { + jmsContext.start(); + + try { + jmsConsumer.receiveBody(String.class, 5000); + fail("Should have thrown MessageFormatRuntimeException"); + } catch (MessageFormatRuntimeException e) { + // expected + } + + // After rollback(), the message should be redelivered + jmsContext.rollback(); + Message redelivered = jmsConsumer.receive(5000); + assertNotNull("Message should be redelivered after transaction rollback", redelivered); + assertTrue("Redelivered flag should be set", redelivered.getJMSRedelivered()); + } + } + } + + // ---- Credit window: consumer must not get stuck after failed receiveBody ---- + + @Test + public void testReceiveBodyClientAckCreditWindowNotExhausted() { + try (JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.CLIENT_ACKNOWLEDGE)) { + Destination destination = jmsContext.createQueue(methodNameDestinationName); + // Send two messages: first will fail receiveBody, second should still be receivable + jmsContext.createProducer().send(destination, new byte[] {1, 2, 3}); + jmsContext.createProducer().send(destination, "hello"); + + try (JMSConsumer jmsConsumer = jmsContext.createConsumer(destination)) { + jmsContext.start(); + + // First message: type mismatch + try { + jmsConsumer.receiveBody(String.class, 5000); + fail("Should have thrown MessageFormatRuntimeException"); + } catch (MessageFormatRuntimeException e) { + // expected + } + + // Second message should still be consumable (credit window expanded) + String body = jmsConsumer.receiveBody(String.class, 5000); + assertEquals("hello", body); + } + } + } + + @Test + public void testReceiveBodyTransactedCreditWindowNotExhausted() { + try (JMSContext jmsContext = activemqConnectionFactory.createContext(DEFAULT_JMS_USER, DEFAULT_JMS_PASS, Session.SESSION_TRANSACTED)) { + Destination destination = jmsContext.createQueue(methodNameDestinationName); + // Send two messages: first will fail receiveBody, second should still be receivable + jmsContext.createProducer().send(destination, new byte[] {1, 2, 3}); + jmsContext.createProducer().send(destination, "hello"); + jmsContext.commit(); + + try (JMSConsumer jmsConsumer = jmsContext.createConsumer(destination)) { + jmsContext.start(); + + // First message: type mismatch + try { + jmsConsumer.receiveBody(String.class, 5000); + fail("Should have thrown MessageFormatRuntimeException"); + } catch (MessageFormatRuntimeException e) { + // expected + } + + // Second message should still be consumable (credit window expanded) + String body = jmsConsumer.receiveBody(String.class, 5000); + assertEquals("hello", body); + } + } + } +}