1 files changed, 25 insertions, 2 deletions
diff --git a/cpp/src/tests/storePerftools/asyncPerf/MessageConsumer.cpp b/cpp/src/tests/storePerftools/asyncPerf/MessageConsumer.cpp
index 9b015fc428..1859bde947 100644
--- a/cpp/src/tests/storePerftools/asyncPerf/MessageConsumer.cpp
+++ b/cpp/src/tests/storePerftools/asyncPerf/MessageConsumer.cpp
@@ -23,9 +23,12 @@
 
 #include "MessageConsumer.h"
 
-#include "SimplePersistableQueue.h"
+#include "SimpleQueue.h"
 #include "TestOptions.h"
 
+#include "qpid/asyncStore/AsyncStoreImpl.h"
+#include "qpid/broker/TxnBuffer.h"
+
 #include <stdint.h> // uint32_t
 
 namespace tests {
@@ -33,8 +36,12 @@ namespace storePerftools {
 namespace asyncPerf {
 
 MessageConsumer::MessageConsumer(const TestOptions& perfTestParams,
-                                 boost::shared_ptr<SimplePersistableQueue> queue) :
+                                 qpid::asyncStore::AsyncStoreImpl* store,
+                                 qpid::broker::AsyncResultQueue& arq,
+                                 boost::shared_ptr<SimpleQueue> queue) :
         m_perfTestParams(perfTestParams),
+        m_store(store),
+        m_resultQueue(arq),
         m_queue(queue)
 {}
 
@@ -44,6 +51,13 @@ MessageConsumer::~MessageConsumer()
 void*
 MessageConsumer::runConsumers()
 {
+    const bool useTxns = m_perfTestParams.m_deqTxnBlockSize > 0U;
+    uint16_t txnCnt = 0U;
+    qpid::broker::TxnBuffer* tb = 0;
+    if (useTxns) {
+        tb = new qpid::broker::TxnBuffer(m_resultQueue);
+    }
+
     uint32_t numMsgs = 0;
     while (numMsgs < m_perfTestParams.m_numMsgs) {
         if (m_queue->dispatch()) {
@@ -52,6 +66,15 @@ MessageConsumer::runConsumers()
             ::usleep(1000); // TODO - replace this poller with condition variable
         }
     }
+
+    if (txnCnt) {
+        if (m_perfTestParams.m_durable) {
+            tb->commitLocal(m_store);
+        } else {
+            tb->commit();
+        }
+    }
+
     return reinterpret_cast<void*>(0);
 }