This is related to QPID-1163.

This is already in trunk and I am porting it to the qpid.0-10 branch.


git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/qpid.0-10@676613 13f79535-47bb-0310-9956-ffa450edef68

2 files changed, 15 insertions, 14 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 c4245d4fc8..8288581538 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
@@ -290,6 +290,7 @@ public abstract class BasicMessageConsumer<H, B> extends Closeable implements Me
         }
         
         _session.setInRecovery(false);
+        preDeliver(jmsMsg);
     }
 
     /**
@@ -642,7 +643,7 @@ public abstract class BasicMessageConsumer<H, B> extends Closeable implements Me
                 // if (!_closed.get())
                 {
 
-                    preDeliver(jmsMessage);
+                    //preDeliver(jmsMessage);
 
                     notifyMessage(jmsMessage);
                 }
@@ -724,6 +725,17 @@ public abstract class BasicMessageConsumer<H, B> extends Closeable implements Me
                 // to send out the appropriate frame
                 msg.setAMQSession(_session);
                 break;
+            case Session.SESSION_TRANSACTED:
+                if (isNoConsume())
+                {
+                    _session.acknowledgeMessage(msg.getDeliveryTag(), false);
+                }
+                else
+                {
+                    _session.addDeliveredMessage(msg.getDeliveryTag());
+                }
+
+                break;
         }
     }
 
@@ -750,17 +762,6 @@ public abstract class BasicMessageConsumer<H, B> extends Closeable implements Me
                 }
 
                 break;
-            case Session.SESSION_TRANSACTED:
-                if (isNoConsume())
-                {
-                    _session.acknowledgeMessage(msg.getDeliveryTag(), false);
-                }
-                else
-                {
-                    _session.addDeliveredMessage(msg.getDeliveryTag());
-                }
-
-                break;
         }
     }
 
diff --git a/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java b/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java
index 6103d92fd8..9230225bd5 100644
--- a/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java
+++ b/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java
@@ -261,11 +261,11 @@ public class BasicMessageConsumer_0_10 extends BasicMessageConsumer<Struct[], By
 
     @Override protected void preApplicationProcessing(AbstractJMSMessage jmsMsg) throws JMSException
     {
-        if (!_session.getTransacted())
+        super.preApplicationProcessing(jmsMsg);
+        if (!_session.getTransacted() && _session.getAcknowledgeMode() != org.apache.qpid.jms.Session.CLIENT_ACKNOWLEDGE)
         {
             _session.addUnacknowledgedMessage(jmsMsg.getDeliveryTag());
         }
-        _session.setInRecovery(false);
     }
 
     @Override public AbstractJMSMessage createJMSMessageFromUnprocessedMessage(