From 8540702068eda8d37c40992ce6d70e9ab4af3237 Mon Sep 17 00:00:00 2001
From: "Carl C. Trieloff" <cctrieloff@apache.org>
Date: Tue, 21 Aug 2007 19:11:07 +0000
Subject: - header correction for buffer - added tests for one msg to multiple
 queue in async mode - added counter for async multiple msg

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@568236 13f79535-47bb-0310-9956-ffa450edef68
---
 cpp/src/qpid/broker/BrokerQueue.cpp      |  1 +
 cpp/src/qpid/broker/PersistableMessage.h | 21 ++++++++++++++++++++-
 cpp/src/qpid/framing/StructHelper.h      |  1 +
 3 files changed, 22 insertions(+), 1 deletion(-)

(limited to 'cpp/src/qpid')

diff --git a/cpp/src/qpid/broker/BrokerQueue.cpp b/cpp/src/qpid/broker/BrokerQueue.cpp
index 5567131c19..706179fb52 100644
--- a/cpp/src/qpid/broker/BrokerQueue.cpp
+++ b/cpp/src/qpid/broker/BrokerQueue.cpp
@@ -249,6 +249,7 @@ bool Queue::canAutoDelete() const{
 bool Queue::enqueue(TransactionContext* ctxt, Message::shared_ptr& msg)
 {
     if (msg->isPersistent() && store) {
+	msg->enqueueAsync(); //increment to async counter -- for message sent to more than one queue
         store->enqueue(ctxt, *msg.get(), *this);
 	return true;
     }
diff --git a/cpp/src/qpid/broker/PersistableMessage.h b/cpp/src/qpid/broker/PersistableMessage.h
index 72457c408f..e47ca0ae48 100644
--- a/cpp/src/qpid/broker/PersistableMessage.h
+++ b/cpp/src/qpid/broker/PersistableMessage.h
@@ -45,6 +45,16 @@ namespace broker {
     * stored.
     */
     bool enqueueCompleted;
+ 
+    /**
+    * Counts the number of times the message has been processed
+    * async - thus when it == 0 the broker knows it has ownership
+    * -> an async store can increment this counter if it writes a
+    * copy to each queue, and case use this counter to know when all
+    * the write are complete
+    */
+    int asyncCounter;
+
     /**
     * Needs to be set false on Message construction, then
     * set once the dequeueis complete, it gets set
@@ -64,10 +74,19 @@ public:
     virtual ~PersistableMessage() {};
     PersistableMessage():
     enqueueCompleted(false),
+    asyncCounter(0),
     dequeueCompleted(false){};
     
     inline bool isEnqueueComplete() {return enqueueCompleted;};
-    inline void enqueueComplete() {enqueueCompleted = true;};
+    inline void enqueueComplete() {
+        if (asyncCounter<=1) {
+     	    asyncCounter =0;
+	    enqueueCompleted = true; 
+        }else{
+	    asyncCounter--;
+	}
+     };
+    inline void enqueueAsync() {asyncCounter++;};
     inline bool isDequeueComplete() {return dequeueCompleted;};
     inline void dequeueComplete() {dequeueCompleted = true;};
     
diff --git a/cpp/src/qpid/framing/StructHelper.h b/cpp/src/qpid/framing/StructHelper.h
index b5d1b1e78c..dc23a30d58 100644
--- a/cpp/src/qpid/framing/StructHelper.h
+++ b/cpp/src/qpid/framing/StructHelper.h
@@ -22,6 +22,7 @@
 #define _StructHelper_
 
 #include "qpid/Exception.h"
+#include "Buffer.h"
 
 namespace qpid {
 namespace framing {
-- 
cgit v1.2.1