QPID-5084: Track durable queue ownership across restarts; a patch from Pavel Moravec and Ernie Allen

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

4 files changed, 27 insertions, 0 deletions

diff --git a/qpid/cpp/src/qpid/broker/Broker.cpp b/qpid/cpp/src/qpid/broker/Broker.cpp
index 85900f8061..80e08dd25c 100644
--- a/qpid/cpp/src/qpid/broker/Broker.cpp
+++ b/qpid/cpp/src/qpid/broker/Broker.cpp
@@ -355,6 +355,11 @@ Broker::Broker(const Broker::Options& conf) :
     //recover any objects via object factories
     objects.restore(*this);
 
+    // Assign to queues their users who created them (can be done after ACL is loaded in Plugin::initializeAll above
+    if ((getAcl()) && (store.get())) {
+        queues.eachQueue(boost::bind(&qpid::broker::Queue::updateAclUserQueueCount, _1));
+    }
+
     if(conf.enableMgmt) {
         if (getAcl()) {
             mgmtObject->set_maxConns(getAcl()->getMaxConnectTotal());
diff --git a/qpid/cpp/src/qpid/broker/Queue.cpp b/qpid/cpp/src/qpid/broker/Queue.cpp
index 5a43df44a4..d9d6781c9f 100644
--- a/qpid/cpp/src/qpid/broker/Queue.cpp
+++ b/qpid/cpp/src/qpid/broker/Queue.cpp
@@ -1182,18 +1182,27 @@ void Queue::encode(Buffer& buffer) const
     buffer.putShortString(name);
     buffer.put(encodableSettings);
     buffer.putShortString(alternateExchange.get() ? alternateExchange->getName() : std::string(""));
+    buffer.putShortString(userId);
 }
 
 uint32_t Queue::encodedSize() const
 {
     return name.size() + 1/*short string size octet*/
         + (alternateExchange.get() ? alternateExchange->getName().size() : 0) + 1 /* short string */
+        + userId.size() + 1 /* short string */
         + encodableSettings.encodedSize();
 }
 
+void Queue::updateAclUserQueueCount()
+{
+  if (broker->getAcl())
+    broker->getAcl()->approveCreateQueue(userId, name);
+}
+
 Queue::shared_ptr Queue::restore( QueueRegistry& queues, Buffer& buffer )
 {
     string name;
+    string _userId;
     buffer.getShortString(name);
     FieldTable ft;
     buffer.get(ft);
@@ -1207,6 +1216,12 @@ Queue::shared_ptr Queue::restore( QueueRegistry& queues, Buffer& buffer )
         result.first->alternateExchangeName.assign(altExch);
     }
 
+    //get userId of queue's creator; ACL counters for userId are done after ACL plugin is initialized
+    if (buffer.available()) {
+        buffer.getShortString(_userId);
+        result.first->setOwningUser(_userId);
+    }
+
     return result.first;
 }
 
diff --git a/qpid/cpp/src/qpid/broker/Queue.h b/qpid/cpp/src/qpid/broker/Queue.h
index a832b95feb..a7eb71c6bb 100644
--- a/qpid/cpp/src/qpid/broker/Queue.h
+++ b/qpid/cpp/src/qpid/broker/Queue.h
@@ -204,6 +204,7 @@ class Queue : public boost::enable_shared_from_this<Queue>,
     QueueDepth current;
     QueueBindings bindings;
     std::string alternateExchangeName;
+    std::string userId; // queue owner for ACL quota purposes
     boost::shared_ptr<Exchange> alternateExchange;
     framing::SequenceNumber sequence;
     qmf::org::apache::qpid::broker::Queue::shared_ptr mgmtObject;
@@ -384,6 +385,10 @@ class Queue : public boost::enable_shared_from_this<Queue>,
     /** Get the message at position pos, returns true if found and sets msg */
     QPID_BROKER_EXTERN bool find(framing::SequenceNumber pos, Message& msg ) const;
 
+    // Remember the queue's owner so acl quotas can be restored after restart
+    void setOwningUser(std::string& _userId) { userId  = _userId; }
+    void updateAclUserQueueCount();
+
     QPID_BROKER_EXTERN void setAlternateExchange(boost::shared_ptr<Exchange> exchange);
     QPID_BROKER_EXTERN boost::shared_ptr<Exchange> getAlternateExchange();
     QPID_BROKER_EXTERN bool isLocal(const Message& msg);
diff --git a/qpid/cpp/src/qpid/broker/QueueRegistry.cpp b/qpid/cpp/src/qpid/broker/QueueRegistry.cpp
index 606a8cceae..1283a42e6d 100644
--- a/qpid/cpp/src/qpid/broker/QueueRegistry.cpp
+++ b/qpid/cpp/src/qpid/broker/QueueRegistry.cpp
@@ -64,6 +64,8 @@ QueueRegistry::declare(const string& name, const QueueSettings& settings,
             //Move this to factory also?
             if (alternate)
                 queue->setAlternateExchange(alternate);//need to do this *before* create
+            queue->setOwningUser(userId);
+
             if (!recovering) {
                 //create persistent record if required
                 queue->create();