svn commit: r581540 - /incubator/qpid/branches/M2.1/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java

[ritchiem]

Author: ritchiem
Date: Wed Oct  3 02:32:22 2007
New Revision: 581540

URL: http://svn.apache.org/viewvc?rev=581540&view=rev
Log:
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.

Modified:
    
incubator/qpid/branches/M2.1/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java

Modified: 
incubator/qpid/branches/M2.1/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
URL: 
http://svn.apache.org/viewvc/incubator/qpid/branches/M2.1/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java?rev=581540&r1=581539&r2=581540&view=diff
==============================================================================
--- 
incubator/qpid/branches/M2.1/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
 (original)
+++ 
incubator/qpid/branches/M2.1/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
 Wed Oct  3 02:32:22 2007
@@ -250,12 +250,28 @@
         }
     }
 
-    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);
@@ -711,18 +727,6 @@
             if (!_session.isInRecovery())
             {
                 _session.acknowledgeMessage(msg.getDeliveryTag(), false);
-            }
-
-            break;
-
-        case Session.SESSION_TRANSACTED:
-            if (isNoConsume())
-            {
-                _session.acknowledgeMessage(msg.getDeliveryTag(), false);
-            }
-            else
-            {
-                _receivedDeliveryTags.add(msg.getDeliveryTag());
             }
 
             break;