Added durability property to queues and pass this to broker on declare. (This change also applied on trunk)

Minor update of accumulated ack and test.



git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/qpid.0-9@507622 13f79535-47bb-0310-9956-ffa450edef68

5 files changed, 63 insertions, 23 deletions

diff --git a/cpp/lib/broker/AccumulatedAck.cpp b/cpp/lib/broker/AccumulatedAck.cpp
index 5f66adb2b9..34547224ec 100644
--- a/cpp/lib/broker/AccumulatedAck.cpp
+++ b/cpp/lib/broker/AccumulatedAck.cpp
@@ -27,12 +27,12 @@ using std::bind2nd;
 using namespace qpid::broker;
 
 void AccumulatedAck::update(u_int64_t firstTag, u_int64_t lastTag){
-	assert(firstTag<=lastTag);
-    if (firstTag <= range+1) {
-    	range = lastTag;
+    assert(firstTag<=lastTag);
+    if (firstTag <= range + 1) {
+        if (lastTag > range) range = lastTag;
     } else {
     	for (u_int64_t tag = firstTag; tag<=lastTag; tag++)
-        	individual.push_back(tag);
+            individual.push_back(tag);
     }
 }
 
@@ -40,6 +40,11 @@ void AccumulatedAck::consolidate(){
     individual.sort();
     //remove any individual tags that are covered by range
     individual.remove_if(bind2nd(less_equal<u_int64_t>(), range));
+    //update range if possible (using <= allows for duplicates from overlapping ranges)
+    while (individual.front() <= range + 1) {
+        range = individual.front();
+        individual.pop_front();
+    }
 }
 
 void AccumulatedAck::clear(){
diff --git a/cpp/lib/client/ClientChannel.cpp b/cpp/lib/client/ClientChannel.cpp
index 613469c4ba..9d1c8ec011 100644
--- a/cpp/lib/client/ClientChannel.cpp
+++ b/cpp/lib/client/ClientChannel.cpp
@@ -159,7 +159,7 @@ void Channel::declareQueue(Queue& queue, bool synch){
     sendAndReceiveSync<QueueDeclareOkBody>(
         synch,
         new QueueDeclareBody(
-            version, 0, name, false, false, 
+            version, 0, name, false/*passive*/, queue.isDurable(),
             queue.isExclusive(), queue.isAutoDelete(), !synch, args));        
     if (synch) {
         if(queue.getName().length() == 0){
diff --git a/cpp/lib/client/ClientQueue.cpp b/cpp/lib/client/ClientQueue.cpp
index 455bb64d0b..773be504d8 100644
--- a/cpp/lib/client/ClientQueue.cpp
+++ b/cpp/lib/client/ClientQueue.cpp
@@ -20,14 +20,14 @@
  */
 #include <ClientQueue.h>
 
-qpid::client::Queue::Queue() : name(""), autodelete(true), exclusive(true){}
+qpid::client::Queue::Queue() : name(""), autodelete(true), exclusive(true), durable(false){}
 
-qpid::client::Queue::Queue(std::string _name) : name(_name), autodelete(false), exclusive(false){}
+qpid::client::Queue::Queue(std::string _name) : name(_name), autodelete(false), exclusive(false), durable(false){}
 
-qpid::client::Queue::Queue(std::string _name, bool temp) : name(_name), autodelete(temp), exclusive(temp){}
+qpid::client::Queue::Queue(std::string _name, bool temp) : name(_name), autodelete(temp), exclusive(temp), durable(false){}
 
-qpid::client::Queue::Queue(std::string _name, bool _autodelete, bool _exclusive) 
-  : name(_name), autodelete(_autodelete), exclusive(_exclusive){}
+qpid::client::Queue::Queue(std::string _name, bool _autodelete, bool _exclusive, bool _durable) 
+    : name(_name), autodelete(_autodelete), exclusive(_exclusive), durable(_durable){}
 
 const std::string& qpid::client::Queue::getName() const{
     return name;
@@ -45,6 +45,14 @@ bool qpid::client::Queue::isExclusive() const{
     return exclusive;
 }
 
+bool qpid::client::Queue::isDurable() const{
+    return durable;
+}
+
+void qpid::client::Queue::setDurable(bool _durable){
+    durable = _durable;
+}
+
 
 
 
diff --git a/cpp/lib/client/ClientQueue.h b/cpp/lib/client/ClientQueue.h
index 6b7aaf7c41..b37a44b004 100644
--- a/cpp/lib/client/ClientQueue.h
+++ b/cpp/lib/client/ClientQueue.h
@@ -55,24 +55,25 @@ namespace client {
 	std::string name;
         const bool autodelete;
         const bool exclusive;
+        bool durable;
 
     public:
 
         /**
-         * Creates an unnamed, temporary queue. A name will be
-         * assigned to this queue instance by a call to
+         * Creates an unnamed, non-durable, temporary queue. A name
+         * will be assigned to this queue instance by a call to
          * Channel::declareQueue().
          */
 	Queue();
         /**
-         * Creates a shared queue with a given name, that will not be
-         * autodeleted.
+         * Creates a shared, non-durable, queue with a given name,
+         * that will not be autodeleted.
          * 
          * @param name the name of the queue
          */
 	Queue(std::string name);
         /**
-         * Creates a queue with a given name.
+         * Creates a non-durable queue with a given name.
          * 
          * @param name the name of the queue
          * 
@@ -81,17 +82,19 @@ namespace client {
          */
 	Queue(std::string name, bool temp);
         /**
-         * This constructor allows the autodelete and exclusive
-         * propeties to be explictly set. Note however that if
+         * This constructor allows the autodelete, exclusive and
+         * durable propeties to be explictly set. Note however that if
          * exclusive is true, autodelete has no meaning as exclusive
          * queues are always destroyed when the connection that
          * created them is closed.
          */
-	Queue(std::string name, bool autodelete, bool exclusive);
+	Queue(std::string name, bool autodelete, bool exclusive, bool durable);
 	const std::string& getName() const;
 	void setName(const std::string&);
         bool isAutoDelete() const;
         bool isExclusive() const;
+        bool isDurable() const;
+        void setDurable(bool durable);
     };
 
 }
diff --git a/cpp/tests/AccumulatedAckTest.cpp b/cpp/tests/AccumulatedAckTest.cpp
index da984d3e69..64c1f979c0 100644
--- a/cpp/tests/AccumulatedAckTest.cpp
+++ b/cpp/tests/AccumulatedAckTest.cpp
@@ -29,11 +29,38 @@ using namespace qpid::broker;
 class AccumulatedAckTest : public CppUnit::TestCase  
 {
         CPPUNIT_TEST_SUITE(AccumulatedAckTest);
+        CPPUNIT_TEST(testGeneral);
         CPPUNIT_TEST(testCovers);
         CPPUNIT_TEST(testUpdateAndConsolidate);
         CPPUNIT_TEST_SUITE_END();
 
     public:
+        void testGeneral()
+        {
+            AccumulatedAck ack;
+            ack.clear();
+            ack.update(3,3);
+            ack.update(7,7);
+            ack.update(9,9);
+            ack.update(1,2);
+            ack.update(4,5);
+            ack.update(6,6);
+
+            for(int i = 1; i <= 7; i++) CPPUNIT_ASSERT(ack.covers(i));
+            CPPUNIT_ASSERT(ack.covers(9));
+
+            CPPUNIT_ASSERT(!ack.covers(8));
+            CPPUNIT_ASSERT(!ack.covers(10));
+
+            ack.consolidate();
+
+            for(int i = 1; i <= 7; i++) CPPUNIT_ASSERT(ack.covers(i));
+            CPPUNIT_ASSERT(ack.covers(9));
+
+            CPPUNIT_ASSERT(!ack.covers(8));
+            CPPUNIT_ASSERT(!ack.covers(10));
+        }
+
         void testCovers()
         {
             AccumulatedAck ack;
@@ -67,15 +94,12 @@ class AccumulatedAckTest : public CppUnit::TestCase
             ack.update(2, 2);
             ack.update(0, 5);
             ack.consolidate();
-            CPPUNIT_ASSERT_EQUAL((u_int64_t) 5, ack.range);
-            CPPUNIT_ASSERT_EQUAL((size_t) 3, ack.individual.size());
+            CPPUNIT_ASSERT_EQUAL((u_int64_t) 6, ack.range);
+            CPPUNIT_ASSERT_EQUAL((size_t) 2, ack.individual.size());
             list<u_int64_t>::iterator i = ack.individual.begin();
-            CPPUNIT_ASSERT_EQUAL((u_int64_t) 6, *i);
-            i++;
             CPPUNIT_ASSERT_EQUAL((u_int64_t) 8, *i);
             i++;
             CPPUNIT_ASSERT_EQUAL((u_int64_t) 10, *i);
-
         }
 };