QPID-617 : Transactional consume does not ack messages.

Problem is that we were not classing msgs as consumed until onMessage completed in the transactional case. This patch corrects that.

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/M2.1@581540 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 19 insertions, 15 deletions

diff --git a/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java b/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
index 014fd36414..4ac77f1374 100644
--- a/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
+++ b/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
@@ -250,12 +250,28 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
         }
     }
 
-    private void preApplicationProcessing(AbstractJMSMessage jmsMsg) throws JMSException
+    private void preApplicationProcessing(AbstractJMSMessage msg) throws JMSException
     {
 
-        if (_session.getAcknowledgeMode() == Session.CLIENT_ACKNOWLEDGE)
+        switch (_acknowledgeMode)
         {
-            _unacknowledgedDeliveryTags.add(jmsMsg.getDeliveryTag());
+
+            case Session.CLIENT_ACKNOWLEDGE:
+                _unacknowledgedDeliveryTags.add(msg.getDeliveryTag());
+                break;
+
+            case Session.SESSION_TRANSACTED:
+                if (isNoConsume())
+                {
+                    _session.acknowledgeMessage(msg.getDeliveryTag(), false);
+                }
+                else
+                {
+                    _logger.info("Recording tag for commit:" + msg.getDeliveryTag());
+                    _receivedDeliveryTags.add(msg.getDeliveryTag());
+                }
+
+                break;
         }
 
         _session.setInRecovery(false);
@@ -714,18 +730,6 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
             }
 
             break;
-
-        case Session.SESSION_TRANSACTED:
-            if (isNoConsume())
-            {
-                _session.acknowledgeMessage(msg.getDeliveryTag(), false);
-            }
-            else
-            {
-                _receivedDeliveryTags.add(msg.getDeliveryTag());
-            }
-
-            break;
         }
     }