- 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

4 files changed, 55 insertions, 1 deletions

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 {
diff --git a/cpp/src/tests/BrokerChannelTest.cpp b/cpp/src/tests/BrokerChannelTest.cpp
index 6c9b910637..3253a3d27a 100644
--- a/cpp/src/tests/BrokerChannelTest.cpp
+++ b/cpp/src/tests/BrokerChannelTest.cpp
@@ -75,6 +75,7 @@ class BrokerChannelTest : public CppUnit::TestCase
     CPPUNIT_TEST(testStaging);
     CPPUNIT_TEST(testQueuePolicy);
     CPPUNIT_TEST(testFlow);
+    CPPUNIT_TEST(testAsyncMesgToMoreThanOneQueue);
     CPPUNIT_TEST_SUITE_END();
 
     shared_ptr<Broker> broker;
@@ -307,6 +308,38 @@ class BrokerChannelTest : public CppUnit::TestCase
         store.check();
     }
 
+
+    //NOTE: message or queue test,
+    //but as it can usefully use the same utility classes as this
+    //class it is defined here for simpllicity
+    void testAsyncMesgToMoreThanOneQueue()
+    {
+        MockMessageStore store;
+        {//must ensure that store is last thing deleted
+        const string data1("abcd");
+        Message::shared_ptr msg1(createMessage("e", "A", "MsgA", data1.size()));
+        addContent(msg1, data1);
+ 
+        Queue::shared_ptr queue1(new Queue("my_queue1", false, &store, 0));
+        Queue::shared_ptr queue2(new Queue("my_queue2", false, &store, 0));
+        Queue::shared_ptr queue3(new Queue("my_queue3", false, &store, 0));
+        queue1->deliver(msg1);
+        queue2->deliver(msg1);
+        queue3->deliver(msg1);
+	sleep(2);
+        
+        Message::shared_ptr next = queue1->dequeue();
+        CPPUNIT_ASSERT_EQUAL(msg1, next);
+        next = queue2->dequeue();
+        CPPUNIT_ASSERT_EQUAL(msg1, next);
+        next = queue3->dequeue();
+        CPPUNIT_ASSERT_EQUAL(msg1, next);
+
+        }
+    }
+
+
+
     void testFlow(){
         Channel channel(connection, recorder, 7);
         channel.open();