You are viewing a plain text version of this content. The canonical link for it is here.
Posted to gitbox@activemq.apache.org by "jbonofre (via GitHub)" <gi...@apache.org> on 2025/11/19 20:27:49 UTC
[PR] AMQ-8464: Implement receiveBody methods in the Consumer [activemq]
jbonofre opened a new pull request, #1543:
URL: https://github.com/apache/activemq/pull/1543
(no comment)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "jeanouii (via GitHub)" <gi...@apache.org>.
jeanouii commented on PR #1543:
URL: https://github.com/apache/activemq/pull/1543#issuecomment-4053934941
LGTM
Type validation prior to consumption - enqueue first if not good. Better approach than consuming and catching exception because then you would need remove from deliveredMessages when AUTO_ACK before doing the enqueue first.
TRANSACTED/CLIENT_ACK call to before and after to manage the credit window. Avoid blocking the next receive.
Test coverage looks good with edge cases for credit window expansion, transacted rollback, auto_ack redelivery, plain message.
Good job @jbonofre
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] AMQ-8464: Implement receiveBody methods in the Consumer [activemq]
Posted by "jbonofre (via GitHub)" <gi...@apache.org>.
jbonofre commented on PR #1543:
URL: https://github.com/apache/activemq/pull/1543#issuecomment-3912638699
I updated this PR with a new implementation approach.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "jbonofre (via GitHub)" <gi...@apache.org>.
jbonofre commented on PR #1543:
URL: https://github.com/apache/activemq/pull/1543#issuecomment-4016963394
@tabish121 @cshannon @jeanouii I reworked this PR and added specific unit tests for `receiveBody()` methods.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "jbonofre (via GitHub)" <gi...@apache.org>.
jbonofre commented on code in PR #1543:
URL: https://github.com/apache/activemq/pull/1543#discussion_r2996107326
##########
activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java:
##########
@@ -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)) {
Review Comment:
I think I did that in `isBodyAssignableTo()` method. Let me check (I did different changes).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "cshannon (via GitHub)" <gi...@apache.org>.
cshannon commented on PR #1543:
URL: https://github.com/apache/activemq/pull/1543#issuecomment-5768607998
@jbonofre - This looks pretty good to me now, can you bring it up to date?
@mattrpav - Did you take a peak? It looks like the spec issues were worked out now but it would be good to get you to quickly look too
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "cshannon (via GitHub)" <gi...@apache.org>.
cshannon commented on code in PR #1543:
URL: https://github.com/apache/activemq/pull/1543#discussion_r2996115073
##########
activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java:
##########
@@ -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);
Review Comment:
@jbonofre - Did you have changes that got blown away with a force push? I thought at one point there was an update to handle the fact that for transactions and individual ack we should not be re-adding first? I'm confused as there has been a lot of updates and force pushes but this PR still isnt' handling that part of the spec and there is also the comment @tabish121 just made a few minutes ago about what happens if the type is Message or StreamMessage.
At some point it might be easier to close this and open a new PR as it's getting hard to follow everything with all the comments
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] AMQ-8464: Implement receiveBody methods in the Consumer [activemq]
Posted by "tabish121 (via GitHub)" <gi...@apache.org>.
tabish121 commented on code in PR #1543:
URL: https://github.com/apache/activemq/pull/1543#discussion_r2543503024
##########
activemq-client/src/main/java/org/apache/activemq/ActiveMQConsumer.java:
##########
@@ -100,17 +100,44 @@ public void close() {
@Override
public <T> T receiveBody(Class<T> c) {
- throw new UnsupportedOperationException("receiveBody(Class<T>) is not supported");
+ Message message = receive();
+ if (message == null) {
+ return null;
+ }
+ try {
+ Object body = message.getBody(c);
+ return c.cast(body);
+ } catch (JMSException e) {
Review Comment:
This would violate the JMS specification in that it does not return the message to the head of the prefetch queue to be read again on the next receive call for the AUTO and DUPS_OK ack modes. Same for the other methods below.
> The result of this method throwing a MessageFormatRuntimeException depends on the session mode:
>
> AUTO_ACKNOWLEDGE or DUPS_OK_ACKNOWLEDGE: The JMS provider will behave as if the unsuccessful call to receiveBody had not occurred. The message will be delivered again before any subsequent messages.
>
> This is not considered to be redelivery and does not cause the JMSRedelivered message header field to be set or the JMSXDeliveryCount message property to be incremented.
>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "jbonofre (via GitHub)" <gi...@apache.org>.
jbonofre commented on code in PR #1543:
URL: https://github.com/apache/activemq/pull/1543#discussion_r3657343959
##########
activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java:
##########
@@ -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)) {
Review Comment:
Confirmed — the fix that got lost in an earlier force-push is back in the current commit. `doReceiveBody()` now checks the message type explicitly before trusting `isBodyAssignableTo()`:
```java
boolean bodyNotAssignable = message.getClass() == ActiveMQMessage.class // plain Message: always reject
|| message instanceof ActiveMQStreamMessage // StreamMessage: always reject
|| !message.isBodyAssignableTo(c);
```
This resolves the conflict you pointed out where `Message#isBodyAssignableTo()` must return `true` for any type while `receiveBody()` must reject `Message`/`StreamMessage`. Covered by `testReceiveBodyPlainMessageThrows` and `testReceiveBodyStreamMessageThrows` (23 tests total, all passing).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] AMQ-8464: Implement receiveBody methods in the Consumer [activemq]
Posted by "jeanouii (via GitHub)" <gi...@apache.org>.
jeanouii commented on PR #1543:
URL: https://github.com/apache/activemq/pull/1543#issuecomment-3848399233
As @tabish121 pointed out, the current implementation violates the spec I think.
I think the best way to implement this is by creating a ActiveMQMessageConsumer#receiveBody that ActiveMQConsumer can delegate to.
Currently, ActiveMQConsumer first delegates to ActiveMQMessageConsumer that will ack the message before reading the body. But we need to read first the body for receiveBody.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "jbonofre (via GitHub)" <gi...@apache.org>.
jbonofre commented on code in PR #1543:
URL: https://github.com/apache/activemq/pull/1543#discussion_r2914233503
##########
activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java:
##########
@@ -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)) {
Review Comment:
@tabish121 @cshannon So, I did a new read on the spec.
Here's my take/understanding.
1. If the ack mode is AUTO or DUPS_OK, the spec says "behave as if the unsuccessful call has not occurred. The message will be delivered again before any subsequent messages. This is not considered to be redelivery and does not cause JMSRedelivered to be set or JMSXDeliveryCount to be incremented." My understanding here is that we should use `unconsumedMessages.enqueueFirst(md)`, meaning that we put back to front of local prefetch, no broker interaction, no header changes.
2. If the ack mode is CLIENT, the spec says "behave as if the call has been successful and will not deliver the message again. The message will not be acknowledged until acknowledge is called. If an application wishes redelivery, it must call recover." My understanding here is that we should use `beforeMessageIsConsumed(md)` and `afterMessageIsConsumed(md, false)`, meaning tracked as delivered with credit window expansion.
3. if Transacted, the spec says "behave as if the call had been successful and will not deliver the message again. The transaction will remain uncommitted. If an application wishes redelivery, it mull roll back the transaction." Here, my understand is that we should use the same approach as CLIENT ack mode (delivered within transaction).
Also, I see in the spec that the message type has an "impact".
If the message doesn't have body, we throw the exception.
The spec says "may be used to receive any type of message except for StreamMessage", meaning that `ActiveMQStreamMessage.isBodyAssignableTo()` always return `false`.
The spec says "may be used to receive any type of message except for... Message", meaning we need an explicit check for `ACTIVEMQ_MESSAGE` data structure type.
Am I right ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "cshannon (via GitHub)" <gi...@apache.org>.
cshannon commented on PR #1543:
URL: https://github.com/apache/activemq/pull/1543#issuecomment-4137765160
> If you want me to open a clean new PR, just let me know.
@jbonofre - On second thought it's probably better to just keep this PR open to preserve the comment history, so I would just keep this one open.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "jbonofre (via GitHub)" <gi...@apache.org>.
jbonofre commented on PR #1543:
URL: https://github.com/apache/activemq/pull/1543#issuecomment-5041801338
@cshannon I didn't define target project on this one as it's best effort. I propose to move forward for now and define the target project (hopefully 6.4.0) when we have a consensus on the implementation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] AMQ-8464: Implement receiveBody methods in the Consumer [activemq]
Posted by "jbonofre (via GitHub)" <gi...@apache.org>.
jbonofre commented on code in PR #1543:
URL: https://github.com/apache/activemq/pull/1543#discussion_r2543563507
##########
activemq-client/src/main/java/org/apache/activemq/ActiveMQConsumer.java:
##########
@@ -100,17 +100,44 @@ public void close() {
@Override
public <T> T receiveBody(Class<T> c) {
- throw new UnsupportedOperationException("receiveBody(Class<T>) is not supported");
+ Message message = receive();
+ if (message == null) {
+ return null;
+ }
+ try {
+ Object body = message.getBody(c);
+ return c.cast(body);
+ } catch (JMSException e) {
Review Comment:
@tabish121 yes, for sure. It's a first draft.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "tabish121 (via GitHub)" <gi...@apache.org>.
tabish121 commented on code in PR #1543:
URL: https://github.com/apache/activemq/pull/1543#discussion_r2914341492
##########
activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java:
##########
@@ -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)) {
Review Comment:
You should implement isBodyAssignableTo in each message type to follow the guidance in the API docs.
```
boolean isBodyAssignableTo([Class](https://docs.oracle.com/javase/8/docs/api/java/lang/Class.html?is-external=true) c)
throws [JMSException](https://jakarta.ee/specifications/messaging/2.0/apidocs/javax/jms/jmsexception)
Returns whether the message body is capable of being assigned to the specified type. If this method returns true then a subsequent call to the method getBody on the same message with the same type argument would not throw a MessageFormatException.
If the message is a StreamMessage then false is always returned. If the message is a ObjectMessage and object deserialization fails then false is returned. If the message has no body then any type may be specified and true is returned.
Parameters:
c - The specified type
If the message is a TextMessage then this method will only return true if this parameter is set to String.class or another type to which a String is assignable.
If the message is a ObjectMessage then this method will only return true if this parameter is set to java.io.Serializable.class or another class to which the body is assignable.
If the message is a MapMessage then this method will only return true if this parameter is set to java.util.Map.class (or java.lang.Object.class).
If the message is a BytesMessage then this this method will only return true if this parameter is set to byte[].class (or java.lang.Object.class).
If the message is a TextMessage, ObjectMessage, MapMessage or BytesMessage and the message has no body, then the above does not apply and this method will return true irrespective of the value of this parameter.
If the message is a Message (but not one of its subtypes) then this method will return true irrespective of the value of this parameter.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "cshannon (via GitHub)" <gi...@apache.org>.
cshannon commented on PR #1543:
URL: https://github.com/apache/activemq/pull/1543#issuecomment-5376269165
@jbonofre - Can you do another rebase and bring this up to date again? I should have time soon to get back to this, it would be good to get into 6.4.0
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "cshannon (via GitHub)" <gi...@apache.org>.
cshannon commented on code in PR #1543:
URL: https://github.com/apache/activemq/pull/1543#discussion_r4073502152
##########
activemq-client/src/main/java/org/apache/activemq/command/ActiveMQObjectMessage.java:
##########
@@ -290,8 +290,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) {
Review Comment:
There is no need to catch JMSRuntimeException here as getObject() method will only throw JMSException. That exception is only used in ActiveMQConsumer so I don't really think we need a comment
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] AMQ-8464: Implement receiveBody methods in the Consumer [activemq]
Posted by "jbonofre (via GitHub)" <gi...@apache.org>.
jbonofre commented on code in PR #1543:
URL: https://github.com/apache/activemq/pull/1543#discussion_r2543563507
##########
activemq-client/src/main/java/org/apache/activemq/ActiveMQConsumer.java:
##########
@@ -100,17 +100,44 @@ public void close() {
@Override
public <T> T receiveBody(Class<T> c) {
- throw new UnsupportedOperationException("receiveBody(Class<T>) is not supported");
+ Message message = receive();
+ if (message == null) {
+ return null;
+ }
+ try {
+ Object body = message.getBody(c);
+ return c.cast(body);
+ } catch (JMSException e) {
Review Comment:
@tabish121 yes, for sure. It's a first draft. Gonna work on this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] AMQ-8464: Implement receiveBody methods in the Consumer [activemq]
Posted by "jbonofre (via GitHub)" <gi...@apache.org>.
jbonofre commented on code in PR #1543:
URL: https://github.com/apache/activemq/pull/1543#discussion_r2658787087
##########
activemq-client/src/main/java/org/apache/activemq/ActiveMQConsumer.java:
##########
@@ -100,17 +100,44 @@ public void close() {
@Override
public <T> T receiveBody(Class<T> c) {
- throw new UnsupportedOperationException("receiveBody(Class<T>) is not supported");
+ Message message = receive();
+ if (message == null) {
+ return null;
+ }
+ try {
+ Object body = message.getBody(c);
+ return c.cast(body);
+ } catch (JMSException e) {
Review Comment:
@tabish121 I started the implementation of the body receive failure logic.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "jbonofre (via GitHub)" <gi...@apache.org>.
jbonofre commented on PR #1543:
URL: https://github.com/apache/activemq/pull/1543#issuecomment-4137721565
@tabish121 @cshannon I was not able to find the previous correct commit. So I implemented again with what I remember I did.
If you want me to open a clean new PR, just let me know.
Thanks !
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "cshannon (via GitHub)" <gi...@apache.org>.
cshannon commented on PR #1543:
URL: https://github.com/apache/activemq/pull/1543#issuecomment-5038346588
i am going to move this to 6.4.0
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] AMQ-8464: Implement receiveBody methods in the Consumer [activemq]
Posted by "jbonofre (via GitHub)" <gi...@apache.org>.
jbonofre commented on PR #1543:
URL: https://github.com/apache/activemq/pull/1543#issuecomment-3914878810
@cshannon yup, it's what I realized while working on the PR. I will update.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "jbonofre (via GitHub)" <gi...@apache.org>.
jbonofre commented on code in PR #1543:
URL: https://github.com/apache/activemq/pull/1543#discussion_r2917469972
##########
activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java:
##########
@@ -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)) {
Review Comment:
OK, I added a try/catch on `getObject()` to return `false` in `isBodyAssignableTo()` to be compliant with the spec.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "jbonofre (via GitHub)" <gi...@apache.org>.
jbonofre commented on PR #1543:
URL: https://github.com/apache/activemq/pull/1543#issuecomment-5795023097
@cshannon @mattrpav thanks guys. I rebased the PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "mattrpav (via GitHub)" <gi...@apache.org>.
mattrpav commented on code in PR #1543:
URL: https://github.com/apache/activemq/pull/1543#discussion_r4073359306
##########
activemq-client/src/main/java/org/apache/activemq/command/ActiveMQObjectMessage.java:
##########
@@ -290,8 +290,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) {
Review Comment:
Do we alos need to catch JMSRuntimeException here?
##########
activemq-client/src/main/java/org/apache/activemq/command/ActiveMQObjectMessage.java:
##########
@@ -290,8 +290,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) {
Review Comment:
Do we also need to catch JMSRuntimeException here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "jbonofre (via GitHub)" <gi...@apache.org>.
jbonofre commented on code in PR #1543:
URL: https://github.com/apache/activemq/pull/1543#discussion_r3657346653
##########
activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java:
##########
@@ -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);
Review Comment:
You were right — a force-push had reverted this. It is restored in the current commit. On a body-type mismatch `doReceiveBody()` now branches on the ack mode:
```java
if (session.isAutoAcknowledge() || session.isDupsOkAcknowledge()) {
// AUTO / DUPS_OK: put back at the head of the prefetch, no ack
unconsumedMessages.enqueueFirst(md);
} else {
// CLIENT / TRANSACTED / INDIVIDUAL: treated as delivered, NOT re-enqueued.
// beforeMessageIsConsumed + afterMessageIsConsumed send a DELIVERED_ACK,
// which expands the credit window so the consumer does not stall.
beforeMessageIsConsumed(md);
afterMessageIsConsumed(md, false);
}
throw new MessageFormatException(...);
```
So transacted, client and individual-ack modes no longer re-add to the head — the application recovers via `session.recover()` / `session.rollback()`. Both the credit-window-not-exhausted concern and the `Message`/`StreamMessage` type check are now covered by tests (`testReceiveBodyClientAckCreditWindowNotExhausted`, `testReceiveBodyTransactedCreditWindowNotExhausted`, `testReceiveBodyClientAckRecoverRedelivers`, `testReceiveBodyTransactedRollbackRedelivers`, plus the plain-Message/StreamMessage tests). Rebased on latest `main`, 23/23 tests green.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] AMQ-8464: Implement receiveBody methods in the Consumer [activemq]
Posted by "jbonofre (via GitHub)" <gi...@apache.org>.
jbonofre commented on PR #1543:
URL: https://github.com/apache/activemq/pull/1543#issuecomment-3848411311
@jeanouii thanks for the review. I have to resume my work on this PR. I agree that the approach you are proposing makes more sense (else the ack will be send too early).
I will update accordingly.
@tabish121 @jeanouii thanks for your inputs ! I will resume my work on this one.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "tabish121 (via GitHub)" <gi...@apache.org>.
tabish121 commented on code in PR #1543:
URL: https://github.com/apache/activemq/pull/1543#discussion_r2921217326
##########
activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java:
##########
@@ -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)) {
Review Comment:
You are correct about the base Message type needing to be accounted for. It is the only case where you have an explicitly defined Message that has "no body" whereas the others might arrive without a body but a null body is itself a body in those cases so the method can return null.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "jbonofre (via GitHub)" <gi...@apache.org>.
jbonofre commented on PR #1543:
URL: https://github.com/apache/activemq/pull/1543#issuecomment-4016962001
@cshannon I updated this PR by using the existing `ActiveMQConsumer` (which implements `JMSConsumer`) and wrapping the `receiveBody()` methods from `ActiveMQMessageConsumer`.
Thanks to that we have a similar approach, not impacting the exception handling.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #1543:
URL: https://github.com/apache/activemq/pull/1543#issuecomment-4539159492
This PR is stale because it has been open 60 days with no activity. Remove stale label or comment or this will be closed in 5 days.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "jbonofre (via GitHub)" <gi...@apache.org>.
jbonofre commented on PR #1543:
URL: https://github.com/apache/activemq/pull/1543#issuecomment-5091803034
@cshannon @tabish121 I rebased this PR and did again the expected change (that have been changed by a force push 😄 ). If you can do a new pass, that would be great.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "jbonofre (via GitHub)" <gi...@apache.org>.
jbonofre commented on code in PR #1543:
URL: https://github.com/apache/activemq/pull/1543#discussion_r2900094039
##########
activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java:
##########
@@ -580,20 +594,93 @@ private boolean redeliveryExceeded(MessageDispatch md) {
* this message consumer is concurrently closed
*/
@Override
- public Message receive() throws JMSException {
- checkClosed();
- checkMessageListener();
-
- sendPullCommand(0);
- MessageDispatch md = dequeue(-1);
- if (md == null) {
- return null;
+ public Message receive() {
+ try {
+ checkClosed();
+ checkMessageListener();
+
+ sendPullCommand(0);
+ MessageDispatch md = dequeue(-1);
+ if (md == null) {
+ return null;
+ }
+
+ beforeMessageIsConsumed(md);
+ afterMessageIsConsumed(md, false);
+
+ return createActiveMQMessage(md);
+ } catch (JMSException e) {
+ throw JMSExceptionSupport.convertToJMSRuntimeException(e);
}
+ }
- beforeMessageIsConsumed(md);
- afterMessageIsConsumed(md, false);
+ /**
+ * Receives the next message and returns its body converted to the requested type.
+ * <p>
+ * This call follows the same synchronous consume flow as {@link #receive()}:
+ * pull/dequeue, mark delivered, then acknowledge bookkeeping after successful
+ * conversion. Conversion failures are delegated to
+ * {@link #handleReceiveBodyFailure(MessageDispatch, MessageFormatException)}
+ * before rethrowing the {@link MessageFormatException}.
+ *
+ * @param bodyType
+ * the expected body type.
+ *
+ * @return the converted message body, or {@code null} if no message is available.
+ */
+ @Override
+ public <T> T receiveBody(Class<T> bodyType) {
+ try {
+ checkClosed();
+ checkMessageListener();
- return createActiveMQMessage(md);
+ sendPullCommand(0);
+ MessageDispatch md = dequeue(-1);
+ if (md == null) {
+ return null;
+ }
+
+ beforeMessageIsConsumed(md);
Review Comment:
Thanks @tabish121 . I have updated the PR to use the approach you suggested (with `beforeMessaageIsConsumed(md)`).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "jbonofre (via GitHub)" <gi...@apache.org>.
jbonofre commented on code in PR #1543:
URL: https://github.com/apache/activemq/pull/1543#discussion_r2914008262
##########
activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java:
##########
@@ -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)) {
Review Comment:
Thanks @tabish121 , I forgot this part in my latest change. I was focusing too much on `beforeMessageConsumed()` and `afterMessageConsumed()` methods. My bad.
Let me update the `doReceiveBody()` to behave correctly depending of the ack mode.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "jbonofre (via GitHub)" <gi...@apache.org>.
jbonofre commented on code in PR #1543:
URL: https://github.com/apache/activemq/pull/1543#discussion_r2996141116
##########
activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java:
##########
@@ -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);
Review Comment:
I'm sure I did that, so maybe a force push and different local changes overwrite my previous changes.
Let me double check.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "tabish121 (via GitHub)" <gi...@apache.org>.
tabish121 commented on code in PR #1543:
URL: https://github.com/apache/activemq/pull/1543#discussion_r2996148171
##########
activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java:
##########
@@ -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)) {
Review Comment:
That's what I'm pointing out though, if you read the API docs for `isBodyAssignableTo()` and the docs for `receiveBody()` they conflict because the Message API require it say yes to everything where the receive body method must say no to everything.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "skrcode (via GitHub)" <gi...@apache.org>.
skrcode commented on PR #1543:
URL: https://github.com/apache/activemq/pull/1543#issuecomment-5381278854
JAIPilot Cloud reviewed exact head `5ad7be8834181a68651f270231d71f0e43e2f86a` and found one focused test gap in the changed `ActiveMQObjectMessage` contract.
I opened an AI-assisted, test-only draft against this branch: https://github.com/jbonofre/activemq/pull/1
It verifies that an untrusted serialized payload makes `isBodyAssignableTo()` return `false` and `getBody()` throw `MessageFormatException`, instead of leaking the raw deserialization `JMSException`. The focused JMS 2 suites passed 41 tests and the two affected modules built/compiled cleanly. The draft states the scoped-build limitation and includes explicit model provenance.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "jbonofre (via GitHub)" <gi...@apache.org>.
jbonofre merged PR #1543:
URL: https://github.com/apache/activemq/pull/1543
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "tabish121 (via GitHub)" <gi...@apache.org>.
tabish121 commented on code in PR #1543:
URL: https://github.com/apache/activemq/pull/1543#discussion_r2912301851
##########
activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java:
##########
@@ -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)) {
Review Comment:
I think I've pointed this out already a couple times but this still violates the specification. Please refer to section 8.6 on pages 84 and 85 of the [Jakarta Messaging](https://jakarta.ee/specifications/messaging/3.1/jakarta-messaging-spec-3.1.pdf) 3.1 specification for how the client should handle this in terms of the ACK mode of the consumer.
The basic concept is as follows:
1. If the client acknowledgement mode is AUTO or DUPS_OK then you are meant to put the message back on the prefetch queue and allow the caller to read it again as they have no recourse to recover those messages once you throw here.
2. If in CLIENT acknowledgement mode the message **should not** immediately go back to the prefetch buffer as the application can either call `session.recover()` to redeliver all unacknowledged messages or can read the next message and call `message.acknowledge()` to acknowledge that message and all previous messages.
3. If in TRANSACTED mode the message **should not** go back into the prefetch buffer because the application can either call `session.rollback()` to recover messages that failed or cal simply continue consuming messages and call `session.commit()` to retire this and any other failed or consumed messages.
Testing here should account for these cases and also ensure that in CLIENT and TRANSACTED modes that the consumer credit window is expanded when message fail here to avoid a stuck consumer receive due to the current credit window having been exhausted.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "tabish121 (via GitHub)" <gi...@apache.org>.
tabish121 commented on code in PR #1543:
URL: https://github.com/apache/activemq/pull/1543#discussion_r2996055834
##########
activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java:
##########
@@ -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)) {
Review Comment:
I think this was right at one point but now looks wrong, the recevieBody method should throw always for any message of type Message or StreamMessage and in the case of Message that conflicts with what the spec says is required for Message#isBodyAssignableTo so you absolutely need to check it and for good measure should just also check if its a StreamMessage and throw early to make it explicit.
Its a bit annoying that the spec conflicts here but it does and you do have to check the message type in this specific case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "jbonofre (via GitHub)" <gi...@apache.org>.
jbonofre commented on code in PR #1543:
URL: https://github.com/apache/activemq/pull/1543#discussion_r2996226499
##########
activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java:
##########
@@ -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)) {
Review Comment:
Yes, but I'm sure I fixed that in the past. I suspect I overwrite my changes with a force-push.
I'm checking.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] AMQ-8464: Implement receiveBody methods in the Consumer [activemq]
Posted by "tabish121 (via GitHub)" <gi...@apache.org>.
tabish121 commented on code in PR #1543:
URL: https://github.com/apache/activemq/pull/1543#discussion_r2818095361
##########
activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java:
##########
@@ -580,20 +594,93 @@ private boolean redeliveryExceeded(MessageDispatch md) {
* this message consumer is concurrently closed
*/
@Override
- public Message receive() throws JMSException {
- checkClosed();
- checkMessageListener();
-
- sendPullCommand(0);
- MessageDispatch md = dequeue(-1);
- if (md == null) {
- return null;
+ public Message receive() {
+ try {
+ checkClosed();
+ checkMessageListener();
+
+ sendPullCommand(0);
+ MessageDispatch md = dequeue(-1);
+ if (md == null) {
+ return null;
+ }
+
+ beforeMessageIsConsumed(md);
+ afterMessageIsConsumed(md, false);
+
+ return createActiveMQMessage(md);
+ } catch (JMSException e) {
+ throw JMSExceptionSupport.convertToJMSRuntimeException(e);
}
+ }
- beforeMessageIsConsumed(md);
- afterMessageIsConsumed(md, false);
+ /**
+ * Receives the next message and returns its body converted to the requested type.
+ * <p>
+ * This call follows the same synchronous consume flow as {@link #receive()}:
+ * pull/dequeue, mark delivered, then acknowledge bookkeeping after successful
+ * conversion. Conversion failures are delegated to
+ * {@link #handleReceiveBodyFailure(MessageDispatch, MessageFormatException)}
+ * before rethrowing the {@link MessageFormatException}.
+ *
+ * @param bodyType
+ * the expected body type.
+ *
+ * @return the converted message body, or {@code null} if no message is available.
+ */
+ @Override
+ public <T> T receiveBody(Class<T> bodyType) {
+ try {
+ checkClosed();
+ checkMessageListener();
- return createActiveMQMessage(md);
+ sendPullCommand(0);
+ MessageDispatch md = dequeue(-1);
+ if (md == null) {
+ return null;
+ }
+
+ beforeMessageIsConsumed(md);
Review Comment:
I think you could avoid a lot of complexity here by deferring `beforeMessageIsConsumed(md);` until after you've actually checked that you want to consume the message and if you can't / won't because the body is not assignable to the target you just enqueue the message back into the head of the unconsumed list as you haven't consumed it. That way you don't need to perform the various rollback work since you won't have done anything that triggers actual tracking work in the consumer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #1543: feat(#1716): Implement receiveBody()
URL: https://github.com/apache/activemq/pull/1543
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "jbonofre (via GitHub)" <gi...@apache.org>.
jbonofre commented on PR #1543:
URL: https://github.com/apache/activemq/pull/1543#issuecomment-4991395208
Reopen for new round.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] feat(#1716): Implement receiveBody() [activemq]
Posted by "mattrpav (via GitHub)" <gi...@apache.org>.
mattrpav commented on code in PR #1543:
URL: https://github.com/apache/activemq/pull/1543#discussion_r4073359306
##########
activemq-client/src/main/java/org/apache/activemq/command/ActiveMQObjectMessage.java:
##########
@@ -290,8 +290,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) {
Review Comment:
Do we also need to catch JMSRuntimeException here? Perhaps maybe a comment that if the exception types get refactored to be JMSRuntimeException-first this is a spot that needs special handling.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact