This is an automated email from the ASF dual-hosted git repository.
jbonofre pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/activemq.git
The following commit(s) were added to refs/heads/main by this push:
new ad9a30dfa0 feat(#1797): add message delivery tracking to
QueueMessageReference (#1796)
ad9a30dfa0 is described below
commit ad9a30dfa0835c1712b794d0ea00f16cd79981fa
Author: Jean-Louis Monteiro <[email protected]>
AuthorDate: Thu Jul 9 16:37:59 2026 +0200
feat(#1797): add message delivery tracking to QueueMessageReference (#1796)
* feat: add message delivery tracking to QueueMessageReference
Add isDelivered()/setDelivered() to QueueMessageReference interface
with implementations in IndirectMessageReference (boolean field) and
NullMessageReference (no-op). PrefetchSubscription.acknowledge() now
marks messages as delivered=true when processing DeliveredAck, enabling
downstream consumers to distinguish delivered-but-unacked messages.
* fix: incorporate feedback
---
.../activemq/broker/region/IndirectMessageReference.java | 16 ++++++++++++++--
.../activemq/broker/region/NullMessageReference.java | 9 +++++++++
.../activemq/broker/region/PrefetchSubscription.java | 16 +++++++++-------
.../activemq/broker/region/QueueMessageReference.java | 4 ++++
4 files changed, 36 insertions(+), 9 deletions(-)
diff --git
a/activemq-broker/src/main/java/org/apache/activemq/broker/region/IndirectMessageReference.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/IndirectMessageReference.java
index 104dcb0a4a..047c34f860 100644
---
a/activemq-broker/src/main/java/org/apache/activemq/broker/region/IndirectMessageReference.java
+++
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/IndirectMessageReference.java
@@ -35,6 +35,8 @@ public class IndirectMessageReference implements
QueueMessageReference {
private boolean dropped;
/** Has the message been acked? */
private boolean acked;
+ /** Has the message been delivered? */
+ private boolean delivered;
/** Direct reference to the message */
private final Message message;
private final MessageId messageId;
@@ -197,16 +199,26 @@ public class IndirectMessageReference implements
QueueMessageReference {
@Override
public synchronized int getSize() {
- return message.getSize();
+ return message.getSize();
}
@Override
public boolean isAdvisory() {
- return message.isAdvisory();
+ return message.isAdvisory();
}
@Override
public boolean canProcessAsExpired() {
return message.canProcessAsExpired();
}
+
+ @Override
+ public synchronized boolean isDelivered() {
+ return delivered;
+ }
+
+ @Override
+ public synchronized void setDelivered(final boolean delivered) {
+ this.delivered = delivered;
+ }
}
diff --git
a/activemq-broker/src/main/java/org/apache/activemq/broker/region/NullMessageReference.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/NullMessageReference.java
index e8d26a84fc..8d2edf8895 100644
---
a/activemq-broker/src/main/java/org/apache/activemq/broker/region/NullMessageReference.java
+++
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/NullMessageReference.java
@@ -159,4 +159,13 @@ public final class NullMessageReference implements
QueueMessageReference {
return false;
}
+ @Override
+ public boolean isDelivered() {
+ return false;
+ }
+
+ @Override
+ public void setDelivered(boolean delivered) {
+
+ }
}
diff --git
a/activemq-broker/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
index b2adf56a67..2fa6203bc5 100644
---
a/activemq-broker/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
+++
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
@@ -270,13 +270,15 @@ public abstract class PrefetchSubscription extends
AbstractSubscription {
} else if (ack.isDeliveredAck()) {
// Message was delivered but not acknowledged: update pre-fetch
// counters.
- int index = 0;
- for (Iterator<MessageReference> iter = dispatched.iterator();
iter.hasNext(); index++) {
- final MessageReference node = iter.next();
- Destination nodeDest = (Destination)
node.getRegionDestination();
- if (ack.getLastMessageId().equals(node.getMessageId())) {
+
+ for (final MessageReference node : dispatched) {
+ final MessageId messageId = node.getMessageId();
+ if (node instanceof QueueMessageReference) {
+ ((QueueMessageReference) node).setDelivered(true);
+ }
+ if (ack.getLastMessageId().equals(messageId)) {
expandPrefetchExtension(ack.getMessageCount());
- destination = nodeDest;
+ destination = (Destination)
node.getRegionDestination();
callDispatchMatched = true;
break;
}
@@ -396,7 +398,7 @@ public abstract class PrefetchSubscription extends
AbstractSubscription {
}
protected void processExpiredAck(final ConnectionContext context, final
Destination dest,
- final MessageReference node) {
+ final MessageReference node) {
dest.messageExpired(context, this, node);
}
diff --git
a/activemq-broker/src/main/java/org/apache/activemq/broker/region/QueueMessageReference.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/QueueMessageReference.java
index 0f9bf47150..e7aa06add4 100644
---
a/activemq-broker/src/main/java/org/apache/activemq/broker/region/QueueMessageReference.java
+++
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/QueueMessageReference.java
@@ -42,4 +42,8 @@ public interface QueueMessageReference extends
MessageReference {
boolean unlock();
LockOwner getLockOwner();
+
+ boolean isDelivered();
+
+ void setDelivered(boolean delivered);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
For further information, visit: https://activemq.apache.org/contact
