QPID-3858: More restructuring: extracted MessageContext from MockPersistableMessage, also TransactionAsyncContext from MockTransactionContext. Further name changes will be needed to resolve some bad naming.

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/asyncstore@1338262 13f79535-47bb-0310-9956-ffa450edef68

13 files changed, 281 insertions, 117 deletions

diff --git a/cpp/src/tests/CMakeLists.txt b/cpp/src/tests/CMakeLists.txt
index e8ad5ee8bd..c9b4538b1d 100644
--- a/cpp/src/tests/CMakeLists.txt
+++ b/cpp/src/tests/CMakeLists.txt
@@ -379,6 +379,7 @@ endif (UNIX)
 
 # Async store perf test (asyncPerf)
 set (asyncStorePerf_SOURCES
+    storePerftools/asyncPerf/MessageContext.cpp
 	storePerftools/asyncPerf/MockPersistableMessage.cpp
 	storePerftools/asyncPerf/MockPersistableQueue.cpp
 	storePerftools/asyncPerf/MockTransactionContext.cpp
@@ -387,6 +388,7 @@ set (asyncStorePerf_SOURCES
 	storePerftools/asyncPerf/QueuedMessage.cpp
 	storePerftools/asyncPerf/TestOptions.cpp
 	storePerftools/asyncPerf/TestResult.cpp
+	storePerftools/asyncPerf/TransactionAsyncContext.cpp
 	
 	storePerftools/common/Parameters.cpp
     storePerftools/common/PerftoolError.cpp
diff --git a/cpp/src/tests/storePerftools/asyncPerf/MessageContext.cpp b/cpp/src/tests/storePerftools/asyncPerf/MessageContext.cpp
new file mode 100644
index 0000000000..3d18213c7d
--- /dev/null
+++ b/cpp/src/tests/storePerftools/asyncPerf/MessageContext.cpp
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * \file MessageContext.cpp
+ */
+
+#include "MessageContext.h"
+
+namespace tests {
+namespace storePerftools {
+namespace asyncPerf {
+
+MessageContext::MessageContext(MockPersistableMessage::shared_ptr msg,
+                                                       const qpid::asyncStore::AsyncOperation::opCode op,
+                                                       MockPersistableQueue* q) :
+        m_msg(msg),
+        m_op(op),
+        m_q(q)
+{
+    assert(m_msg.get() != 0);
+    assert(m_q != 0);
+}
+
+MessageContext::~MessageContext()
+{}
+
+qpid::asyncStore::AsyncOperation::opCode
+MessageContext::getOpCode() const
+{
+    return m_op;
+}
+
+const char*
+MessageContext::getOpStr() const
+{
+    return qpid::asyncStore::AsyncOperation::getOpStr(m_op);
+}
+
+MockPersistableMessage::shared_ptr
+MessageContext::getMessage() const
+{
+    return m_msg;
+}
+
+MockPersistableQueue*
+MessageContext::getQueue() const
+{
+    return m_q;
+}
+
+void
+MessageContext::destroy()
+{
+    delete this;
+}
+
+}}} // namespace tests::storePerftools::asyncPerf
diff --git a/cpp/src/tests/storePerftools/asyncPerf/MessageContext.h b/cpp/src/tests/storePerftools/asyncPerf/MessageContext.h
new file mode 100644
index 0000000000..49a00fa356
--- /dev/null
+++ b/cpp/src/tests/storePerftools/asyncPerf/MessageContext.h
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * \file MessageContext.h
+ */
+
+#ifndef tests_storePerfTools_asyncPerf_MessageContext_h_
+#define tests_storePerfTools_asyncPerf_MessageContext_h_
+
+#include "MockPersistableMessage.h"
+
+namespace tests {
+namespace storePerftools {
+namespace asyncPerf {
+
+class MessageContext : public qpid::broker::BrokerContext
+{
+public:
+    MessageContext(MockPersistableMessage::shared_ptr msg,
+                   const qpid::asyncStore::AsyncOperation::opCode op,
+                   MockPersistableQueue* q);
+    virtual ~MessageContext();
+    qpid::asyncStore::AsyncOperation::opCode getOpCode() const;
+    const char* getOpStr() const;
+    MockPersistableMessage::shared_ptr getMessage() const;
+    MockPersistableQueue* getQueue() const;
+    void destroy();
+protected:
+    MockPersistableMessage::shared_ptr m_msg;
+    const qpid::asyncStore::AsyncOperation::opCode m_op;
+    MockPersistableQueue* m_q;
+};
+
+}}} // namespace tests::storePerftools::asyncPerf
+
+#endif // tests_storePerfTools_asyncPerf_MessageContext_h_
diff --git a/cpp/src/tests/storePerftools/asyncPerf/MockPersistableMessage.cpp b/cpp/src/tests/storePerftools/asyncPerf/MockPersistableMessage.cpp
index 5cc829f4d2..564d133c76 100644
--- a/cpp/src/tests/storePerftools/asyncPerf/MockPersistableMessage.cpp
+++ b/cpp/src/tests/storePerftools/asyncPerf/MockPersistableMessage.cpp
@@ -22,6 +22,8 @@
  */
 
 #include "MockPersistableMessage.h"
+
+#include "MessageContext.h"
 #include "MockPersistableQueue.h" // debug statements in enqueueComplete() and dequeueComplete()
 
 #include "qpid/asyncStore/AsyncStoreImpl.h"
@@ -30,34 +32,6 @@ namespace tests {
 namespace storePerftools {
 namespace asyncPerf {
 
-// --- Inner class Queue::MessageContext ---
-
-MockPersistableMessage::MessageContext::MessageContext(MockPersistableMessagePtr msg,
-                                                       const qpid::asyncStore::AsyncOperation::opCode op,
-                                                       MockPersistableQueue* q) :
-        m_msg(msg),
-        m_op(op),
-        m_q(q)
-{}
-
-MockPersistableMessage::MessageContext::~MessageContext()
-{}
-
-const char*
-MockPersistableMessage::MessageContext::getOp() const
-{
-    return qpid::asyncStore::AsyncOperation::getOpStr(m_op);
-}
-
-void
-MockPersistableMessage::MessageContext::destroy()
-{
-    delete this;
-}
-
-// --- Class MockPersistableMessage ---
-
-
 MockPersistableMessage::MockPersistableMessage(const char* msgData,
                                                const uint32_t msgSize,
                                                qpid::asyncStore::AsyncStoreImpl* store,
@@ -78,26 +52,24 @@ MockPersistableMessage::handleAsyncResult(const qpid::broker::AsyncResult* res,
 {
     if (bc) {
         MessageContext* mc = dynamic_cast<MessageContext*>(bc);
-        if (mc->m_msg) {
-            if (res->errNo) {
-                // TODO: Handle async failure here
-                std::cerr << "Message pid=0x" << std::hex << mc->m_msg->m_persistenceId << std::dec << ": Operation "
-                          << mc->getOp() << ": failure " << res->errNo << " (" << res->errMsg << ")" << std::endl;
-            } else {
-                // Handle async success here
-                switch(mc->m_op) {
-                case qpid::asyncStore::AsyncOperation::MSG_DEQUEUE:
-                    mc->m_msg->dequeueComplete(mc);
-                    break;
-                case qpid::asyncStore::AsyncOperation::MSG_ENQUEUE:
-                    mc->m_msg->enqueueComplete(mc);
-                    break;
-                default:
-                    std::ostringstream oss;
-                    oss << "tests::storePerftools::asyncPerf::MockPersistableMessage::handleAsyncResult(): Unknown async queue operation: " << mc->m_op;
-                    throw qpid::Exception(oss.str());
-                };
-            }
+        if (res->errNo) {
+            // TODO: Handle async failure here
+            std::cerr << "Message pid=0x" << std::hex << mc->getMessage()->m_persistenceId << std::dec << ": Operation "
+                      << mc->getOpStr() << ": failure " << res->errNo << " (" << res->errMsg << ")" << std::endl;
+        } else {
+            // Handle async success here
+            switch(mc->getOpCode()) {
+            case qpid::asyncStore::AsyncOperation::MSG_DEQUEUE:
+                mc->getMessage()->dequeueComplete(mc);
+                break;
+            case qpid::asyncStore::AsyncOperation::MSG_ENQUEUE:
+                mc->getMessage()->enqueueComplete(mc);
+                break;
+            default:
+                std::ostringstream oss;
+                oss << "tests::storePerftools::asyncPerf::MockPersistableMessage::handleAsyncResult(): Unknown async queue operation: " << mc->getOpCode();
+                throw qpid::Exception(oss.str());
+            };
         }
     }
     if (bc) delete bc;
@@ -167,7 +139,7 @@ void
 MockPersistableMessage::enqueueComplete(const MessageContext* mc)
 {
 //std::cout << "~~~~~ Message pid=0x" << std::hex << mc->m_msg->getPersistenceId() << std::dec << ": enqueueComplete() on queue \"" << mc->m_q->getName() << "\"" << std::endl << std::flush;
-    assert(mc->m_msg.get() == this);
+    assert(mc->getMessage().get() == this);
 }
 
 // protected
@@ -175,7 +147,7 @@ void
 MockPersistableMessage::dequeueComplete(const MessageContext* mc)
 {
 //std::cout << "~~~~~ Message pid=0x" << std::hex << mc->m_msg->getPersistenceId() << std::dec << ": dequeueComplete() on queue \"" << mc->m_q->getName() << "\"" << std::endl << std::flush;
-    assert(mc->m_msg.get() == this);
+    assert(mc->getMessage().get() == this);
 }
 
 }}} // namespace tests::storePerftools::asyncPerf
diff --git a/cpp/src/tests/storePerftools/asyncPerf/MockPersistableMessage.h b/cpp/src/tests/storePerftools/asyncPerf/MockPersistableMessage.h
index 7039c4bd08..83ef189f9e 100644
--- a/cpp/src/tests/storePerftools/asyncPerf/MockPersistableMessage.h
+++ b/cpp/src/tests/storePerftools/asyncPerf/MockPersistableMessage.h
@@ -41,27 +41,13 @@ namespace tests {
 namespace storePerftools {
 namespace asyncPerf {
 
-class MockPersistableMessage;
+class MessageContext;
 class MockPersistableQueue;
 
-typedef boost::shared_ptr<MockPersistableMessage> MockPersistableMessagePtr;
-
 class MockPersistableMessage: public qpid::broker::PersistableMessage, qpid::broker::DataSource
 {
 public:
-    class MessageContext : public qpid::broker::BrokerContext
-    {
-    public:
-        MessageContext(MockPersistableMessagePtr msg,
-                       const qpid::asyncStore::AsyncOperation::opCode op,
-                       MockPersistableQueue* q);
-        virtual ~MessageContext();
-        const char* getOp() const;
-        void destroy();
-        MockPersistableMessagePtr m_msg;
-        const qpid::asyncStore::AsyncOperation::opCode m_op;
-        MockPersistableQueue* m_q;
-    };
+    typedef boost::shared_ptr<MockPersistableMessage> shared_ptr;
 
     MockPersistableMessage(const char* msgData,
                            const uint32_t msgSize,
diff --git a/cpp/src/tests/storePerftools/asyncPerf/MockPersistableQueue.cpp b/cpp/src/tests/storePerftools/asyncPerf/MockPersistableQueue.cpp
index aa55ee5a7e..e607649cc7 100644
--- a/cpp/src/tests/storePerftools/asyncPerf/MockPersistableQueue.cpp
+++ b/cpp/src/tests/storePerftools/asyncPerf/MockPersistableQueue.cpp
@@ -23,6 +23,7 @@
 
 #include "MockPersistableQueue.h"
 
+#include "MessageContext.h"
 #include "MockPersistableMessage.h"
 #include "MockTransactionContext.h"
 #include "QueueContext.h"
@@ -129,12 +130,12 @@ MockPersistableQueue::runEnqueues()
         if (useTxn && txnCnt == 0) {
             txn.reset(new MockTransactionContext(m_store)); // equivalent to begin()
         }
-        MockPersistableMessagePtr msg(new MockPersistableMessage(m_msgData, m_perfTestOpts.m_msgSize, m_store, true));
+        MockPersistableMessage::shared_ptr msg(new MockPersistableMessage(m_msgData, m_perfTestOpts.m_msgSize, m_store, true));
         msg->setPersistenceId(m_store->getNextRid());
         qpid::broker::EnqueueHandle enqHandle = m_store->createEnqueueHandle(msg->getHandle(), m_queueHandle);
-        MockPersistableMessage::MessageContext* msgCtxt = new MockPersistableMessage::MessageContext(msg,
-                                                                                                     qpid::asyncStore::AsyncOperation::MSG_ENQUEUE,
-                                                                                                     this);
+        MessageContext* msgCtxt = new MessageContext(msg,
+                                                     qpid::asyncStore::AsyncOperation::MSG_ENQUEUE,
+                                                     this);
         if (useTxn) {
             m_store->submitEnqueue(enqHandle,
                                    txn->getHandle(),
@@ -176,9 +177,9 @@ MockPersistableQueue::runDequeues()
         pop(qm);
         if (qm.get()) {
             qpid::broker::EnqueueHandle enqHandle = qm->getEnqueueHandle();
-            qpid::broker::BrokerContext* bc = new MockPersistableMessage::MessageContext(qm->getMessage(),
-                                                                                         qpid::asyncStore::AsyncOperation::MSG_DEQUEUE,
-                                                                                         this);
+            qpid::broker::BrokerContext* bc = new MessageContext(qm->getMessage(),
+                                                                 qpid::asyncStore::AsyncOperation::MSG_DEQUEUE,
+                                                                 this);
             if (useTxn) {
                 m_store->submitDequeue(enqHandle,
                                        txn->getHandle(),
diff --git a/cpp/src/tests/storePerftools/asyncPerf/MockTransactionContext.cpp b/cpp/src/tests/storePerftools/asyncPerf/MockTransactionContext.cpp
index 10be34c6f5..b1a375f04c 100644
--- a/cpp/src/tests/storePerftools/asyncPerf/MockTransactionContext.cpp
+++ b/cpp/src/tests/storePerftools/asyncPerf/MockTransactionContext.cpp
@@ -24,6 +24,7 @@
 #include "MockTransactionContext.h"
 
 #include "QueuedMessage.h"
+#include "TransactionAsyncContext.h"
 
 #include "qpid/asyncStore/AsyncStoreImpl.h"
 
@@ -33,6 +34,7 @@ namespace asyncPerf {
 
 // --- Inner class MockTransactionContext::QueueContext ---
 
+/*
 MockTransactionContext::TransactionContext::TransactionContext(MockTransactionContext* tc,
                                                                const qpid::asyncStore::AsyncOperation::opCode op) :
         m_tc(tc),
@@ -53,6 +55,7 @@ MockTransactionContext::TransactionContext::destroy()
 {
     delete this;
 }
+*/
 
 // --- Class MockTransactionContext ---
 
@@ -76,30 +79,28 @@ MockTransactionContext::handleAsyncResult(const qpid::broker::AsyncResult* res,
                                           qpid::broker::BrokerContext* bc)
 {
     if (bc && res) {
-        TransactionContext* tc = dynamic_cast<TransactionContext*>(bc);
-        if (tc->m_tc) {
-            if (res->errNo) {
-                // TODO: Handle async failure here
-                std::cerr << "Transaction xid=\"" << tc->m_tc->getXid() << "\": Operation " << tc->getOp() << ": failure "
-                          << res->errNo << " (" << res->errMsg << ")" << std::endl;
-            } else {
-                // Handle async success here
-                switch(tc->m_op) {
-                case qpid::asyncStore::AsyncOperation::TXN_PREPARE:
-                    tc->m_tc->prepareComplete(tc);
-                    break;
-                case qpid::asyncStore::AsyncOperation::TXN_COMMIT:
-                    tc->m_tc->commitComplete(tc);
-                    break;
-                case qpid::asyncStore::AsyncOperation::TXN_ABORT:
-                    tc->m_tc->abortComplete(tc);
-                    break;
-                default:
-                    std::ostringstream oss;
-                    oss << "tests::storePerftools::asyncPerf::MockTransactionContext::handleAsyncResult(): Unknown async operation: " << tc->m_op;
-                    throw qpid::Exception(oss.str());
-                };
-            }
+        TransactionAsyncContext* tac = dynamic_cast<TransactionAsyncContext*>(bc);
+        if (res->errNo) {
+            // TODO: Handle async failure here
+            std::cerr << "Transaction xid=\"" << tac->getTransactionContext()->getXid() << "\": Operation " << tac->getOpStr() << ": failure "
+                      << res->errNo << " (" << res->errMsg << ")" << std::endl;
+        } else {
+            // Handle async success here
+            switch(tac->getOpCode()) {
+            case qpid::asyncStore::AsyncOperation::TXN_PREPARE:
+                tac->getTransactionContext()->prepareComplete(tac);
+                break;
+            case qpid::asyncStore::AsyncOperation::TXN_COMMIT:
+                tac->getTransactionContext()->commitComplete(tac);
+                break;
+            case qpid::asyncStore::AsyncOperation::TXN_ABORT:
+                tac->getTransactionContext()->abortComplete(tac);
+                break;
+            default:
+                std::ostringstream oss;
+                oss << "tests::storePerftools::asyncPerf::MockTransactionContext::handleAsyncResult(): Unknown async operation: " << tac->getOpCode();
+                throw qpid::Exception(oss.str());
+            };
         }
     }
     if (bc) delete bc;
@@ -154,7 +155,7 @@ MockTransactionContext::abort()
     }
     m_store->submitAbort(m_txnHandle,
                          &handleAsyncResult,
-                         dynamic_cast<qpid::broker::BrokerContext*>(new TransactionContext(this, qpid::asyncStore::AsyncOperation::TXN_ABORT)));
+                         dynamic_cast<qpid::broker::BrokerContext*>(new TransactionAsyncContext(this, qpid::asyncStore::AsyncOperation::TXN_ABORT)));
 //std::cout << "*TXN* abort: xid=" << m_txnHandle.getXid() << "; 2PC=" << (m_txnHandle.is2pc()?"T":"F") << std::endl;
 }
 
@@ -173,7 +174,7 @@ MockTransactionContext::commit()
     }
     m_store->submitCommit(m_txnHandle,
                           &handleAsyncResult,
-                          dynamic_cast<qpid::broker::BrokerContext*>(new TransactionContext(this, qpid::asyncStore::AsyncOperation::TXN_COMMIT)));
+                          dynamic_cast<qpid::broker::BrokerContext*>(new TransactionAsyncContext(this, qpid::asyncStore::AsyncOperation::TXN_COMMIT)));
 //std::cout << "*TXN* commit: xid=" << m_txnHandle.getXid() << "; 2PC=" << (m_txnHandle.is2pc()?"T":"F") << std::endl;
 }
 
@@ -184,13 +185,13 @@ MockTransactionContext::localPrepare()
 {
     m_store->submitPrepare(m_txnHandle,
                            &handleAsyncResult,
-                           dynamic_cast<qpid::broker::BrokerContext*>(new TransactionContext(this, qpid::asyncStore::AsyncOperation::TXN_PREPARE)));
+                           dynamic_cast<qpid::broker::BrokerContext*>(new TransactionAsyncContext(this, qpid::asyncStore::AsyncOperation::TXN_PREPARE)));
 //std::cout << "*TXN* localPrepare: xid=" << m_txnHandle.getXid() << "; 2PC=" << (m_txnHandle.is2pc()?"T":"F") << std::endl;
 }
 
 // protected
 void
-MockTransactionContext::prepareComplete(const TransactionContext* tc)
+MockTransactionContext::prepareComplete(const TransactionAsyncContext* tc)
 {
     qpid::sys::ScopedLock<qpid::sys::Mutex> l(m_enqueuedMsgsMutex);
     while (!m_enqueuedMsgs.empty()) {
@@ -198,25 +199,25 @@ MockTransactionContext::prepareComplete(const TransactionContext* tc)
         m_enqueuedMsgs.pop_front();
     }
 //std::cout << "~~~~~ Transaction xid=\"" << tc->m_tc->getXid() << "\": prepareComplete()" << std::endl << std::flush;
-    assert(tc->m_tc == this);
+    assert(tc->getTransactionContext() == this);
 }
 
 
 // protected
 void
-MockTransactionContext::abortComplete(const TransactionContext* tc)
+MockTransactionContext::abortComplete(const TransactionAsyncContext* tc)
 {
 //std::cout << "~~~~~ Transaction xid=\"" << tc->m_tc->getXid() << "\": abortComplete()" << std::endl << std::flush;
-    assert(tc->m_tc == this);
+    assert(tc->getTransactionContext() == this);
 }
 
 
 // protected
 void
-MockTransactionContext::commitComplete(const TransactionContext* tc)
+MockTransactionContext::commitComplete(const TransactionAsyncContext* tc)
 {
 //std::cout << "~~~~~ Transaction xid=\"" << tc->m_tc->getXid() << "\": commitComplete()" << std::endl << std::flush;
-    assert(tc->m_tc == this);
+    assert(tc->getTransactionContext() == this);
 }
 
 }}} // namespace tests::storePerftools::asyncPerf
diff --git a/cpp/src/tests/storePerftools/asyncPerf/MockTransactionContext.h b/cpp/src/tests/storePerftools/asyncPerf/MockTransactionContext.h
index 883da198bb..eb56645d47 100644
--- a/cpp/src/tests/storePerftools/asyncPerf/MockTransactionContext.h
+++ b/cpp/src/tests/storePerftools/asyncPerf/MockTransactionContext.h
@@ -44,13 +44,17 @@ namespace storePerftools {
 namespace asyncPerf {
 
 class QueuedMessage;
+class TransactionAsyncContext;
 
 class MockTransactionContext : public qpid::broker::TransactionContext
 {
 public:
+    typedef boost::shared_ptr<MockTransactionContext> shared_ptr;
+
     // NOTE: TransactionContext - Bad naming? This context is the async return handling context for class
     //       MockTransactionContext async ops. Other classes using this pattern simply use XXXContext for this class
     //       (e.g. QueueContext in MockPersistableQueue), but in this case it may be confusing.
+/*
     class TransactionContext : public qpid::broker::BrokerContext
     {
     public:
@@ -62,6 +66,7 @@ public:
         MockTransactionContext* m_tc;
         const qpid::asyncStore::AsyncOperation::opCode m_op;
     };
+*/
 
     MockTransactionContext(qpid::asyncStore::AsyncStoreImpl* store,
                            const std::string& xid = std::string());
@@ -88,9 +93,9 @@ protected:
     void localPrepare();
 
     // --- Ascnc op completions (called through handleAsyncResult) ---
-    void prepareComplete(const TransactionContext* tc);
-    void abortComplete(const TransactionContext* tc);
-    void commitComplete(const TransactionContext* tc);
+    void prepareComplete(const TransactionAsyncContext* tc);
+    void abortComplete(const TransactionAsyncContext* tc);
+    void commitComplete(const TransactionAsyncContext* tc);
 
 };
 
diff --git a/cpp/src/tests/storePerftools/asyncPerf/QueueContext.h b/cpp/src/tests/storePerftools/asyncPerf/QueueContext.h
index 0c69381e31..025a9dcf4f 100644
--- a/cpp/src/tests/storePerftools/asyncPerf/QueueContext.h
+++ b/cpp/src/tests/storePerftools/asyncPerf/QueueContext.h
@@ -25,6 +25,7 @@
 #define tests_storePerftools_asyncPerf_QueueContext_h_
 
 #include "MockPersistableQueue.h"
+#include "qpid/broker/BrokerContext.h"
 
 namespace tests {
 namespace storePerftools {
diff --git a/cpp/src/tests/storePerftools/asyncPerf/QueuedMessage.cpp b/cpp/src/tests/storePerftools/asyncPerf/QueuedMessage.cpp
index 315e202d8b..9e5e131a28 100644
--- a/cpp/src/tests/storePerftools/asyncPerf/QueuedMessage.cpp
+++ b/cpp/src/tests/storePerftools/asyncPerf/QueuedMessage.cpp
@@ -29,7 +29,7 @@ namespace tests {
 namespace storePerftools {
 namespace asyncPerf {
 
-QueuedMessage::QueuedMessage(MockPersistableMessagePtr msg,
+QueuedMessage::QueuedMessage(MockPersistableMessage::shared_ptr msg,
                              qpid::broker::EnqueueHandle& enqHandle,
                              MockTransactionContextPtr txn) :
         m_msg(msg),
@@ -44,7 +44,7 @@ QueuedMessage::QueuedMessage(MockPersistableMessagePtr msg,
 QueuedMessage::~QueuedMessage()
 {}
 
-MockPersistableMessagePtr
+MockPersistableMessage::shared_ptr
 QueuedMessage::getMessage() const
 {
     return m_msg;
diff --git a/cpp/src/tests/storePerftools/asyncPerf/QueuedMessage.h b/cpp/src/tests/storePerftools/asyncPerf/QueuedMessage.h
index be9694f6e5..a11b23888a 100644
--- a/cpp/src/tests/storePerftools/asyncPerf/QueuedMessage.h
+++ b/cpp/src/tests/storePerftools/asyncPerf/QueuedMessage.h
@@ -24,6 +24,8 @@
 #ifndef tests_storePerftools_asyncPerf_QueuedMessage_h_
 #define tests_storePerftools_asyncPerf_QueuedMessage_h_
 
+#include "MockPersistableMessage.h"
+
 #include "qpid/broker/EnqueueHandle.h"
 #include <boost/shared_ptr.hpp>
 
@@ -31,27 +33,24 @@ namespace tests {
 namespace storePerftools {
 namespace asyncPerf {
 
-class MockPersistableMessage;
 class MockTransactionContext;
-
-typedef boost::shared_ptr<MockPersistableMessage> MockPersistableMessagePtr;
 typedef boost::shared_ptr<MockTransactionContext> MockTransactionContextPtr;
 
 class QueuedMessage
 {
 public:
-    QueuedMessage(MockPersistableMessagePtr msg,
+    QueuedMessage(MockPersistableMessage::shared_ptr msg,
                   qpid::broker::EnqueueHandle& enqHandle,
                   MockTransactionContextPtr txn);
     virtual ~QueuedMessage();
-    MockPersistableMessagePtr getMessage() const;
+    MockPersistableMessage::shared_ptr getMessage() const;
     qpid::broker::EnqueueHandle getEnqueueHandle() const;
     MockTransactionContextPtr getTransactionContext() const;
     bool isTransactional() const;
     void clearTransaction();
 
 protected:
-    MockPersistableMessagePtr m_msg;
+    MockPersistableMessage::shared_ptr m_msg;
     qpid::broker::EnqueueHandle m_enqHandle;
     MockTransactionContextPtr m_txn;
 };
diff --git a/cpp/src/tests/storePerftools/asyncPerf/TransactionAsyncContext.cpp b/cpp/src/tests/storePerftools/asyncPerf/TransactionAsyncContext.cpp
new file mode 100644
index 0000000000..c0c94e3a18
--- /dev/null
+++ b/cpp/src/tests/storePerftools/asyncPerf/TransactionAsyncContext.cpp
@@ -0,0 +1,43 @@
+#include "TransactionAsyncContext.h"
+
+namespace tests {
+namespace storePerftools {
+namespace asyncPerf {
+
+TransactionAsyncContext::TransactionAsyncContext(MockTransactionContext* tc,
+                                                 const qpid::asyncStore::AsyncOperation::opCode op):
+        qpid::broker::BrokerContext(),
+        m_tc(tc),
+        m_op(op)
+{
+    assert(tc != 0);
+}
+
+TransactionAsyncContext::~TransactionAsyncContext()
+{}
+
+qpid::asyncStore::AsyncOperation::opCode
+TransactionAsyncContext::getOpCode() const
+{
+    return m_op;
+}
+
+const char*
+TransactionAsyncContext::getOpStr() const
+{
+    return qpid::asyncStore::AsyncOperation::getOpStr(m_op);
+}
+
+MockTransactionContext*
+TransactionAsyncContext::getTransactionContext() const
+{
+    return m_tc;
+}
+
+void
+TransactionAsyncContext::destroy()
+{
+    delete this;
+}
+
+}}} // namespace tests::storePerftools::asyncPerf
diff --git a/cpp/src/tests/storePerftools/asyncPerf/TransactionAsyncContext.h b/cpp/src/tests/storePerftools/asyncPerf/TransactionAsyncContext.h
new file mode 100644
index 0000000000..092958a93b
--- /dev/null
+++ b/cpp/src/tests/storePerftools/asyncPerf/TransactionAsyncContext.h
@@ -0,0 +1,27 @@
+#ifndef tests_storePerftools_asyncPerf_TransactionAsyncContext_h_
+#define tests_storePerftools_asyncPerf_TransactionAsyncContext_h_
+
+#include "MockTransactionContext.h"
+#include "qpid/broker/BrokerContext.h"
+
+namespace tests {
+namespace storePerftools {
+namespace asyncPerf {
+
+class TransactionAsyncContext: public qpid::broker::BrokerContext {
+public:
+    TransactionAsyncContext(MockTransactionContext* tc,
+                            const qpid::asyncStore::AsyncOperation::opCode op);
+    virtual ~TransactionAsyncContext();
+    qpid::asyncStore::AsyncOperation::opCode getOpCode() const;
+    const char* getOpStr() const;
+    MockTransactionContext* getTransactionContext() const;
+    void destroy();
+protected:
+    MockTransactionContext* m_tc;
+    const qpid::asyncStore::AsyncOperation::opCode m_op;
+};
+
+}}} // namespace tests::storePerftools::asyncPerf
+
+#endif // tests_storePerftools_asyncPerf_TransactionAsyncContext_h_