From 8e6fe414ad9078e765d3f8579dcd2898636b9ea1 Mon Sep 17 00:00:00 2001
From: gtully <gary.tully@gmail.com>
Date: Wed, 21 Sep 2016 13:59:45 +0100
Subject: [PATCH] https://issues.apache.org/jira/browse/AMQ-6422 - move
 delivery tracking to pumpoutbound and additional test that shows how the
 presettle case breaks. Thanks to Robbie Gemmell for the feedback

---
 .../transport/amqp/protocol/AmqpSender.java   |  2 +-
 .../amqp/interop/AmqpReceiverTest.java        | 41 +++++++++++++++++++
 2 files changed, 42 insertions(+), 1 deletion(-)

diff --git a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/protocol/AmqpSender.java b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/protocol/AmqpSender.java
index 75f237157f..455e0b09af 100644
--- a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/protocol/AmqpSender.java
+++ b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/protocol/AmqpSender.java
@@ -292,7 +292,6 @@ public class AmqpSender extends AmqpAbstractLink<Sender> {
         }
 
         pumpOutbound();
-        logicalDeliveryCount++;
     }
 
     @Override
@@ -410,6 +409,7 @@ public class AmqpSender extends AmqpAbstractLink<Sender> {
                         }
                         currentBuffer = null;
                         currentDelivery = null;
+                        logicalDeliveryCount++;
                     }
                 } else {
                     return;
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/interop/AmqpReceiverTest.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/interop/AmqpReceiverTest.java
index c68e850024..b73f08739e 100644
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/interop/AmqpReceiverTest.java
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/interop/AmqpReceiverTest.java
@@ -223,6 +223,47 @@ public class AmqpReceiverTest extends AmqpClientTestSupport {
         connection.close();
     }
 
+    @Test(timeout = 60000)
+    @Repeat(repetitions = 1)
+    public void testPresettledReceiverReadsAllMessagesInNonFlowBatch() throws Exception {
+        final int MSG_COUNT = 100;
+        sendMessages(getTestName(), MSG_COUNT, false);
+
+        AmqpClient client = createAmqpClient();
+        AmqpConnection connection = client.connect();
+        AmqpSession session = connection.createSession();
+
+        AmqpReceiver receiver = session.createReceiver("queue://" + getTestName(), null, false, true);
+
+        QueueViewMBean queueView = getProxyToQueue(getTestName());
+        assertEquals(MSG_COUNT, queueView.getQueueSize());
+        assertEquals(0, queueView.getDispatchCount());
+
+        receiver.flow(20);
+        // consume less that flow
+        for (int j=0;j<10;j++) {
+            assertNotNull(receiver.receive(5, TimeUnit.SECONDS));
+        }
+
+        // flow more and consume all
+        receiver.flow(10);
+        for (int j=0;j<20;j++) {
+            assertNotNull(receiver.receive(5, TimeUnit.SECONDS));
+        }
+
+        // remainder
+        receiver.flow(70);
+        for (int j=0;j<70;j++) {
+            assertNotNull(receiver.receive(5, TimeUnit.SECONDS));
+        }
+
+        receiver.close();
+
+        assertEquals(0, queueView.getQueueSize());
+
+        connection.close();
+    }
+
     @Test(timeout = 60000)
     @Repeat(repetitions = 1)
     public void testTwoQueueReceiversOnSameConnectionReadMessagesNoDispositions() throws Exception {