Update from trunk r1375509 through r1450773asyncstore

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

14 files changed, 729 insertions, 215 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)
diff --git a/cpp/src/qpid/acl/Acl.h b/cpp/src/qpid/acl/Acl.h
index 4787934275..ea3c6586a3 100644
--- a/cpp/src/qpid/acl/Acl.h
+++ b/cpp/src/qpid/acl/Acl.h
@@ -43,12 +43,14 @@ class Connection;
 
 namespace acl {
 class ConnectionCounter;
+class ResourceCounter;
 
 struct AclValues {
     std::string aclFile;
     uint16_t    aclMaxConnectPerUser;
     uint16_t    aclMaxConnectPerIp;
     uint16_t    aclMaxConnectTotal;
+    uint16_t    aclMaxQueuesPerUser;
 };
 
 
@@ -60,10 +62,11 @@ private:
     broker::Broker*                      broker;
     bool                                 transferAcl;
     boost::shared_ptr<AclData>           data;
-    qmf::org::apache::qpid::acl::Acl*    mgmtObject; // mgnt owns lifecycle
+    qmf::org::apache::qpid::acl::Acl::shared_ptr mgmtObject;
     qpid::management::ManagementAgent*   agent;
     mutable qpid::sys::Mutex             dataLock;
     boost::shared_ptr<ConnectionCounter> connectionCounter;
+    boost::shared_ptr<ResourceCounter>   resourceCounter;
 
 public:
     Acl (AclValues& av, broker::Broker& b);
@@ -72,11 +75,16 @@ public:
      * issue management counts and alerts for denied connections
      */
     void reportConnectLimit(const std::string user, const std::string addr);
+    void reportQueueLimit(const std::string user, const std::string queueName);
 
     inline virtual bool doTransferAcl() {
         return transferAcl;
     };
 
+    inline virtual uint16_t getMaxConnectTotal() {
+        return aclValues.aclMaxConnectTotal;
+    };
+
 // create specilied authorise methods for cases that need faster matching as needed.
     virtual bool authorise(
         const std::string&               id,
@@ -92,9 +100,10 @@ public:
         const std::string&               ExchangeName,
         const std::string&               RoutingKey);
 
+    // Resource quota tracking
     virtual bool approveConnection(const broker::Connection& connection);
-
-    virtual void setUserId(const broker::Connection& connection, const std::string& username);
+    virtual bool approveCreateQueue(const std::string& userId, const std::string& queueName);
+    virtual void recordDestroyQueue(const std::string& queueName);
 
     virtual ~Acl();
 private:
@@ -108,7 +117,7 @@ private:
     bool readAclFile(std::string& aclFile, std::string& errorText);
     Manageable::status_t lookup       (management::Args& args, std::string& text);
     Manageable::status_t lookupPublish(management::Args& args, std::string& text);
-    virtual qpid::management::ManagementObject* GetManagementObject(void) const;
+    virtual qpid::management::ManagementObject::shared_ptr GetManagementObject(void) const;
     virtual management::Manageable::status_t ManagementMethod (uint32_t methodId, management::Args& args, std::string& text);
 
 };
diff --git a/cpp/src/qpid/acl/AclConnectionCounter.cpp b/cpp/src/qpid/acl/AclConnectionCounter.cpp
index 8c6e3eef6e..875137bf55 100644
--- a/cpp/src/qpid/acl/AclConnectionCounter.cpp
+++ b/cpp/src/qpid/acl/AclConnectionCounter.cpp
@@ -85,32 +85,32 @@ bool ConnectionCounter::limitApproveLH(
 //
 // countConnectionLH
 //
-// Increment the name's count in map and return a comparison against the limit.
-// called with dataLock already taken
+// Increment the name's count in map and return an optional comparison
+//  against a connection limit.
+// Called with dataLock already taken.
 //
 bool ConnectionCounter::countConnectionLH(
     connectCountsMap_t& theMap,
     const std::string& theName,
     uint16_t theLimit,
-    bool emitLog) {
+    bool emitLog,
+    bool enforceLimit) {
 
     bool result(true);
     uint16_t count(0);
-    if (theLimit > 0) {
-        connectCountsMap_t::iterator eRef = theMap.find(theName);
-        if (eRef != theMap.end()) {
-            count = (uint16_t)(*eRef).second + 1;
-            (*eRef).second = count;
-            result = count <= theLimit;
-        } else {
-            theMap[theName] = count = 1;
-        }
-        if (emitLog) {
-            QPID_LOG(trace, "ACL ConnectionApprover user=" << theName
-                << " limit=" << theLimit
-                << " curValue=" << count
-                << " result=" << (result ? "allow" : "deny"));
-        }
+    connectCountsMap_t::iterator eRef = theMap.find(theName);
+    if (eRef != theMap.end()) {
+        count = (uint16_t)(*eRef).second + 1;
+        (*eRef).second = count;
+        result = (enforceLimit ? count <= theLimit : true);
+    } else {
+        theMap[theName] = count = 1;
+    }
+    if (emitLog) {
+        QPID_LOG(trace, "ACL ConnectionApprover user=" << theName
+            << " limit=" << theLimit
+            << " curValue=" << count
+            << " result=" << (result ? "allow" : "deny"));
     }
     return result;
 }
@@ -123,23 +123,21 @@ bool ConnectionCounter::countConnectionLH(
 // called with dataLock already taken
 //
 void ConnectionCounter::releaseLH(
-    connectCountsMap_t& theMap, const std::string& theName, uint16_t theLimit) {
-
-    if (theLimit > 0) {
-        connectCountsMap_t::iterator eRef = theMap.find(theName);
-        if (eRef != theMap.end()) {
-            uint16_t count = (uint16_t) (*eRef).second;
-            assert (count > 0);
-            if (1 == count) {
-                theMap.erase (eRef);
-            } else {
-                (*eRef).second = count - 1;
-            }
+    connectCountsMap_t& theMap, const std::string& theName) {
+
+    connectCountsMap_t::iterator eRef = theMap.find(theName);
+    if (eRef != theMap.end()) {
+        uint16_t count = (uint16_t) (*eRef).second;
+        assert (count > 0);
+        if (1 == count) {
+            theMap.erase (eRef);
         } else {
-            // User had no connections.
-            QPID_LOG(notice, "ACL ConnectionCounter Connection for '" << theName
-                << "' not found in connection count pool");
+            (*eRef).second = count - 1;
         }
+    } else {
+        // User had no connections.
+        QPID_LOG(notice, "ACL ConnectionCounter Connection for '" << theName
+            << "' not found in connection count pool");
     }
 }
 
@@ -161,7 +159,7 @@ void ConnectionCounter::connection(broker::Connection& connection) {
     connectProgressMap[connection.getMgmtId()] = C_CREATED;
 
     // Count the connection from this host.
-    (void) countConnectionLH(connectByHostMap, hostName, hostLimit, false);
+    (void) countConnectionLH(connectByHostMap, hostName, hostLimit, false, false);
 }
 
 
@@ -180,8 +178,7 @@ void ConnectionCounter::closed(broker::Connection& connection) {
             // Normal case: connection was created and opened.
             // Decrement user in-use counts
             releaseLH(connectByNameMap,
-                      connection.getUserId(),
-                      nameLimit);
+                      connection.getUserId());
         } else {
             // Connection was created but not opened.
             // Don't decrement user count.
@@ -189,8 +186,7 @@ void ConnectionCounter::closed(broker::Connection& connection) {
 
         // Decrement host in-use count.
         releaseLH(connectByHostMap,
-                  getClientHost(connection.getMgmtId()),
-                  hostLimit);
+                  getClientHost(connection.getMgmtId()));
 
         // destroy connection progress indicator
         connectProgressMap.erase(eRef);
@@ -211,7 +207,10 @@ void ConnectionCounter::closed(broker::Connection& connection) {
 //  check total connections, connections from IP, connections by user and
 //  disallow if over any limit
 //
-bool ConnectionCounter::approveConnection(const broker::Connection& connection)
+bool ConnectionCounter::approveConnection(
+        const broker::Connection& connection,
+        bool enforcingConnectionQuotas,
+        uint16_t connectionUserQuota )
 {
     const std::string& hostName(getClientHost(connection.getMgmtId()));
     const std::string& userName(              connection.getUserId());
@@ -220,122 +219,53 @@ bool ConnectionCounter::approveConnection(const broker::Connection& connection)
 
     // Bump state from CREATED to OPENED
     (void) countConnectionLH(connectProgressMap, connection.getMgmtId(),
-                             C_OPENED, false);
+                             C_OPENED, false, false);
 
     // Approve total connections
     bool okTotal  = true;
     if (totalLimit > 0) {
         okTotal = totalCurrentConnections <= totalLimit;
-        if (!connection.isShadow()) {
-            QPID_LOG(trace, "ACL ConnectionApprover totalLimit=" << totalLimit
-                << " curValue=" << totalCurrentConnections
-                << " result=" << (okTotal ? "allow" : "deny"));
-        }
+        QPID_LOG(trace, "ACL ConnectionApprover totalLimit=" << totalLimit
+                 << " curValue=" << totalCurrentConnections
+                 << " result=" << (okTotal ? "allow" : "deny"));
     }
 
     // Approve by IP host connections
-    bool okByIP   = limitApproveLH(connectByHostMap, hostName, hostLimit, !connection.isShadow());
+    bool okByIP   = limitApproveLH(connectByHostMap, hostName, hostLimit, true);
 
     // Count and Approve the connection by the user
-    bool okByUser = countConnectionLH(connectByNameMap, userName, nameLimit, !connection.isShadow());
-
-    if (!connection.isShadow()) {
-        // Emit separate log for each disapproval
-        if (!okTotal) {
-            QPID_LOG(error, "Client max total connection count limit of " << totalLimit
-                << " exceeded by '"
-                << connection.getMgmtId() << "', user: '"
-                << userName << "'. Connection refused");
-        }
-        if (!okByIP) {
-            QPID_LOG(error, "Client max per-host connection count limit of "
-                << hostLimit << " exceeded by '"
-                << connection.getMgmtId() << "', user: '"
-                << userName << "'. Connection refused.");
-        }
-        if (!okByUser) {
-            QPID_LOG(error, "Client max per-user connection count limit of "
-                << nameLimit << " exceeded by '"
-                << connection.getMgmtId() << "', user: '"
-                << userName << "'. Connection refused.");
-        }
-
-        // Count/Event once for each disapproval
-        bool result = okTotal && okByIP && okByUser;
-        if (!result) {
-            acl.reportConnectLimit(userName, hostName);
-        }
-
-        return result;
-    } else {
-        // Always allow shadow connections
-        if (!okTotal) {
-            QPID_LOG(warning, "Client max total connection count limit of " << totalLimit
-                << " exceeded by '"
-                << connection.getMgmtId() << "', user: '"
-                << userName << "' but still within tolerance. Cluster connection allowed");
-        }
-        if (!okByIP) {
-            QPID_LOG(warning, "Client max per-host connection count limit of "
-                << hostLimit << " exceeded by '"
-                << connection.getMgmtId() << "', user: '"
-                << userName << "' but still within tolerance. Cluster connection allowed");
-        }
-        if (!okByUser) {
-            QPID_LOG(warning, "Client max per-user connection count limit of "
-                << nameLimit << " exceeded by '"
-                << connection.getMgmtId() << "', user: '"
-                << userName << "' but still within tolerance. Cluster connection allowed");
-        }
-        if (okTotal && okByIP && okByUser) {
-            QPID_LOG(debug, "Cluster client connection: '"
-                << connection.getMgmtId() << "', user '"
-                <<  userName << "' allowed");
-        }
-        return true;
+    bool okByUser = countConnectionLH(connectByNameMap, userName,
+                                      connectionUserQuota, true,
+                                      enforcingConnectionQuotas);
+
+    // Emit separate log for each disapproval
+    if (!okTotal) {
+        QPID_LOG(error, "Client max total connection count limit of " << totalLimit
+                 << " exceeded by '"
+                 << connection.getMgmtId() << "', user: '"
+                 << userName << "'. Connection refused");
+    }
+    if (!okByIP) {
+        QPID_LOG(error, "Client max per-host connection count limit of "
+                 << hostLimit << " exceeded by '"
+                 << connection.getMgmtId() << "', user: '"
+                 << userName << "'. Connection refused.");
+    }
+    if (!okByUser) {
+        QPID_LOG(error, "Client max per-user connection count limit of "
+                 << connectionUserQuota << " exceeded by '"
+                 << connection.getMgmtId() << "', user: '"
+                 << userName << "'. Connection refused.");
     }
-}
-
-
-//
-// setUserId
-//  On cluster shadow connections, track a new user id for this connection.
-//
-void ConnectionCounter::setUserId(const broker::Connection& connection,
-                                  const std::string& username)
-{
-    Mutex::ScopedLock locker(dataLock);
 
-    connectCountsMap_t::iterator eRef = connectProgressMap.find(connection.getMgmtId());
-    if (eRef != connectProgressMap.end()) {
-        if ((*eRef).second == C_OPENED){
-            // Connection has been opened so that current user has been counted
-            if (connection.isShadow()) {
-                // This is a shadow connection and therefore receives userId changes
-                QPID_LOG(debug, "Changing User ID for cluster connection: "
-                    << connection.getMgmtId() << ", old user:'" << connection.getUserId()
-                    << "', new user:'" << username << "'");
-
-                // Decrement user in-use count for old userId
-                releaseLH(connectByNameMap,
-                        connection.getUserId(),
-                        nameLimit);
-                // Increment user in-use count for new userId
-                (void) countConnectionLH(connectByNameMap, username, nameLimit, false);
-            } else {
-                QPID_LOG(warning, "Changing User ID for non-cluster connections is not supported: "
-                    << connection.getMgmtId() << ", old user " << connection.getUserId()
-                    << ", new user " << username);
-            }
-        } else {
-            // connection exists  but has not been opened.
-            // setUserId is called in normal course. The user gets counted when connection is opened.
-        }
-    } else {
-        // Connection does not exist.
+    // Count/Event once for each disapproval
+    bool result = okTotal && okByIP && okByUser;
+    if (!result) {
+        acl.reportConnectLimit(userName, hostName);
     }
-}
 
+    return result;
+}
 
 //
 // getClientIp - given a connection's mgmtId return the client host part.
diff --git a/cpp/src/qpid/acl/AclConnectionCounter.h b/cpp/src/qpid/acl/AclConnectionCounter.h
index 54fa6933ff..e8ef35c1ba 100644
--- a/cpp/src/qpid/acl/AclConnectionCounter.h
+++ b/cpp/src/qpid/acl/AclConnectionCounter.h
@@ -77,12 +77,12 @@ private:
     bool countConnectionLH(connectCountsMap_t& theMap,
                            const std::string& theName,
                            uint16_t theLimit,
-                           bool emitLog);
+                           bool emitLog,
+                           bool enforceLimit);
 
     /** Release a connection */
     void releaseLH(connectCountsMap_t& theMap,
-                   const std::string& theName,
-                   uint16_t theLimit);
+                   const std::string& theName);
 
 public:
     ConnectionCounter(Acl& acl, uint16_t nl, uint16_t hl, uint16_t tl);
@@ -93,8 +93,9 @@ public:
     void     closed(broker::Connection& connection);
 
     // Connection counting
-    bool approveConnection(const broker::Connection& conn);
-    void setUserId(const broker::Connection& connection, const std::string& username);
+    bool approveConnection(const broker::Connection& conn,
+                           bool enforcingConnectionQuotas,
+                           uint16_t connectionLimit );
 };
 
 }} // namespace qpid::ha
diff --git a/cpp/src/qpid/acl/AclData.cpp b/cpp/src/qpid/acl/AclData.cpp
index 7c14d0985d..ca866ab7d3 100644
--- a/cpp/src/qpid/acl/AclData.cpp
+++ b/cpp/src/qpid/acl/AclData.cpp
@@ -25,11 +25,19 @@ namespace qpid {
 namespace acl {
 
     //
-    // Instantiate the substitution keyword string
+    // Instantiate the keyword strings
     //
-    const std::string AclData::USER_SUBSTITUTION_KEYWORD       = "${user}";
-    const std::string AclData::DOMAIN_SUBSTITUTION_KEYWORD     = "${domain}";
-    const std::string AclData::USERDOMAIN_SUBSTITUTION_KEYWORD = "${userdomain}";
+    const std::string AclData::ACL_KEYWORD_USER_SUBST        = "${user}";
+    const std::string AclData::ACL_KEYWORD_DOMAIN_SUBST      = "${domain}";
+    const std::string AclData::ACL_KEYWORD_USERDOMAIN_SUBST  = "${userdomain}";
+    const std::string AclData::ACL_KEYWORD_ALL               = "all";
+    const std::string AclData::ACL_KEYWORD_ACL               = "acl";
+    const std::string AclData::ACL_KEYWORD_GROUP             = "group";
+    const std::string AclData::ACL_KEYWORD_QUOTA             = "quota";
+    const std::string AclData::ACL_KEYWORD_QUOTA_CONNECTIONS = "connections";
+    const char        AclData::ACL_SYMBOL_WILDCARD           = '*';
+    const std::string AclData::ACL_KEYWORD_WILDCARD          = "*";
+    const char        AclData::ACL_SYMBOL_LINE_CONTINUATION  = '\\';
 
     //
     // constructor
@@ -37,7 +45,9 @@ namespace acl {
     AclData::AclData():
         decisionMode(qpid::acl::DENY),
         transferAcl(false),
-        aclSource("UNKNOWN")
+        aclSource("UNKNOWN"),
+        connQuotaRulesExist(false),
+        connQuotaRuleSettings(new quotaRuleSet)
     {
         for (unsigned int cnt=0; cnt< qpid::acl::ACTIONSIZE; cnt++)
         {
@@ -60,6 +70,9 @@ namespace acl {
             }
             delete[] actionList[cnt];
         }
+        transferAcl = false;
+        connQuotaRulesExist = false;
+        connQuotaRuleSettings->clear();
     }
 
 
@@ -73,7 +86,7 @@ namespace acl {
                             const std::string& lookupStr)
     {
         // allow wildcard on the end of rule strings...
-        if (ruleStr.data()[ruleStr.size()-1]=='*')
+        if (ruleStr.data()[ruleStr.size()-1]==ACL_SYMBOL_WILDCARD)
         {
             return ruleStr.compare(0,
                                    ruleStr.size()-1,
@@ -124,7 +137,7 @@ namespace acl {
 
             // If individual actorId not found then find a rule set for '*'.
             if (itrRule == actionList[action][objType]->end())
-                itrRule = actionList[action][objType]->find("*");
+                itrRule = actionList[action][objType]->find(ACL_KEYWORD_WILDCARD);
 
             if (itrRule != actionList[action][objType]->end())
             {
@@ -199,6 +212,16 @@ namespace acl {
                                     lookupParamItr = params->find(PROP_MAXQUEUESIZE);
                                     break;
 
+                                case acl::SPECPROP_MAXFILECOUNTUPPERLIMIT:
+                                case acl::SPECPROP_MAXFILECOUNTLOWERLIMIT:
+                                    lookupParamItr = params->find(PROP_MAXFILECOUNT);
+                                    break;
+
+                                case acl::SPECPROP_MAXFILESIZEUPPERLIMIT:
+                                case acl::SPECPROP_MAXFILESIZELOWERLIMIT:
+                                    lookupParamItr = params->find(PROP_MAXFILESIZE);
+                                    break;
+
                                 default:
                                     lookupParamItr = params->find((Property)rulePropMapItr->first);
                                     break;
@@ -222,6 +245,8 @@ namespace acl {
                                     {
                                     case acl::SPECPROP_MAXQUEUECOUNTUPPERLIMIT:
                                     case acl::SPECPROP_MAXQUEUESIZEUPPERLIMIT:
+                                    case acl::SPECPROP_MAXFILECOUNTUPPERLIMIT:
+                                    case acl::SPECPROP_MAXFILESIZEUPPERLIMIT:
                                         limitChecked &=
                                             compareIntMax(
                                                 rulePropMapItr->first,
@@ -231,6 +256,8 @@ namespace acl {
 
                                     case acl::SPECPROP_MAXQUEUECOUNTLOWERLIMIT:
                                     case acl::SPECPROP_MAXQUEUESIZELOWERLIMIT:
+                                    case acl::SPECPROP_MAXFILECOUNTLOWERLIMIT:
+                                    case acl::SPECPROP_MAXFILESIZELOWERLIMIT:
                                         limitChecked &=
                                             compareIntMin(
                                                 rulePropMapItr->first,
@@ -241,14 +268,31 @@ namespace acl {
                                     default:
                                         bool result;
                                         if ((SPECPROP_ALTERNATE  == rulePropMapItr->first && rsItr->ruleHasUserSub[PROP_ALTERNATE])  ||
-                                            (SPECPROP_ROUTINGKEY == rulePropMapItr->first && rsItr->ruleHasUserSub[PROP_ROUTINGKEY]) ||
                                             (SPECPROP_QUEUENAME  == rulePropMapItr->first && rsItr->ruleHasUserSub[PROP_QUEUENAME]))
                                         {
                                             // These properties are allowed to have username substitution
                                             std::string sName(rulePropMapItr->second);
                                             substituteUserId(sName, id);
                                             result = matchProp(sName, lookupParamItr->second);
-                                        } else {
+                                        }
+                                        else if (SPECPROP_ROUTINGKEY == rulePropMapItr->first)
+                                        {
+                                            // Routing key is allowed to have username substitution
+                                            // and it gets topic exchange matching
+                                            if (rsItr->ruleHasUserSub[PROP_ROUTINGKEY])
+                                            {
+                                                std::string sKey(lookupParamItr->second);
+                                                substituteKeywords(sKey, id);
+                                                result = rsItr->matchRoutingKey(sKey);
+                                            }
+                                            else
+                                            {
+                                                result = rsItr->matchRoutingKey(lookupParamItr->second);
+                                            }
+                                        }
+                                        else
+                                        {
+                                            // Rules without substitution
                                             result = matchProp(rulePropMapItr->second, lookupParamItr->second);
                                         }
 
@@ -359,7 +403,7 @@ namespace acl {
             AclData::actObjItr itrRule = actionList[action][objType]->find(id);
 
             if (itrRule == actionList[action][objType]->end())
-                itrRule = actionList[action][objType]->find("*");
+                itrRule = actionList[action][objType]->find(ACL_KEYWORD_WILDCARD);
 
             if (itrRule != actionList[action][objType]->end() )
             {
@@ -405,9 +449,9 @@ namespace acl {
 
                     if (match && rsItr->pubRoutingKeyInRule)
                     {
-                        if ((routingKey.find(USER_SUBSTITUTION_KEYWORD, 0)       != std::string::npos) ||
-                            (routingKey.find(DOMAIN_SUBSTITUTION_KEYWORD, 0)     != std::string::npos) ||
-                            (routingKey.find(USERDOMAIN_SUBSTITUTION_KEYWORD, 0) != std::string::npos))
+                        if ((routingKey.find(ACL_KEYWORD_USER_SUBST, 0)       != std::string::npos) ||
+                            (routingKey.find(ACL_KEYWORD_DOMAIN_SUBST, 0)     != std::string::npos) ||
+                            (routingKey.find(ACL_KEYWORD_USERDOMAIN_SUBST, 0) != std::string::npos))
                         {
                             // The user is not allowed to present a routing key with the substitution key in it
                             QPID_LOG(debug, "ACL: Rule: " << rsItr->rawRuleNum <<
@@ -458,6 +502,62 @@ namespace acl {
     }
 
 
+
+    //
+    //
+    //
+    void AclData::setConnQuotaRuleSettings (
+                bool rulesExist, boost::shared_ptr<quotaRuleSet> quotaPtr)
+    {
+        connQuotaRulesExist = rulesExist;
+        connQuotaRuleSettings = quotaPtr;
+    }
+
+
+    //
+    // getConnQuotaForUser
+    //
+    // Return the true or false value of connQuotaRulesExist,
+    //  indicating whether any kind of lookup was done or not.
+    //
+    // When lookups are performed return the result value of
+    //  1. The user's setting else
+    //  2. The 'all' user setting else
+    //  3. Zero
+    // When lookups are not performed then return a result value of Zero.
+    //
+    bool AclData::getConnQuotaForUser(const std::string& theUserName,
+                                      uint16_t* theResult) const {
+        if (connQuotaRulesExist) {
+            // look for this user explicitly
+            quotaRuleSetItr nameItr = (*connQuotaRuleSettings).find(theUserName);
+            if (nameItr != (*connQuotaRuleSettings).end()) {
+                QPID_LOG(trace, "ACL: Connection quota for user " << theUserName
+                    << " explicitly set to : " << (*nameItr).second);
+                *theResult = (*nameItr).second;
+            } else {
+                // Look for the 'all' user
+                nameItr = (*connQuotaRuleSettings).find(ACL_KEYWORD_ALL);
+                if (nameItr != (*connQuotaRuleSettings).end()) {
+                    QPID_LOG(trace, "ACL: Connection quota for user " << theUserName
+                        << " chosen through value for 'all' : " << (*nameItr).second);
+                    *theResult = (*nameItr).second;
+                } else {
+                    // Neither userName nor "all" found.
+                    QPID_LOG(trace, "ACL: Connection quota for user " << theUserName
+                        << " absent in quota settings. Return value : 0");
+                    *theResult = 0;
+                }
+            }
+        } else {
+            // Rules do not exist
+            QPID_LOG(trace, "ACL: Connection quota for user " << theUserName
+                << " unavailable; quota settings are not specified. Return value : 0");
+            *theResult = 0;
+        }
+        return connQuotaRulesExist;
+    }
+
     //
     //
     //
@@ -607,8 +707,8 @@ namespace acl {
     //   Given an Acl rule and an authenticated userId
     //   do the keyword substitutions on the rule.
     //
-    void AclData::AclData::substituteUserId(std::string& ruleString,
-                                            const std::string& userId)
+    void AclData::substituteUserId(std::string& ruleString,
+                                   const std::string& userId)
     {
         size_t locDomSeparator(0);
         std::string user("");
@@ -625,9 +725,9 @@ namespace acl {
             domain = normalizeUserId(userId.substr(locDomSeparator+1));
         }
 
-        substituteString(ruleString, USER_SUBSTITUTION_KEYWORD,       user);
-        substituteString(ruleString, DOMAIN_SUBSTITUTION_KEYWORD,     domain);
-        substituteString(ruleString, USERDOMAIN_SUBSTITUTION_KEYWORD, userdomain);
+        substituteString(ruleString, ACL_KEYWORD_USER_SUBST,       user);
+        substituteString(ruleString, ACL_KEYWORD_DOMAIN_SUBST,     domain);
+        substituteString(ruleString, ACL_KEYWORD_USERDOMAIN_SUBST, userdomain);
     }
 
 
@@ -640,8 +740,8 @@ namespace acl {
     //   topic key lookups where the keyword string proper is in the
     //   topic key search tree.
     //
-    void AclData::AclData::substituteKeywords(std::string& ruleString,
-                                              const std::string& userId)
+    void AclData::substituteKeywords(std::string& ruleString,
+                                     const std::string& userId)
     {
         size_t locDomSeparator(0);
         std::string user("");
@@ -658,8 +758,8 @@ namespace acl {
             domain = normalizeUserId(userId.substr(locDomSeparator+1));
         }
         std::string oRule(ruleString);
-        substituteString(ruleString, userdomain, USERDOMAIN_SUBSTITUTION_KEYWORD);
-        substituteString(ruleString, user,       USER_SUBSTITUTION_KEYWORD);
-        substituteString(ruleString, domain,     DOMAIN_SUBSTITUTION_KEYWORD);
+        substituteString(ruleString, userdomain, ACL_KEYWORD_USERDOMAIN_SUBST);
+        substituteString(ruleString, user,       ACL_KEYWORD_USER_SUBST);
+        substituteString(ruleString, domain,     ACL_KEYWORD_DOMAIN_SUBST);
     }
 }}
diff --git a/cpp/src/qpid/acl/AclData.h b/cpp/src/qpid/acl/AclData.h
index b4b13c44b6..43cb5193f5 100644
--- a/cpp/src/qpid/acl/AclData.h
+++ b/cpp/src/qpid/acl/AclData.h
@@ -111,6 +111,8 @@ public:
     typedef  std::map<std::string, ruleSet > actionObject; // user
     typedef  actionObject::iterator          actObjItr;
     typedef  actionObject*                   aclAction;
+    typedef  std::map<std::string, uint16_t> quotaRuleSet; // <username, N>
+    typedef  quotaRuleSet::const_iterator    quotaRuleSetItr;
 
     // Action*[] -> Object*[] -> map<user -> set<Rule> >
     aclAction*           actionList[qpid::acl::ACTIONSIZE];
@@ -134,9 +136,18 @@ public:
 
     bool matchProp(const std::string & src, const std::string& src1);
     void clear ();
-    static const std::string USER_SUBSTITUTION_KEYWORD;
-    static const std::string DOMAIN_SUBSTITUTION_KEYWORD;
-    static const std::string USERDOMAIN_SUBSTITUTION_KEYWORD;
+    static const std::string ACL_KEYWORD_USER_SUBST;
+    static const std::string ACL_KEYWORD_DOMAIN_SUBST;
+    static const std::string ACL_KEYWORD_USERDOMAIN_SUBST;
+    static const std::string ACL_KEYWORD_ALL;
+    static const std::string ACL_KEYWORD_ACL;
+    static const std::string ACL_KEYWORD_GROUP;
+    static const std::string ACL_KEYWORD_QUOTA;
+    static const std::string ACL_KEYWORD_QUOTA_CONNECTIONS;
+    static const char        ACL_SYMBOL_WILDCARD;
+    static const std::string ACL_KEYWORD_WILDCARD;
+    static const char        ACL_SYMBOL_LINE_CONTINUATION;
+
     void substituteString(std::string& targetString,
                           const std::string& placeholder,
                           const std::string& replacement);
@@ -146,6 +157,31 @@ public:
     void substituteKeywords(std::string& ruleString,
                             const std::string& userId);
 
+    // Per user connection quotas extracted from acl rule file
+    //   Set by reader
+    void setConnQuotaRuleSettings (bool, boost::shared_ptr<quotaRuleSet>);
+    //   Get by connection approvers
+    bool enforcingConnectionQuotas() { return connQuotaRulesExist; }
+    bool getConnQuotaForUser(const std::string&, uint16_t*) const;
+
+    /** getConnectMaxSpec
+     * Connection quotas are held in uint16_t variables.
+     * This function specifies the largest value that a user is allowed
+     * to declare for a connection quota. The upper limit serves two
+     * purposes: 1. It leaves room for magic numbers that may be declared
+     * by keyword names in Acl files and not have those numbers conflict
+     * with innocent user declared values, and 2. It makes the unsigned
+     * math very close to _MAX work reliably with no risk of accidental
+     * wrapping back to zero.
+     */
+    static uint16_t getConnectMaxSpec() {
+        return 65530;
+    }
+    static std::string getMaxConnectSpecStr() {
+        return "65530";
+    }
+
+
     AclData();
     virtual ~AclData();
 
@@ -157,6 +193,10 @@ private:
     bool compareIntMin(const qpid::acl::SpecProperty theProperty,
                        const std::string             theAclValue,
                        const std::string             theLookupValue);
+
+    // Per-user connection quota
+    bool connQuotaRulesExist;
+    boost::shared_ptr<quotaRuleSet> connQuotaRuleSettings; // Map of user-to-N values from rule file
 };
 
 }} // namespace qpid::acl
diff --git a/cpp/src/qpid/acl/AclPlugin.cpp b/cpp/src/qpid/acl/AclPlugin.cpp
index ebf5e90afe..c666eb5420 100644
--- a/cpp/src/qpid/acl/AclPlugin.cpp
+++ b/cpp/src/qpid/acl/AclPlugin.cpp
@@ -42,9 +42,10 @@ struct AclOptions : public Options {
         values.aclMaxConnectTotal = 500;
         addOptions()
             ("acl-file",           optValue(values.aclFile, "FILE"), "The policy file to load from, loaded from data dir")
+            ("connection-limit-per-user", optValue(values.aclMaxConnectPerUser, "N"), "The maximum number of connections allowed per user. 0 implies no limit.")
             ("max-connections"         , optValue(values.aclMaxConnectTotal, "N"),   "The maximum combined number of connections allowed. 0 implies no limit.")
-            ("max-connections-per-user", optValue(values.aclMaxConnectPerUser, "N"), "The maximum number of connections allowed per user. 0 implies no limit.")
-            ("max-connections-per-ip"  , optValue(values.aclMaxConnectPerIp, "N"),   "The maximum number of connections allowed per host IP address. 0 implies no limit.")
+            ("connection-limit-per-ip"  , optValue(values.aclMaxConnectPerIp, "N"),   "The maximum number of connections allowed per host IP address. 0 implies no limit.")
+            ("max-queues-per-user",      optValue(values.aclMaxQueuesPerUser, "N"),  "The maximum number of queues allowed per user. 0 implies no limit.")
             ;
     }
 };
diff --git a/cpp/src/qpid/acl/AclReader.cpp b/cpp/src/qpid/acl/AclReader.cpp
index fae67d0325..7eb9b82c64 100644
--- a/cpp/src/qpid/acl/AclReader.cpp
+++ b/cpp/src/qpid/acl/AclReader.cpp
@@ -24,6 +24,7 @@
 #include <sstream>
 #include "qpid/log/Statement.h"
 #include "qpid/Exception.h"
+#include <boost/lexical_cast.hpp>
 
 #include <iomanip> // degug
 #include <iostream> // debug
@@ -95,7 +96,7 @@ namespace acl {
                 << cnt << " " << (*i)->toString());
 
             if (!foundmode && (*i)->actionAll && (*i)->names.size() == 1
-                && (*((*i)->names.begin())).compare("*") == 0) {
+                && (*((*i)->names.begin())).compare(AclData::ACL_KEYWORD_WILDCARD) == 0) {
                     d->decisionMode = (*i)->res;
                     QPID_LOG(debug, "ACL: FoundMode "
                         << AclHelper::getAclResultStr(d->decisionMode));
@@ -105,13 +106,23 @@ namespace acl {
 
                 // Record which properties have the user substitution string
                 for (pmCitr pItr=rule.props.begin(); pItr!=rule.props.end(); pItr++) {
-                    if ((pItr->second.find(AclData::USER_SUBSTITUTION_KEYWORD, 0)       != std::string::npos) ||
-                        (pItr->second.find(AclData::DOMAIN_SUBSTITUTION_KEYWORD, 0)     != std::string::npos) ||
-                        (pItr->second.find(AclData::USERDOMAIN_SUBSTITUTION_KEYWORD, 0) != std::string::npos)) {
+                    if ((pItr->second.find(AclData::ACL_KEYWORD_USER_SUBST, 0)       != std::string::npos) ||
+                        (pItr->second.find(AclData::ACL_KEYWORD_DOMAIN_SUBST, 0)     != std::string::npos) ||
+                        (pItr->second.find(AclData::ACL_KEYWORD_USERDOMAIN_SUBST, 0) != std::string::npos)) {
                         rule.ruleHasUserSub[pItr->first] = true;
                     }
                 }
 
+                // Find possible routingkey property and cache its pattern
+                for (pmCitr pItr=rule.props.begin(); pItr!=rule.props.end(); pItr++) {
+                    if (acl::SPECPROP_ROUTINGKEY == pItr->first)
+                    {
+                        rule.pubRoutingKeyInRule = true;
+                        rule.pubRoutingKey = (std::string)pItr->second;
+                        rule.addTopicTest(rule.pubRoutingKey);
+                    }
+                }
+
                 // Action -> Object -> map<user -> set<Rule> >
                 std::ostringstream actionstr;
                 for (int acnt = ((*i)->actionAll ? 0 : (*i)->action);
@@ -126,13 +137,6 @@ namespace acl {
                         // Go through the rule properties and find the name and the key.
                         // If found then place them specially for the lookup engine.
                         for (pmCitr pItr=(*i)->props.begin(); pItr!=(*i)->props.end(); pItr++) {
-                            if (acl::SPECPROP_ROUTINGKEY == pItr->first)
-                            {
-                                rule.pubRoutingKeyInRule = true;
-                                rule.pubRoutingKey = (std::string)pItr->second;
-                                rule.addTopicTest(rule.pubRoutingKey);
-                                break;
-                            }
                             if (acl::SPECPROP_NAME == pItr->first)
                             {
                                 rule.pubExchNameInRule = true;
@@ -164,7 +168,7 @@ namespace acl {
                         // add users and Rule to object set
                         bool allNames = false;
                         // check to see if names.begin is '*'
-                        if ((*(*i)->names.begin()).compare("*") == 0)
+                        if ((*(*i)->names.begin()).compare(AclData::ACL_KEYWORD_WILDCARD) == 0)
                             allNames = true;
 
                         for (nsCitr itr  = (allNames ? names.begin() : (*i)->names.begin());
@@ -196,7 +200,7 @@ namespace acl {
                         objstr << AclHelper::getObjectTypeStr((ObjectType) ocnt) << ",";
                 }
 
-                bool allNames = ((*(*i)->names.begin()).compare("*") == 0);
+                bool allNames = ((*(*i)->names.begin()).compare(AclData::ACL_KEYWORD_WILDCARD) == 0);
                 std::ostringstream userstr;
                 for (nsCitr itr  = (allNames ? names.begin() : (*i)->names.begin());
                             itr != (allNames ? names.end()   : (*i)->names.end());
@@ -215,12 +219,15 @@ namespace acl {
                     << "}" );
             }
         }
+
+        // connection quota
+        d->setConnQuotaRuleSettings(connQuotaRulesExist, connQuota);
     }
 
 
     void AclReader::aclRule::processName(const std::string& name, const groupMap& groups) {
-        if (name.compare("all") == 0) {
-            names.insert("*");
+        if (name.compare(AclData::ACL_KEYWORD_ALL) == 0) {
+            names.insert(AclData::ACL_KEYWORD_WILDCARD);
         } else {
             gmCitr itr = groups.find(name);
             if (itr == groups.end()) {
@@ -231,9 +238,13 @@ namespace acl {
         }
     }
 
-    AclReader::AclReader() : lineNumber(0), contFlag(false), validationMap(new AclHelper::objectMap) {
+    AclReader::AclReader(uint16_t theCliMaxConnPerUser) : lineNumber(0), contFlag(false),
+        validationMap(new AclHelper::objectMap),
+        cliMaxConnPerUser (theCliMaxConnPerUser),
+        connQuotaRulesExist(false),
+        connQuota(new AclData::quotaRuleSet) {
         AclHelper::loadValidationMap(validationMap);
-        names.insert("*");
+        names.insert(AclData::ACL_KEYWORD_WILDCARD);
     }
 
     AclReader::~AclReader() {}
@@ -251,6 +262,11 @@ namespace acl {
             errorStream << "Unable to open ACL file \"" << fn << "\": eof=" << (ifs.eof()?"T":"F") << "; fail=" << (ifs.fail()?"T":"F") << "; bad=" << (ifs.bad()?"T":"F");
             return -1;
         }
+        // Propagate nonzero per-user max connection setting from CLI
+        if (cliMaxConnPerUser > 0) {
+            (*connQuota)[AclData::ACL_KEYWORD_ACL] = cliMaxConnPerUser;
+        }
+        // Loop to process the Acl file
         try {
             bool err = false;
             while (ifs.good()) {
@@ -279,6 +295,7 @@ namespace acl {
         }
         printNames();
         printRules();
+        printConnectionQuotas();
         loadDecisionData(d);
 
         return 0;
@@ -289,7 +306,7 @@ namespace acl {
         std::vector<std::string> toks;
 
         // Check for continuation
-        char* contCharPtr = std::strrchr(line, '\\');
+        char* contCharPtr = std::strrchr(line, AclData::ACL_SYMBOL_LINE_CONTINUATION);
         bool cont = contCharPtr != 0;
         if (cont) *contCharPtr = 0;
 
@@ -300,10 +317,12 @@ namespace acl {
             return false;
         }
 
-        if (numToks && (toks[0].compare("group") == 0 || contFlag)) {
+        if (numToks && (toks[0].compare(AclData::ACL_KEYWORD_GROUP) == 0 || contFlag)) {
             ret = processGroupLine(toks, cont);
-        } else if (numToks && toks[0].compare("acl") == 0) {
+        } else if (numToks && toks[0].compare(AclData::ACL_KEYWORD_ACL) == 0) {
             ret = processAclLine(toks);
+        } else if (numToks && toks[0].compare(AclData::ACL_KEYWORD_QUOTA) == 0) {
+            ret = processQuotaLine(toks);
         } else {
             // Check for whitespace only line, ignore these
             bool ws = true;
@@ -314,7 +333,10 @@ namespace acl {
                 ret = true;
             } else {
                 errorStream << ACL_FORMAT_ERR_LOG_PREFIX << "Line : " << lineNumber
-                    << ", Non-continuation line must start with \"group\" or \"acl\".";
+                    << ", Non-continuation line must start with \""
+                    << AclData::ACL_KEYWORD_GROUP << "\", \""
+                    << AclData::ACL_KEYWORD_ACL << "\". or \""
+                    << AclData::ACL_KEYWORD_QUOTA << "\".";
                 ret = false;
             }
         }
@@ -334,6 +356,102 @@ namespace acl {
         return cnt;
     }
 
+
+    // Process 'quota' rule lines
+    // Return true if the line is successfully processed without errors
+    bool AclReader::processQuotaLine(tokList& toks) {
+        const unsigned toksSize = toks.size();
+        const unsigned minimumSize = 3;
+        if (toksSize < minimumSize) {
+            errorStream << ACL_FORMAT_ERR_LOG_PREFIX << "Line : " << lineNumber
+                << ", Insufficient tokens for quota definition.";
+            return false;
+        }
+
+        if (toks[1].compare(AclData::ACL_KEYWORD_QUOTA_CONNECTIONS) == 0) {
+            return processQuotaConnLine(toks);
+        } else {
+            errorStream << ACL_FORMAT_ERR_LOG_PREFIX << "Line : " << lineNumber
+                << ", Quota type \"" << toks[1] << "\" unrecognized.";
+            return false;
+        }
+    }
+
+
+    // Process 'quota connections' rule lines
+    // Return true if the line is successfully processed without errors
+    bool AclReader::processQuotaConnLine(tokList& toks) {
+        const unsigned toksSize = toks.size();
+
+        uint16_t nConns(0);
+        try {
+            nConns = boost::lexical_cast<uint16_t>(toks[2]);
+        } catch(const boost::bad_lexical_cast&) {
+            errorStream << ACL_FORMAT_ERR_LOG_PREFIX << "Line : " << lineNumber
+                << ", Connection quota value \"" << toks[2]
+                << "\" cannot be converted to a 16-bit unsigned integer.";
+            return false;
+        }
+
+        // limit check the connection setting
+        if (nConns > AclData::getConnectMaxSpec())
+        {
+            errorStream << ACL_FORMAT_ERR_LOG_PREFIX << "Line : " << lineNumber
+                << ", Connection quota value \"" << toks[2]
+                << "\" exceeds maximum configuration setting of "
+                << AclData::getConnectMaxSpec();
+            return false;
+        }
+
+        // Apply the connection count to all names in rule
+        for (unsigned idx = 3; idx < toksSize; idx++) {
+            if (groups.find(toks[idx]) == groups.end()) {
+                // This is the name of an individual, not a group
+                (*connQuota)[toks[idx]] = nConns;
+            } else {
+                if (!processQuotaConnGroup(toks[idx], nConns))
+                    return false;
+            }
+        }
+        return true;
+    }
+
+
+    // Process 'quota connections' group expansion
+    // Return true if the quota is applied to all members of the group
+    bool AclReader::processQuotaConnGroup(const std::string& theGroup, uint16_t theQuota) {
+        gmCitr citr = groups.find(theGroup);
+
+        if (citr == groups.end()) {
+            errorStream << ACL_FORMAT_ERR_LOG_PREFIX << "Line : " << lineNumber
+                << ", Failed to expand group \"" << theGroup << "\".";
+            return false;
+        }
+
+        for (nsCitr gni=citr->second->begin(); gni!=citr->second->end(); gni++) {
+            if (groups.find(*gni) == groups.end()) {
+                (*connQuota)[*gni] = theQuota;
+            } else {
+                if (!processQuotaConnGroup(*gni, theQuota))
+                    return false;
+            }
+        }
+        return true;
+    }
+
+
+    void AclReader::printConnectionQuotas() const {
+        QPID_LOG(debug, "ACL: connection quota: " << (*connQuota).size() << " rules found:");
+        int cnt = 1;
+        for (AclData::quotaRuleSetItr itr=(*connQuota).begin();
+                                      itr != (*connQuota).end();
+                                      ++itr,++cnt) {
+            QPID_LOG(debug, "ACL: quota " << cnt << " : " << (*itr).second
+                << " connections for " << (*itr).first)
+        }
+    }
+
+
     // Return true if the line is successfully processed without errors
     // If cont is true, then groupName must be set to the continuation group name
     bool AclReader::processGroupLine(tokList& toks, const bool cont) {
@@ -459,8 +577,8 @@ namespace acl {
             return false;
         }
 
-        bool actionAllFlag = toks[3].compare("all") == 0;
-        bool userAllFlag   = toks[2].compare("all") == 0;
+        bool actionAllFlag = toks[3].compare(AclData::ACL_KEYWORD_ALL) == 0;
+        bool userAllFlag   = toks[2].compare(AclData::ACL_KEYWORD_ALL) == 0;
         Action action;
         if (actionAllFlag) {
 
@@ -489,7 +607,7 @@ namespace acl {
         }
 
         if (toksSize >= 5) { // object name-value pair
-            if (toks[4].compare("all") == 0) {
+            if (toks[4].compare(AclData::ACL_KEYWORD_ALL) == 0) {
                 rule->setObjectTypeAll();
             } else {
                 try {
@@ -523,7 +641,7 @@ namespace acl {
             }
         }
         // Check if name (toks[2]) is group; if not, add as name of individual
-        if (toks[2].compare("all") != 0) {
+        if (toks[2].compare(AclData::ACL_KEYWORD_ALL) != 0) {
             if (groups.find(toks[2]) == groups.end()) {
                 addName(toks[2]);
             }
diff --git a/cpp/src/qpid/acl/AclReader.h b/cpp/src/qpid/acl/AclReader.h
index 6351c1e509..1fa374c59c 100644
--- a/cpp/src/qpid/acl/AclReader.h
+++ b/cpp/src/qpid/acl/AclReader.h
@@ -28,6 +28,7 @@
 #include <sstream>
 #include <memory>
 #include "qpid/acl/AclData.h"
+#include "qpid/acl/Acl.h"
 #include "qpid/broker/AclModule.h"
 
 namespace qpid {
@@ -96,7 +97,7 @@ class AclReader {
     std::ostringstream      errorStream;
 
   public:
-    AclReader();
+    AclReader(uint16_t cliMaxConnPerUser);
     virtual ~AclReader();
     int read(const std::string& fn, boost::shared_ptr<AclData> d); // return=0 for success
     std::string getError();
@@ -116,8 +117,17 @@ class AclReader {
     void printRules() const; // debug aid
     bool isValidUserName(const std::string& name);
 
+    bool processQuotaLine(tokList& toks);
+    bool processQuotaConnLine(tokList& toks);
+    bool processQuotaConnGroup(const std::string&, uint16_t);
+    void printConnectionQuotas() const;
+
     static bool isValidGroupName(const std::string& name);
     static nvPair splitNameValuePair(const std::string& nvpString);
+
+    const uint16_t cliMaxConnPerUser;
+    bool connQuotaRulesExist;
+    boost::shared_ptr<AclData::quotaRuleSet> connQuota;
 };
 
 }} // namespace qpid::acl
diff --git a/cpp/src/qpid/acl/AclResourceCounter.cpp b/cpp/src/qpid/acl/AclResourceCounter.cpp
new file mode 100644
index 0000000000..66dfd0777e
--- /dev/null
+++ b/cpp/src/qpid/acl/AclResourceCounter.cpp
@@ -0,0 +1,165 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "AclResourceCounter.h"
+#include "Acl.h"
+#include "qpid/log/Statement.h"
+#include "qpid/sys/Mutex.h"
+#include <assert.h>
+#include <sstream>
+
+using namespace qpid::sys;
+
+namespace qpid {
+namespace acl {
+
+//
+// This module approves various resource creation requests:
+//   Queues
+//
+
+
+//
+//
+//
+ResourceCounter::ResourceCounter(Acl& a, uint16_t ql) :
+    acl(a), queueLimit(ql) {}
+
+ResourceCounter::~ResourceCounter() {}
+
+
+//
+// limitApproveLH
+//
+// Resource creation approver.
+// If user is under limit increment count and return true.
+// Called with lock held.
+//
+bool ResourceCounter::limitApproveLH(
+    const std::string& theTitle,
+    countsMap_t& theMap,
+    const std::string& theName,
+    uint16_t theLimit,
+    bool emitLog) {
+
+    bool result(true);
+    if (theLimit > 0) {
+        uint16_t count;
+        countsMap_t::iterator eRef = theMap.find(theName);
+        if (eRef != theMap.end()) {
+            count = (uint16_t)(*eRef).second;
+            result = count < theLimit;
+            if (result) {
+                count += 1;
+                (*eRef).second = count;
+            }
+        } else {
+            // Not found
+            theMap[theName] = count = 1;
+        }
+        if (emitLog) {
+            QPID_LOG(trace, theTitle << theName
+                << " limit=" << theLimit
+                << " curValue=" << count
+                << " result=" << (result ? "allow" : "deny"));
+        }
+    }
+    return result;
+}
+
+
+//
+// releaseLH
+//
+// Decrement the name's count in map.
+// called with dataLock already taken
+//
+void ResourceCounter::releaseLH(
+    const std::string& theTitle, countsMap_t& theMap, const std::string& theName, uint16_t theLimit) {
+
+    if (theLimit > 0) {
+        countsMap_t::iterator eRef = theMap.find(theName);
+        if (eRef != theMap.end()) {
+            uint16_t count = (uint16_t) (*eRef).second;
+            assert (count > 0);
+            if (1 == count) {
+                theMap.erase (eRef);
+            } else {
+                (*eRef).second = count - 1;
+            }
+        } else {
+            // User had no connections.
+            QPID_LOG(notice, theTitle << theName
+                << "' not found in resource count pool");
+        }
+    }
+}
+
+
+//
+// approveCreateQueue
+//  Count an attempted queue creation by this user.
+//  Disapprove if over limit.
+//
+bool ResourceCounter::approveCreateQueue(const std::string& userId, const std::string& queueName)
+{
+    Mutex::ScopedLock locker(dataLock);
+
+    bool okByQ = limitApproveLH("ACL Queue creation approver. userId:", queuePerUserMap, userId, queueLimit, true);
+
+    if (okByQ) {
+        // Queue is owned by this userId
+        queueOwnerMap[queueName] = userId;
+
+        QPID_LOG(trace, "ACL create queue approved for user '" << userId
+            << "' queue '" << queueName << "'");
+    } else {
+
+        QPID_LOG(error, "Client max queue count limit of " << queueLimit
+            << " exceeded by '" << userId << "' creating queue '"
+            << queueName << "'. Queue creation denied.");
+
+        acl.reportQueueLimit(userId, queueName);
+    }
+    return okByQ;
+}
+
+
+//
+// recordDestroyQueue
+//  Return a destroyed queue to a user's quota
+//
+void ResourceCounter::recordDestroyQueue(const std::string& queueName)
+{
+    Mutex::ScopedLock locker(dataLock);
+
+    queueOwnerMap_t::iterator eRef = queueOwnerMap.find(queueName);
+    if (eRef != queueOwnerMap.end()) {
+        releaseLH("ACL resource counter: Queue owner for queue '", queuePerUserMap, (*eRef).second, queueLimit);
+
+        queueOwnerMap.erase(eRef);
+    } else {
+        QPID_LOG(notice, "ACL resource counter: Queue '" << queueName
+            << "' not found in queue owner map");
+    }
+}
+
+}} // namespace qpid::acl
diff --git a/cpp/src/qpid/acl/AclResourceCounter.h b/cpp/src/qpid/acl/AclResourceCounter.h
new file mode 100644
index 0000000000..f5995eb961
--- /dev/null
+++ b/cpp/src/qpid/acl/AclResourceCounter.h
@@ -0,0 +1,78 @@
+#ifndef QPID_ACL_RESOURCECOUNTER_H
+#define QPID_ACL_RESOURCECOUNTER_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "qpid/sys/Mutex.h"
+#include <boost/iterator/iterator_concepts.hpp>
+
+#include <map>
+
+namespace qpid {
+
+namespace acl {
+class Acl;
+
+ /**
+ * Approve or disapprove resource creation requests
+ */
+class ResourceCounter
+{
+private:
+    typedef std::map<std::string, uint32_t> countsMap_t;
+    typedef std::map<std::string, std::string> queueOwnerMap_t;
+
+    Acl&             acl;
+    uint16_t         queueLimit;
+    qpid::sys::Mutex dataLock;
+
+    /** Records queueName-queueUserId */
+    queueOwnerMap_t queueOwnerMap;
+
+    /** Records queue-by-owner counts */
+    countsMap_t queuePerUserMap;
+
+    /** Return approval for proposed resource creation */
+    bool limitApproveLH(const std::string& theTitle,
+                        countsMap_t& theMap,
+                        const std::string& theName,
+                        uint16_t theLimit,
+                        bool emitLog);
+
+    /** Release a connection */
+    void releaseLH(const std::string& theTitle,
+                   countsMap_t& theMap,
+                   const std::string& theName,
+                   uint16_t theLimit);
+
+public:
+    ResourceCounter(Acl& acl, uint16_t ql);
+    ~ResourceCounter();
+
+    // Queue counting
+    bool approveCreateQueue(const std::string& userId, const std::string& queueName);
+    void recordDestroyQueue(const std::string& queueName);
+};
+
+}} // namespace qpid::acl
+
+#endif  /*!QPID_ACL_RESOURCECOUNTER_H*/
diff --git a/cpp/src/qpid/acl/AclTopicMatch.h b/cpp/src/qpid/acl/AclTopicMatch.h
index 486c229ad5..654d1d63d4 100644
--- a/cpp/src/qpid/acl/AclTopicMatch.h
+++ b/cpp/src/qpid/acl/AclTopicMatch.h
@@ -30,7 +30,7 @@ namespace qpid {
 namespace broker {
 
 // Class for executing topic exchange routing key matching rules in
-// Acl code the allows or denies users publishing to an exchange.
+// Acl code. Allows or denies users publishing to an exchange.
 class TopicExchange::TopicExchangeTester {
 
 class boundNode;
diff --git a/cpp/src/qpid/acl/AclValidator.cpp b/cpp/src/qpid/acl/AclValidator.cpp
index 85f0f7c240..73b49b2959 100644
--- a/cpp/src/qpid/acl/AclValidator.cpp
+++ b/cpp/src/qpid/acl/AclValidator.cpp
@@ -94,6 +94,22 @@ namespace acl {
                                     boost::shared_ptr<PropertyType>(
                                         new IntPropertyType(0,std::numeric_limits<int64_t>::max()))));
 
+        validators.insert(Validator(acl::SPECPROP_MAXFILESIZELOWERLIMIT,
+                                    boost::shared_ptr<PropertyType>(
+                                        new IntPropertyType(0,std::numeric_limits<int64_t>::max()))));
+
+        validators.insert(Validator(acl::SPECPROP_MAXFILESIZEUPPERLIMIT,
+                                    boost::shared_ptr<PropertyType>(
+                                        new IntPropertyType(0,std::numeric_limits<int64_t>::max()))));
+
+        validators.insert(Validator(acl::SPECPROP_MAXFILECOUNTLOWERLIMIT,
+                                    boost::shared_ptr<PropertyType>(
+                                        new IntPropertyType(0,std::numeric_limits<int64_t>::max()))));
+
+        validators.insert(Validator(acl::SPECPROP_MAXFILECOUNTUPPERLIMIT,
+                                    boost::shared_ptr<PropertyType>(
+                                        new IntPropertyType(0,std::numeric_limits<int64_t>::max()))));
+
         std::string policyTypes[] = {"ring", "ring_strict", "flow_to_disk", "reject"};
         std::vector<std::string> v(policyTypes, policyTypes + sizeof(policyTypes) / sizeof(std::string));
         validators.insert(Validator(acl::SPECPROP_POLICYTYPE,
diff --git a/cpp/src/qpid/acl/management-schema.xml b/cpp/src/qpid/acl/management-schema.xml
index f52c251bed..2ac20bb324 100644
--- a/cpp/src/qpid/acl/management-schema.xml
+++ b/cpp/src/qpid/acl/management-schema.xml
@@ -25,8 +25,10 @@
     <property name="maxConnections"        type="uint16"  access="RO"       desc="Maximum allowed connections"/>
     <property name="maxConnectionsPerIp"   type="uint16"  access="RO"       desc="Maximum allowed connections"/>
     <property name="maxConnectionsPerUser" type="uint16"  access="RO"       desc="Maximum allowed connections"/>
+    <property name="maxQueuesPerUser"      type="uint16"  access="RO"       desc="Maximum allowed queues"/>
     <statistic name="aclDenyCount"         type="count64" unit="request"    desc="Number of ACL requests denied"/>
     <statistic name="connectionDenyCount"  type="count64" unit="connection" desc="Number of connections denied"/>
+    <statistic name="queueQuotaDenyCount"  type="count64" unit="queue"      desc="Number of queue creations denied"/>
 
     <method name="reloadACLFile" desc="Reload the ACL file"/>
 
@@ -70,11 +72,13 @@
     <arg name="reason"     type="lstr"/>
     <arg name="userId"     type="sstr"/>
     <arg name="clientAddr" type="sstr"/>
+    <arg name="queueName"  type="sstr"/>
   </eventArguments>
 
   <event name="allow"          sev="inform" args="userId, action, objectType, objectName, arguments"/>
   <event name="deny"           sev="notice" args="userId, action, objectType, objectName, arguments"/>
   <event name="connectionDeny" sev="notice" args="userId, clientAddr"/>
+  <event name="queueQuotaDeny" sev="notice" args="userId, queueName"/>
   <event name="fileLoaded"     sev="inform" args="userId"/>
   <event name="fileLoadFailed" sev="error"  args="userId, reason"/>