QPID-1497: Ensure policy count and size reflect transactionality of dequeues

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@722200 13f79535-47bb-0310-9956-ffa450edef68

9 files changed, 83 insertions, 10 deletions

diff --git a/cpp/src/qpid/broker/DeliveryRecord.cpp b/cpp/src/qpid/broker/DeliveryRecord.cpp
index e68cc79d8b..b0c060aea5 100644
--- a/cpp/src/qpid/broker/DeliveryRecord.cpp
+++ b/cpp/src/qpid/broker/DeliveryRecord.cpp
@@ -139,6 +139,10 @@ void DeliveryRecord::dequeue(TransactionContext* ctxt) const{
     }
 }
 
+void DeliveryRecord::committed() const{
+    queue->dequeueCommitted(msg);
+}
+
 void DeliveryRecord::reject() 
 {    
     Exchange::shared_ptr alternate = queue->getAlternateExchange();
diff --git a/cpp/src/qpid/broker/DeliveryRecord.h b/cpp/src/qpid/broker/DeliveryRecord.h
index 4b372b2cfa..d7ccab0726 100644
--- a/cpp/src/qpid/broker/DeliveryRecord.h
+++ b/cpp/src/qpid/broker/DeliveryRecord.h
@@ -99,6 +99,7 @@ class DeliveryRecord
     void complete();
     void accept(TransactionContext* ctxt);
     void setEnded();
+    void committed() const;
 
     bool isAcquired() const { return acquired; }
     bool isComplete() const { return completed; }
diff --git a/cpp/src/qpid/broker/DtxAck.cpp b/cpp/src/qpid/broker/DtxAck.cpp
index 100ce2492d..bc7d662afb 100644
--- a/cpp/src/qpid/broker/DtxAck.cpp
+++ b/cpp/src/qpid/broker/DtxAck.cpp
@@ -48,6 +48,7 @@ bool DtxAck::prepare(TransactionContext* ctxt) throw()
 
 void DtxAck::commit() throw()
 {
+    for_each(pending.begin(), pending.end(), mem_fun_ref(&DeliveryRecord::committed));
     pending.clear();
 }
 
diff --git a/cpp/src/qpid/broker/NullMessageStore.cpp b/cpp/src/qpid/broker/NullMessageStore.cpp
index 5a4b23217c..4c13c2adef 100644
--- a/cpp/src/qpid/broker/NullMessageStore.cpp
+++ b/cpp/src/qpid/broker/NullMessageStore.cpp
@@ -32,6 +32,8 @@ namespace broker{
 
 const std::string nullxid = "";
 
+class SimpleDummyCtxt : public TransactionContext {};
+
 class DummyCtxt : public TPCTransactionContext 
 {
     const std::string xid;
@@ -112,7 +114,7 @@ uint32_t NullMessageStore::outstandingQueueAIO(const PersistableQueue& ) {
 
 std::auto_ptr<TransactionContext> NullMessageStore::begin()
 {
-    return std::auto_ptr<TransactionContext>();
+    return std::auto_ptr<TransactionContext>(new SimpleDummyCtxt());
 }
 
 std::auto_ptr<TPCTransactionContext> NullMessageStore::begin(const std::string& xid)
diff --git a/cpp/src/qpid/broker/Queue.cpp b/cpp/src/qpid/broker/Queue.cpp
index b090ffef43..163c471286 100644
--- a/cpp/src/qpid/broker/Queue.cpp
+++ b/cpp/src/qpid/broker/Queue.cpp
@@ -609,7 +609,9 @@ bool Queue::dequeue(TransactionContext* ctxt, const QueuedMessage& msg)
     if (policy.get() && !policy->isEnqueued(msg)) return false;
     {
         Mutex::ScopedLock locker(messageLock);
-        dequeued(msg);
+        if (!ctxt) { 
+            dequeued(msg);
+        }
     }
     if (msg.payload->isPersistent() && store && !lastValueQueue) {
         msg.payload->dequeueAsync(shared_from_this(), store); //increment to async counter -- for message sent to more than one queue
@@ -620,6 +622,12 @@ bool Queue::dequeue(TransactionContext* ctxt, const QueuedMessage& msg)
     return false;
 }
 
+void Queue::dequeueCommitted(const QueuedMessage& msg)
+{
+    Mutex::ScopedLock locker(messageLock);
+    dequeued(msg);    
+}
+
 /**
  * Removes a message from the in-memory delivery queue as well
  * dequeing it from the logical (and persistent if applicable) queue
diff --git a/cpp/src/qpid/broker/Queue.h b/cpp/src/qpid/broker/Queue.h
index 1f619c8812..c11c03a773 100644
--- a/cpp/src/qpid/broker/Queue.h
+++ b/cpp/src/qpid/broker/Queue.h
@@ -222,6 +222,11 @@ namespace qpid {
              * dequeue from store (only done once messages is acknowledged)
              */
             bool dequeue(TransactionContext* ctxt, const QueuedMessage &msg);
+            /**
+             * Inform the queue that a previous transactional dequeue
+             * committed.
+             */
+            void dequeueCommitted(const QueuedMessage& msg);
 
             /**
              * Gets the next available message 
diff --git a/cpp/src/qpid/broker/QueuePolicy.cpp b/cpp/src/qpid/broker/QueuePolicy.cpp
index 6ce94707d5..41a6709d27 100644
--- a/cpp/src/qpid/broker/QueuePolicy.cpp
+++ b/cpp/src/qpid/broker/QueuePolicy.cpp
@@ -46,18 +46,12 @@ void QueuePolicy::dequeued(uint64_t _size)
         if (count.get() > 0) {
             --count;
         } else {
-            //Temporarily disabling checking as it causes java dtx test to fail
-            //TODO: renable these once all tests are passing again
-            //throw Exception(QPID_MSG("Attempted count underflow on dequeue(" << _size << "): " << *this));
-            QPID_LOG(error, "Attempted count underflow on dequeue(" << _size << "): " << *this);
+            throw Exception(QPID_MSG("Attempted count underflow on dequeue(" << _size << "): " << *this));
         }
     }
     if (maxSize) {
         if (_size > size.get()) {
-            //Temporarily disabling checking as it causes java dtx test to fail
-            //TODO: renable these once all tests are passing again
-            //throw Exception(QPID_MSG("Attempted size underflow on dequeue(" << _size << "): " << *this));
-            QPID_LOG(error, "Attempted size underflow on dequeue(" << _size << "): " << *this);
+            throw Exception(QPID_MSG("Attempted size underflow on dequeue(" << _size << "): " << *this));
         } else {
             size -= _size;
         }
diff --git a/cpp/src/qpid/broker/TxAccept.cpp b/cpp/src/qpid/broker/TxAccept.cpp
index 594a466453..c7001e5526 100644
--- a/cpp/src/qpid/broker/TxAccept.cpp
+++ b/cpp/src/qpid/broker/TxAccept.cpp
@@ -37,6 +37,7 @@ void TxAccept::RangeOp::prepare(TransactionContext* ctxt)
 
 void TxAccept::RangeOp::commit()
 {
+    for_each(range.start, range.end, bind(&DeliveryRecord::committed, _1));
     for_each(range.start, range.end, bind(&DeliveryRecord::setEnded, _1));
 }
 
diff --git a/cpp/src/tests/QueuePolicyTest.cpp b/cpp/src/tests/QueuePolicyTest.cpp
index 28f555cf6a..e72f7b2ce3 100644
--- a/cpp/src/tests/QueuePolicyTest.cpp
+++ b/cpp/src/tests/QueuePolicyTest.cpp
@@ -185,5 +185,62 @@ QPID_AUTO_TEST_CASE(testStrictRingPolicy)
     } catch (const ResourceLimitExceededException&) {}    
 }
 
+QPID_AUTO_TEST_CASE(testPolicyWithDtx) 
+{
+    FieldTable args;
+    std::auto_ptr<QueuePolicy> policy = QueuePolicy::createQueuePolicy(5, 0, QueuePolicy::REJECT);
+    policy->update(args);
+
+    ProxySessionFixture f;
+    std::string q("my-policy-queue");
+    f.session.queueDeclare(arg::queue=q, arg::exclusive=true, arg::autoDelete=true, arg::arguments=args);
+    LocalQueue incoming;
+    SubscriptionSettings settings(FlowControl::unlimited());
+    settings.autoAck = 0; // no auto ack.
+    Subscription sub = f.subs.subscribe(incoming, q, settings); 
+    f.session.dtxSelect();
+    Xid tx1(1, "test-dtx-mgr", "tx1");
+    f.session.dtxStart(arg::xid=tx1);
+    for (int i = 0; i < 5; i++) {
+        f.session.messageTransfer(arg::content=client::Message((boost::format("%1%_%2%") % "Message" % (i+1)).str(), q));
+    }
+    f.session.dtxEnd(arg::xid=tx1);
+    f.session.dtxCommit(arg::xid=tx1, arg::onePhase=true);
+
+    Xid tx2(1, "test-dtx-mgr", "tx2");
+    f.session.dtxStart(arg::xid=tx2);
+    for (int i = 0; i < 5; i++) {
+        BOOST_CHECK_EQUAL(incoming.pop().getData(), (boost::format("%1%_%2%") % "Message" % (i+1)).str());
+    }
+    SequenceSet accepting=sub.getUnaccepted();
+    f.session.messageAccept(accepting);
+    f.session.dtxEnd(arg::xid=tx2);
+    f.session.dtxPrepare(arg::xid=tx2);
+    f.session.dtxRollback(arg::xid=tx2);
+    f.session.messageRelease(accepting);
+
+    Xid tx3(1, "test-dtx-mgr", "tx3");
+    f.session.dtxStart(arg::xid=tx3);
+    for (int i = 0; i < 5; i++) {
+        incoming.pop();
+    }
+    accepting=sub.getUnaccepted();
+    f.session.messageAccept(accepting);
+    f.session.dtxEnd(arg::xid=tx3);
+    f.session.dtxPrepare(arg::xid=tx3);
+
+    Session other = f.connection.newSession();
+    try {
+        ScopedSuppressLogging sl; // Suppress messages for expected errors.
+        other.messageTransfer(arg::content=client::Message("Message_6", q));
+        BOOST_FAIL("expecting ResourceLimitExceededException.");
+    } catch (const ResourceLimitExceededException&) {}    
+
+    f.session.dtxCommit(arg::xid=tx3);
+    //now retry and this time should succeed
+    other = f.connection.newSession();
+    other.messageTransfer(arg::content=client::Message("Message_6", q));
+}
+
 
 QPID_AUTO_TEST_SUITE_END()