1 files changed, 52 insertions, 10 deletions

diff --git a/cpp/src/qpid/acl/Acl.cpp b/cpp/src/qpid/acl/Acl.cpp
index 89c4b3402a..31ad9a38ac 100644
--- a/cpp/src/qpid/acl/Acl.cpp
+++ b/cpp/src/qpid/acl/Acl.cpp
@@ -18,11 +18,13 @@
 
 #include "qpid/acl/Acl.h"
 #include "qpid/acl/AclConnectionCounter.h"
+#include "qpid/acl/AclResourceCounter.h"
 #include "qpid/acl/AclData.h"
 #include "qpid/acl/AclValidator.h"
 #include "qpid/sys/Mutex.h"
 
 #include "qpid/broker/Broker.h"
+#include "qpid/broker/Connection.h"
 #include "qpid/Plugin.h"
 #include "qpid/Options.h"
 #include "qpid/log/Logger.h"
@@ -32,6 +34,7 @@
 #include "qmf/org/apache/qpid/acl/Package.h"
 #include "qmf/org/apache/qpid/acl/EventAllow.h"
 #include "qmf/org/apache/qpid/acl/EventConnectionDeny.h"
+#include "qmf/org/apache/qpid/acl/EventQueueQuotaDeny.h"
 #include "qmf/org/apache/qpid/acl/EventDeny.h"
 #include "qmf/org/apache/qpid/acl/EventFileLoaded.h"
 #include "qmf/org/apache/qpid/acl/EventFileLoadFailed.h"
@@ -50,19 +53,29 @@ using qpid::management::Manageable;
 using qpid::management::Args;
 namespace _qmf = qmf::org::apache::qpid::acl;
 
-Acl::Acl (AclValues& av, Broker& b): aclValues(av), broker(&b), transferAcl(false), mgmtObject(0),
-    connectionCounter(new ConnectionCounter(*this, aclValues.aclMaxConnectPerUser, aclValues.aclMaxConnectPerIp, aclValues.aclMaxConnectTotal))
-{
+Acl::Acl (AclValues& av, Broker& b): aclValues(av), broker(&b), transferAcl(false),
+    connectionCounter(new ConnectionCounter(*this, aclValues.aclMaxConnectPerUser, aclValues.aclMaxConnectPerIp, aclValues.aclMaxConnectTotal)),
+    resourceCounter(new ResourceCounter(*this, aclValues.aclMaxQueuesPerUser)){
+
+    if (aclValues.aclMaxConnectPerUser > AclData::getConnectMaxSpec())
+        throw Exception("--connection-limit-per-user switch cannot be larger than " + AclData::getMaxConnectSpecStr());
+    if (aclValues.aclMaxConnectPerIp > AclData::getConnectMaxSpec())
+        throw Exception("--connection-limit-per-ip switch cannot be larger than " + AclData::getMaxConnectSpecStr());
+    if (aclValues.aclMaxConnectTotal > AclData::getConnectMaxSpec())
+        throw Exception("--max-connections switch cannot be larger than " + AclData::getMaxConnectSpecStr());
+    if (aclValues.aclMaxQueuesPerUser > AclData::getConnectMaxSpec())
+        throw Exception("--max-queues-per-user switch cannot be larger than " + AclData::getMaxConnectSpecStr());
 
     agent = broker->getManagementAgent();
 
     if (agent != 0){
         _qmf::Package  packageInit(agent);
-        mgmtObject = new _qmf::Acl (agent, this, broker);
+        mgmtObject = _qmf::Acl::shared_ptr(new _qmf::Acl (agent, this, broker));
         agent->addObject (mgmtObject);
         mgmtObject->set_maxConnections(aclValues.aclMaxConnectTotal);
         mgmtObject->set_maxConnectionsPerIp(aclValues.aclMaxConnectPerIp);
         mgmtObject->set_maxConnectionsPerUser(aclValues.aclMaxConnectPerUser);
+        mgmtObject->set_maxQueuesPerUser(aclValues.aclMaxQueuesPerUser);
     }
     std::string errorString;
     if (!readAclFile(errorString)){
@@ -84,6 +97,15 @@ void Acl::reportConnectLimit(const std::string user, const std::string addr)
 }
 
 
+void Acl::reportQueueLimit(const std::string user, const std::string queueName)
+{
+    if (mgmtObject!=0)
+        mgmtObject->inc_queueQuotaDenyCount();
+
+    agent->raiseEvent(_qmf::EventQueueQuotaDeny(user, queueName));
+}
+
+
 bool Acl::authorise(
     const std::string&               id,
     const Action&                    action,
@@ -126,13 +148,29 @@ bool Acl::authorise(
 
 bool Acl::approveConnection(const qpid::broker::Connection& conn)
 {
-    return connectionCounter->approveConnection(conn);
+    const std::string& userName(conn.getUserId());
+    uint16_t connectionLimit(0);
+
+    boost::shared_ptr<AclData> dataLocal;
+    {
+        Mutex::ScopedLock locker(dataLock);
+        dataLocal = data;  //rcu copy
+    }
+
+    bool enforcingConnQuotas = dataLocal->getConnQuotaForUser(userName, &connectionLimit);
+
+    return connectionCounter->approveConnection(conn, enforcingConnQuotas, connectionLimit);
+}
+
+bool Acl::approveCreateQueue(const std::string& userId, const std::string& queueName)
+{
+    return resourceCounter->approveCreateQueue(userId, queueName);
 }
 
 
-void Acl::setUserId(const qpid::broker::Connection& connection, const std::string& username)
+void Acl::recordDestroyQueue(const std::string& queueName)
 {
-    connectionCounter->setUserId(connection, username);
+    resourceCounter->recordDestroyQueue(queueName);
 }
 
 
@@ -190,7 +228,7 @@ bool Acl::readAclFile(std::string& errorText)
 
 bool Acl::readAclFile(std::string& aclFile, std::string& errorText) {
     boost::shared_ptr<AclData> d(new AclData);
-    AclReader ar;
+    AclReader ar(aclValues.aclMaxConnectPerUser);
     if (ar.read(aclFile, d)){
         agent->raiseEvent(_qmf::EventFileLoadFailed("", ar.getError()));
         errorText = ar.getError();
@@ -211,6 +249,10 @@ bool Acl::readAclFile(std::string& aclFile, std::string& errorText) {
         QPID_LOG(debug,"ACL: Transfer ACL is Enabled!");
     }
 
+    if (data->enforcingConnectionQuotas()){
+        QPID_LOG(debug, "ACL: Connection quotas are Enabled.");
+    }
+
     data->aclSource = aclFile;
     if (mgmtObject!=0){
         mgmtObject->set_transferAcl(transferAcl?1:0);
@@ -300,9 +342,9 @@ Acl::~Acl(){
     broker->getConnectionObservers().remove(connectionCounter);
 }
 
-ManagementObject* Acl::GetManagementObject(void) const
+ManagementObject::shared_ptr Acl::GetManagementObject(void) const
 {
-    return (ManagementObject*) mgmtObject;
+    return mgmtObject;
 }
 
 Manageable::status_t Acl::ManagementMethod (uint32_t methodId, Args& args, string& text)