feat(#1797): add message delivery tracking to QueueMessageReference

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 acked? */
+    private boolean delivered;
     /** Direct reference to the message */
     private final Message message;
     private final MessageId messageId;
@@ -197,16 +199,25 @@ public boolean isExpired() {
 
     @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 boolean isDelivered() {
+        return delivered;
+    }
+
+    public void setDelivered(final boolean delivered) {
+        this.delivered = delivered;
+    }
 }

activemq-broker/src/main/java/org/apache/activemq/broker/region/NullMessageReference.java

@@ -159,4 +159,13 @@ public boolean canProcessAsExpired() {
         return false;
     }
 
+    @Override
+    public boolean isDelivered() {
+        return false;
+    }
+
+    @Override
+    public void setDelivered(boolean delivered) {
+
+    }
 }

activemq-broker/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java

@@ -270,6 +270,27 @@ public final void acknowledge(final ConnectionContext context,final MessageAck a
             } else if (ack.isDeliveredAck()) {
                 // Message was delivered but not acknowledged: update pre-fetch
                 // counters.
+
+                boolean inAckRange = false;
+                for (final MessageReference node : dispatched) {
+                    MessageId messageId = node.getMessageId();
+                    if (ack.getFirstMessageId() == null
+                            || ack.getFirstMessageId().equals(messageId)) {
+                        inAckRange = true;
+                    }
+                    if (inAckRange) {
+                        if (node instanceof QueueMessageReference) {
+                            ((QueueMessageReference) node).setDelivered(true);
+                        }
+
+                        if (ack.getLastMessageId().equals(messageId)) {
+                            destination = (Destination) node.getRegionDestination();
+                            callDispatchMatched = true;
+                            break;
+                        }
+                    }
+                }
+
                 int index = 0;
                 for (Iterator<MessageReference> iter = dispatched.iterator(); iter.hasNext(); index++) {
                     final MessageReference node = iter.next();
@@ -396,7 +417,7 @@ public final void acknowledge(final ConnectionContext context,final MessageAck a
     }
 
     protected void processExpiredAck(final ConnectionContext context, final Destination dest,
-        final MessageReference node) {
+                                     final MessageReference node) {
         dest.messageExpired(context, this, node);
     }
 

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