Allow xid to be associated with publication and acknowledgements.

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

7 files changed, 19 insertions, 11 deletions

diff --git a/cpp/lib/broker/BrokerChannel.cpp b/cpp/lib/broker/BrokerChannel.cpp
index e5b8336b25..8805b72774 100644
--- a/cpp/lib/broker/BrokerChannel.cpp
+++ b/cpp/lib/broker/BrokerChannel.cpp
@@ -204,7 +204,7 @@ void Channel::ack(u_int64_t deliveryTag, bool multiple){
             throw InvalidAckException();
         }else if(multiple){     
             ack_iterator end = ++i;
-            for_each(unacked.begin(), end, bind2nd(mem_fun_ref(&DeliveryRecord::discard), static_cast<qpid::broker::TransactionContext*>(0)));
+            for_each(unacked.begin(), end, mem_fun_ref(&DeliveryRecord::discard));
             unacked.erase(unacked.begin(), end);
 
             //recalculate the prefetch:
diff --git a/cpp/lib/broker/DeliveryRecord.cpp b/cpp/lib/broker/DeliveryRecord.cpp
index 9d02cb615e..19b01cc312 100644
--- a/cpp/lib/broker/DeliveryRecord.cpp
+++ b/cpp/lib/broker/DeliveryRecord.cpp
@@ -42,8 +42,12 @@ DeliveryRecord::DeliveryRecord(Message::shared_ptr _msg,
                                                                pull(true){}
 
 
-void DeliveryRecord::discard(TransactionContext* ctxt) const{
-    queue->dequeue(ctxt, msg, 0);
+void DeliveryRecord::discard(TransactionContext* ctxt, const std::string* const xid) const{
+    queue->dequeue(ctxt, msg, xid);
+}
+
+void DeliveryRecord::discard() const{
+    discard(0, 0);
 }
 
 bool DeliveryRecord::matches(u_int64_t tag) const{
diff --git a/cpp/lib/broker/DeliveryRecord.h b/cpp/lib/broker/DeliveryRecord.h
index c1c8d6d13c..01a4024b28 100644
--- a/cpp/lib/broker/DeliveryRecord.h
+++ b/cpp/lib/broker/DeliveryRecord.h
@@ -46,7 +46,8 @@ namespace qpid {
             DeliveryRecord(Message::shared_ptr msg, Queue::shared_ptr queue, const std::string consumerTag, const u_int64_t deliveryTag);
             DeliveryRecord(Message::shared_ptr msg, Queue::shared_ptr queue, const u_int64_t deliveryTag);
             
-            void discard(TransactionContext* ctxt = 0) const;
+            void discard() const;
+            void discard(TransactionContext* ctxt, const std::string* const xid) const;
             bool matches(u_int64_t tag) const;
             bool coveredBy(const AccumulatedAck* const range) const;
             void requeue() const;
diff --git a/cpp/lib/broker/TxAck.cpp b/cpp/lib/broker/TxAck.cpp
index 2b55b81c58..b5211158f3 100644
--- a/cpp/lib/broker/TxAck.cpp
+++ b/cpp/lib/broker/TxAck.cpp
@@ -25,7 +25,8 @@ using std::bind2nd;
 using std::mem_fun_ref;
 using namespace qpid::broker;
 
-TxAck::TxAck(AccumulatedAck& _acked, std::list<DeliveryRecord>& _unacked) : acked(_acked), unacked(_unacked){
+TxAck::TxAck(AccumulatedAck& _acked, std::list<DeliveryRecord>& _unacked, const std::string* const _xid) : 
+    acked(_acked), unacked(_unacked), xid(_xid){
 
 }
 
@@ -34,10 +35,9 @@ bool TxAck::prepare(TransactionContext* ctxt) throw(){
         //dequeue all acked messages from their queues
         for (ack_iterator i = unacked.begin(); i != unacked.end(); i++) {
             if (i->coveredBy(&acked)) {
-                i->discard(ctxt);
+                i->discard(ctxt, xid);
             }
         }
-        //for_each(unacked.begin(), unacked.end(), bind2nd(mem_fun_ref(&DeliveryRecord::discardIfCoveredBy), &acked));
         return true;
     }catch(...){
         std::cout << "TxAck::prepare() - Failed to prepare" << std::endl;
diff --git a/cpp/lib/broker/TxAck.h b/cpp/lib/broker/TxAck.h
index d6ff8fea9c..88c321c445 100644
--- a/cpp/lib/broker/TxAck.h
+++ b/cpp/lib/broker/TxAck.h
@@ -37,13 +37,15 @@ namespace qpid {
         class TxAck : public TxOp{
             AccumulatedAck& acked;
             std::list<DeliveryRecord>& unacked;
+            const std::string* const xid;
+
         public:
             /**
              * @param acked a representation of the accumulation of
              * acks received
              * @param unacked the record of delivered messages
              */
-            TxAck(AccumulatedAck& acked, std::list<DeliveryRecord>& unacked);
+            TxAck(AccumulatedAck& acked, std::list<DeliveryRecord>& unacked, const std::string* const xid = 0);
             virtual bool prepare(TransactionContext* ctxt) throw();
             virtual void commit() throw();
             virtual void rollback() throw();
diff --git a/cpp/lib/broker/TxPublish.cpp b/cpp/lib/broker/TxPublish.cpp
index 0de5fbb200..49dd8abd89 100644
--- a/cpp/lib/broker/TxPublish.cpp
+++ b/cpp/lib/broker/TxPublish.cpp
@@ -22,11 +22,11 @@
 
 using namespace qpid::broker;
 
-TxPublish::TxPublish(Message::shared_ptr _msg) : msg(_msg) {}
+TxPublish::TxPublish(Message::shared_ptr _msg, const std::string* const _xid) : msg(_msg), xid(_xid) {}
 
 bool TxPublish::prepare(TransactionContext* ctxt) throw(){
     try{
-        for_each(queues.begin(), queues.end(), Prepare(ctxt, msg, 0));
+        for_each(queues.begin(), queues.end(), Prepare(ctxt, msg, xid));
         return true;
     }catch(...){
         std::cout << "TxPublish::prepare() - Failed to prepare" << std::endl;
diff --git a/cpp/lib/broker/TxPublish.h b/cpp/lib/broker/TxPublish.h
index 2756addab7..75f201257e 100644
--- a/cpp/lib/broker/TxPublish.h
+++ b/cpp/lib/broker/TxPublish.h
@@ -60,10 +60,11 @@ namespace qpid {
             };
 
             Message::shared_ptr msg;
+            const std::string* const xid;
             std::list<Queue::shared_ptr> queues;
 
         public:
-            TxPublish(Message::shared_ptr msg);
+            TxPublish(Message::shared_ptr msg, const std::string* const xid = 0);
             virtual bool prepare(TransactionContext* ctxt) throw();
             virtual void commit() throw();
             virtual void rollback() throw();