Was causing cluster failures.

QPID-1488 Mick Goulish:  QueuePolicy serialization fix for cluster braindump.


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

4 files changed, 42 insertions, 5 deletions

diff --git a/cpp/src/qpid/broker/Queue.cpp b/cpp/src/qpid/broker/Queue.cpp
index 9f722cf8be..b1f9163bb5 100644
--- a/cpp/src/qpid/broker/Queue.cpp
+++ b/cpp/src/qpid/broker/Queue.cpp
@@ -741,11 +741,12 @@ void Queue::encode(Buffer& buffer) const
 {
     buffer.putShortString(name);
     buffer.put(settings);
+    buffer.put(*policy);
 }
 
 uint32_t Queue::encodedSize() const
 {
-    return name.size() + 1/*short string size octet*/ + settings.encodedSize();
+    return name.size() + 1/*short string size octet*/ + settings.encodedSize() + (*policy).encodedSize();
 }
 
 Queue::shared_ptr Queue::decode(QueueRegistry& queues, Buffer& buffer)
@@ -755,6 +756,7 @@ Queue::shared_ptr Queue::decode(QueueRegistry& queues, Buffer& buffer)
     std::pair<Queue::shared_ptr, bool> result = queues.declare(name, true);
     buffer.get(result.first->settings);
     result.first->configure(result.first->settings);
+    buffer.get ( *(result.first->policy) );
     return result.first;
 }
 
diff --git a/cpp/src/qpid/broker/QueuePolicy.cpp b/cpp/src/qpid/broker/QueuePolicy.cpp
index c967823ecc..5c945d2c7f 100644
--- a/cpp/src/qpid/broker/QueuePolicy.cpp
+++ b/cpp/src/qpid/broker/QueuePolicy.cpp
@@ -121,6 +121,36 @@ void QueuePolicy::setDefaultMaxSize(uint64_t s)
     defaultMaxSize = s;
 }
 
+
+
+
+
+void QueuePolicy::encode(Buffer& buffer) const
+{
+  buffer.putLong(maxCount);
+  buffer.putLongLong(maxSize);
+  buffer.putLong(count.get());
+  buffer.putLongLong(size.get());
+}
+
+void QueuePolicy::decode ( Buffer& buffer ) 
+{
+  maxCount = buffer.getLong();
+  maxSize  = buffer.getLongLong();
+  count    = buffer.getLong();
+  size     = buffer.getLongLong();
+}
+
+
+uint32_t QueuePolicy::encodedSize() const {
+  return sizeof(uint32_t) +  // maxCount
+         sizeof(uint64_t) +  // maxSize
+         sizeof(uint32_t) +  // count
+         sizeof(uint64_t);   // size
+}
+
+
+
 const std::string QueuePolicy::maxCountKey("qpid.max_count");
 const std::string QueuePolicy::maxSizeKey("qpid.max_size");
 const std::string QueuePolicy::typeKey("qpid.policy_type");
@@ -231,8 +261,7 @@ std::auto_ptr<QueuePolicy> QueuePolicy::createQueuePolicy(uint32_t maxCount, uin
     }
 
 }
-
-
+ 
 namespace qpid {
     namespace broker {
 
diff --git a/cpp/src/qpid/broker/QueuePolicy.h b/cpp/src/qpid/broker/QueuePolicy.h
index d39ce7dc11..0e8c15aa0e 100644
--- a/cpp/src/qpid/broker/QueuePolicy.h
+++ b/cpp/src/qpid/broker/QueuePolicy.h
@@ -36,8 +36,8 @@ class QueuePolicy
 {
     static uint64_t defaultMaxSize;
 
-    const uint32_t maxCount;
-    const uint64_t maxSize;
+    uint32_t maxCount;
+    uint64_t maxSize;
     const std::string type;
     qpid::sys::AtomicValue<uint32_t> count;
     qpid::sys::AtomicValue<uint64_t> size;
@@ -63,6 +63,10 @@ class QueuePolicy
     void update(qpid::framing::FieldTable& settings);
     uint32_t getMaxCount() const { return maxCount; }
     uint64_t getMaxSize() const { return maxSize; }           
+    void encode(framing::Buffer& buffer) const;
+    void decode ( framing::Buffer& buffer );
+    uint32_t encodedSize() const;
+
 
     static std::auto_ptr<QueuePolicy> createQueuePolicy(const qpid::framing::FieldTable& settings);
     static std::auto_ptr<QueuePolicy> createQueuePolicy(uint32_t maxCount, uint64_t maxSize, const std::string& type = REJECT);
diff --git a/cpp/src/qpid/sys/AtomicValue_mutex.h b/cpp/src/qpid/sys/AtomicValue_mutex.h
index 8871addbda..e4d433e7f5 100644
--- a/cpp/src/qpid/sys/AtomicValue_mutex.h
+++ b/cpp/src/qpid/sys/AtomicValue_mutex.h
@@ -53,6 +53,8 @@ class AtomicValue
     inline T operator++(int) { return fetchAndAdd(1); }
     inline T operator--(int) { return fetchAndSub(1); }
 
+    AtomicValue& operator=(T newval) { Lock l(lock); value = newval; return *this; }
+
     /** If current value == testval then set to newval. Returns the old value. */
     T valueCompareAndSwap(T testval, T newval) {
         Lock l(lock);