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); + } + } + } +}