QPID-3858: WIP - async txns for msg publish pathway, but there are some race/thread issues to sort out.

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

1 files changed, 65 insertions, 8 deletions

diff --git a/cpp/src/qpid/asyncStore/TxnHandleImpl.cpp b/cpp/src/qpid/asyncStore/TxnHandleImpl.cpp
index 7ce01f881c..945b50861d 100644
--- a/cpp/src/qpid/asyncStore/TxnHandleImpl.cpp
+++ b/cpp/src/qpid/asyncStore/TxnHandleImpl.cpp
@@ -23,6 +23,8 @@
 
 #include "TxnHandleImpl.h"
 
+#include "qpid/Exception.h"
+#include "qpid/broker/TxnBuffer.h"
 #include "qpid/messaging/PrivateImplRef.h"
 
 #include <uuid/uuid.h>
@@ -30,16 +32,42 @@
 namespace qpid {
 namespace asyncStore {
 
+TxnHandleImpl::TxnHandleImpl() :
+        m_tpcFlag(false),
+        m_asyncOpCnt(0UL),
+        m_txnBuffer(0)
+{
+    createLocalXid();
+}
+
+TxnHandleImpl::TxnHandleImpl(qpid::broker::TxnBuffer* tb) :
+        m_tpcFlag(false),
+        m_asyncOpCnt(0UL),
+        m_txnBuffer(tb)
+{
+    createLocalXid();
+}
+
 TxnHandleImpl::TxnHandleImpl(const std::string& xid) :
         m_xid(xid),
-        m_tpcFlag(!xid.empty())
-{
-    if (m_xid.empty()) { // create a local xid from a random uuid
-        uuid_t uuid;
-        ::uuid_generate_random(uuid);
-        char uuidStr[37]; // 36-char uuid + trailing '\0'
-        ::uuid_unparse(uuid, uuidStr);
-//        m_xid.assign(uuidStr);
+        m_tpcFlag(!xid.empty()),
+        m_asyncOpCnt(0UL),
+        m_txnBuffer(0)
+{
+    if (m_xid.empty()) {
+        createLocalXid();
+    }
+}
+
+TxnHandleImpl::TxnHandleImpl(const std::string& xid,
+                             qpid::broker::TxnBuffer* tb) :
+        m_xid(xid),
+        m_tpcFlag(!xid.empty()),
+        m_asyncOpCnt(0UL),
+        m_txnBuffer(tb)
+{
+    if (m_xid.empty()) {
+        createLocalXid();
     }
 }
 
@@ -58,4 +86,33 @@ TxnHandleImpl::is2pc() const
     return m_tpcFlag;
 }
 
+void
+TxnHandleImpl::incrOpCnt()
+{
+    ++m_asyncOpCnt;
+}
+
+void
+TxnHandleImpl::decrOpCnt()
+{
+    if (m_asyncOpCnt == 0UL) {
+        throw qpid::Exception("Transaction async operation count underflow");
+    }
+    if (--m_asyncOpCnt == 0UL && m_txnBuffer) {
+        m_txnBuffer->asyncLocalCommit();
+    }
+}
+
+// private
+void
+TxnHandleImpl::createLocalXid()
+{
+    uuid_t uuid;
+    ::uuid_generate_random(uuid);
+    char uuidStr[37]; // 36-char uuid + trailing '\0'
+    ::uuid_unparse(uuid, uuidStr);
+    m_xid.assign(uuidStr);
+//std::cout << "TTT TxnHandleImpl::createLocalXid(): Local XID created: \"" << m_xid << "\"" << std::endl << std::flush;
+}
+
 }} // namespace qpid::asyncStore