Suspension of one xid on a channel should allow for work on other xids to be done prior to resuming the initial xid.

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

2 files changed, 15 insertions, 0 deletions

diff --git a/qpid/cpp/src/qpid/broker/SemanticState.cpp b/qpid/cpp/src/qpid/broker/SemanticState.cpp
index 7339150642..8651b9034c 100644
--- a/qpid/cpp/src/qpid/broker/SemanticState.cpp
+++ b/qpid/cpp/src/qpid/broker/SemanticState.cpp
@@ -185,10 +185,23 @@ void SemanticState::suspendDtx(const std::string& xid)
 
     checkDtxTimeout();
     dtxBuffer->setSuspended(true);
+    suspendedXids[xid] = dtxBuffer;
+    dtxBuffer.reset();
 }
 
 void SemanticState::resumeDtx(const std::string& xid)
 {
+    if (!dtxSelected) {
+        throw CommandInvalidException(QPID_MSG("Session has not been selected for use with dtx"));
+    }
+
+    dtxBuffer = suspendedXids[xid];
+    if (!dtxBuffer) {
+        throw CommandInvalidException(QPID_MSG("xid " << xid << " not attached"));
+    } else {
+        suspendedXids.erase(xid);
+    }
+
     if (dtxBuffer->getXid() != xid) {
         throw CommandInvalidException(
             QPID_MSG("xid specified on start was " << dtxBuffer->getXid() << ", but " << xid << " specified on resume"));
diff --git a/qpid/cpp/src/qpid/broker/SemanticState.h b/qpid/cpp/src/qpid/broker/SemanticState.h
index 87ae937cfb..bb126287a3 100644
--- a/qpid/cpp/src/qpid/broker/SemanticState.h
+++ b/qpid/cpp/src/qpid/broker/SemanticState.h
@@ -103,6 +103,7 @@ class SemanticState : public framing::FrameHandler::Chains,
     };
 
     typedef std::map<std::string,ConsumerImpl::shared_ptr> ConsumerImplMap;
+    typedef std::map<std::string, DtxBuffer::shared_ptr> DtxBufferMap;
 
     SessionState& session;
     DeliveryAdapter& deliveryAdapter;
@@ -117,6 +118,7 @@ class SemanticState : public framing::FrameHandler::Chains,
     TxBuffer::shared_ptr txBuffer;
     DtxBuffer::shared_ptr dtxBuffer;
     bool dtxSelected;
+    DtxBufferMap suspendedXids;
     framing::AccumulatedAck accumulatedAck;
     bool flowActive;