QPID-3858: Updated branch - merged from trunk r.1333987

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

12 files changed, 1193 insertions, 446 deletions

diff --git a/cpp/src/qpid/acl/Acl.cpp b/cpp/src/qpid/acl/Acl.cpp
index 12bf13018c..917c2e3398 100644
--- a/cpp/src/qpid/acl/Acl.cpp
+++ b/cpp/src/qpid/acl/Acl.cpp
@@ -17,6 +17,7 @@
  */
 
 #include "qpid/acl/Acl.h"
+#include "qpid/acl/AclConnectionCounter.h"
 #include "qpid/acl/AclData.h"
 #include "qpid/acl/AclValidator.h"
 #include "qpid/sys/Mutex.h"
@@ -26,8 +27,11 @@
 #include "qpid/Options.h"
 #include "qpid/log/Logger.h"
 #include "qpid/types/Variant.h"
+#include "qmf/org/apache/qpid/acl/ArgsAclLookup.h"
+#include "qmf/org/apache/qpid/acl/ArgsAclLookupPublish.h"
 #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/EventDeny.h"
 #include "qmf/org/apache/qpid/acl/EventFileLoaded.h"
 #include "qmf/org/apache/qpid/acl/EventFileLoadFailed.h"
@@ -35,7 +39,6 @@
 #include <map>
 
 #include <boost/shared_ptr.hpp>
-#include <boost/utility/in_place_factory.hpp>
 
 using namespace std;
 using namespace qpid::acl;
@@ -47,7 +50,8 @@ 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)
+Acl::Acl (AclValues& av, Broker& b): aclValues(av), broker(&b), transferAcl(false), mgmtObject(0),
+    connectionCounter(new ConnectionCounter(*this, aclValues.aclMaxConnectPerUser, aclValues.aclMaxConnectPerIp))
 {
 
     agent = broker->getManagementAgent();
@@ -62,14 +66,30 @@ Acl::Acl (AclValues& av, Broker& b): aclValues(av), broker(&b), transferAcl(fals
         throw Exception("Could not read ACL file " + errorString);
         if (mgmtObject!=0) mgmtObject->set_enforcingAcl(0);
     }
+    broker->getConnectionObservers().add(connectionCounter);
     QPID_LOG(info, "ACL Plugin loaded");
     if (mgmtObject!=0) mgmtObject->set_enforcingAcl(1);
 }
 
-bool Acl::authorise(const std::string& id, const Action& action, const ObjectType& objType, const std::string& name, std::map<Property, std::string>* params)
+
+void Acl::reportConnectLimit(const std::string user, const std::string addr)
+{
+    if (mgmtObject!=0)
+        mgmtObject->inc_connectionDenyCount();
+
+    agent->raiseEvent(_qmf::EventConnectionDeny(user, addr));
+}
+
+
+bool Acl::authorise(
+    const std::string&               id,
+    const Action&                    action,
+    const ObjectType&                objType,
+    const std::string&               name,
+    std::map<Property, std::string>* params)
 {
     boost::shared_ptr<AclData> dataLocal;
-    { 
+    {
         Mutex::ScopedLock locker(dataLock);
         dataLocal = data;  //rcu copy
     }
@@ -81,7 +101,12 @@ bool Acl::authorise(const std::string& id, const Action& action, const ObjectTyp
     return result(aclreslt, id, action, objType, name);
 }
 
-bool Acl::authorise(const std::string& id, const Action& action, const ObjectType& objType, const std::string& ExchangeName, const std::string& RoutingKey)
+bool Acl::authorise(
+    const std::string& id,
+    const Action&      action,
+    const ObjectType& objType,
+    const std::string& ExchangeName,
+    const std::string& RoutingKey)
 {
     boost::shared_ptr<AclData> dataLocal;
     {
@@ -96,31 +121,50 @@ bool Acl::authorise(const std::string& id, const Action& action, const ObjectTyp
 }
 
 
-bool Acl::result(const AclResult& aclreslt, const std::string& id, const Action& action, const ObjectType& objType, const std::string& name)
+bool Acl::result(
+    const AclResult&   aclreslt,
+    const std::string& id,
+    const Action&      action,
+    const ObjectType&  objType,
+    const std::string& name)
 {
+    bool result(false);
+
     switch (aclreslt)
     {
     case ALLOWLOG:
-        QPID_LOG(info, "ACL Allow id:" << id <<" action:" << AclHelper::getActionStr(action) <<
-            " ObjectType:" << AclHelper::getObjectTypeStr(objType) << " Name:" << name );
+        QPID_LOG(info, "ACL Allow id:" << id
+            << " action:" << AclHelper::getActionStr(action)
+            << " ObjectType:" << AclHelper::getObjectTypeStr(objType)
+            << " Name:" << name );
         agent->raiseEvent(_qmf::EventAllow(id,  AclHelper::getActionStr(action),
-            AclHelper::getObjectTypeStr(objType),
-            name, types::Variant::Map()));
+                          AclHelper::getObjectTypeStr(objType),
+                          name, types::Variant::Map()));
+        // FALLTHROUGH
     case ALLOW:
-        return true;
-    case DENY:
-        if (mgmtObject!=0) mgmtObject->inc_aclDenyCount();
-        return false;
+        result = true;
+        break;
+
     case DENYLOG:
-        if (mgmtObject!=0) mgmtObject->inc_aclDenyCount();
-    default:
-        QPID_LOG(info, "ACL Deny id:" << id << " action:" << AclHelper::getActionStr(action) << " ObjectType:" << AclHelper::getObjectTypeStr(objType) << " Name:" << name);
+        QPID_LOG(info, "ACL Deny id:" << id
+            << " action:" << AclHelper::getActionStr(action)
+            << " ObjectType:" << AclHelper::getObjectTypeStr(objType)
+            << " Name:" << name);
         agent->raiseEvent(_qmf::EventDeny(id, AclHelper::getActionStr(action),
-            AclHelper::getObjectTypeStr(objType),
-            name, types::Variant::Map()));
-        return false;
+                                          AclHelper::getObjectTypeStr(objType),
+                                          name, types::Variant::Map()));
+        // FALLTHROUGH
+    case DENY:
+        if (mgmtObject!=0)
+            mgmtObject->inc_aclDenyCount();
+        result = false;
+        break;
+
+    default:
+        assert (false);
     }
-    return false;
+
+    return result;
 }
 
 bool Acl::readAclFile(std::string& errorText)
@@ -129,7 +173,7 @@ bool Acl::readAclFile(std::string& errorText)
     return readAclFile(aclValues.aclFile, errorText);
 }
 
-bool Acl::readAclFile(std::string& aclFile, std::string& errorText) {      
+bool Acl::readAclFile(std::string& aclFile, std::string& errorText) {
     boost::shared_ptr<AclData> d(new AclData);
     AclReader ar;
     if (ar.read(aclFile, d)){
@@ -142,17 +186,17 @@ bool Acl::readAclFile(std::string& aclFile, std::string& errorText) {
     AclValidator validator;
     validator.validate(d);
 
-    {  
+    {
         Mutex::ScopedLock locker(dataLock);
         data = d;
     }
     transferAcl = data->transferAcl; // any transfer ACL
 
     if (data->transferAcl){
-        QPID_LOG(debug,"Transfer ACL is Enabled!");
+        QPID_LOG(debug,"ACL: Transfer ACL is Enabled!");
     }
 
-    data->aclSource = aclFile; 
+    data->aclSource = aclFile;
     if (mgmtObject!=0){
         mgmtObject->set_transferAcl(transferAcl?1:0);
         mgmtObject->set_policyFile(aclFile);
@@ -164,17 +208,92 @@ bool Acl::readAclFile(std::string& aclFile, std::string& errorText) {
     return true;
 }
 
-Acl::~Acl(){}
+
+//
+// management lookup function performs general query on acl engine
+//
+Manageable::status_t Acl::lookup(qpid::management::Args& args, std::string& text)
+{
+    _qmf::ArgsAclLookup& ioArgs = (_qmf::ArgsAclLookup&) args;
+    Manageable::status_t result(STATUS_USER);
+
+   try {
+        ObjectType   objType = AclHelper::getObjectType(ioArgs.i_object);
+        Action       action  = AclHelper::getAction(    ioArgs.i_action);
+        std::map<Property, std::string> propertyMap;
+        for (::qpid::types::Variant::Map::const_iterator
+             iMapIter  = ioArgs.i_propertyMap.begin();
+             iMapIter != ioArgs.i_propertyMap.end();
+             iMapIter++)
+        {
+            Property property = AclHelper::getProperty(iMapIter->first);
+            propertyMap.insert(make_pair(property, iMapIter->second));
+        }
+
+        boost::shared_ptr<AclData> dataLocal;
+        {
+            Mutex::ScopedLock locker(dataLock);
+            dataLocal = data;  //rcu copy
+        }
+        AclResult aclResult = dataLocal->lookup(
+            ioArgs.i_userId,
+            action,
+            objType,
+            ioArgs.i_objectName,
+            &propertyMap);
+
+        ioArgs.o_result = AclHelper::getAclResultStr(aclResult);
+        result = STATUS_OK;
+
+    } catch (const std::exception& e) {
+        std::ostringstream oss;
+        oss << "AclLookup invalid name : " << e.what();
+        ioArgs.o_result =  oss.str();
+        text = oss.str();
+    }
+
+    return result;
+}
+
+
+//
+// management lookupPublish function performs fastpath
+// PUBLISH EXCHANGE query on acl engine
+//
+Manageable::status_t Acl::lookupPublish(qpid::management::Args& args, std::string& /*text*/)
+{
+    _qmf::ArgsAclLookupPublish& ioArgs = (_qmf::ArgsAclLookupPublish&) args;
+    boost::shared_ptr<AclData>       dataLocal;
+    {
+        Mutex::ScopedLock locker(dataLock);
+        dataLocal = data;  //rcu copy
+    }
+    AclResult aclResult = dataLocal->lookup(
+        ioArgs.i_userId,
+        ACT_PUBLISH,
+        OBJ_EXCHANGE,
+        ioArgs.i_exchangeName,
+        ioArgs.i_routingKey);
+
+    ioArgs.o_result = AclHelper::getAclResultStr(aclResult);
+
+    return STATUS_OK;
+}
+
+
+Acl::~Acl(){
+    broker->getConnectionObservers().remove(connectionCounter);
+}
 
 ManagementObject* Acl::GetManagementObject(void) const
 {
     return (ManagementObject*) mgmtObject;
 }
 
-Manageable::status_t Acl::ManagementMethod (uint32_t methodId, Args& /*args*/, string& text)
+Manageable::status_t Acl::ManagementMethod (uint32_t methodId, Args& args, string& text)
 {
     Manageable::status_t status = Manageable::STATUS_UNKNOWN_METHOD;
-    QPID_LOG (debug, "Queue::ManagementMethod [id=" << methodId << "]");
+    QPID_LOG (debug, "ACL: Queue::ManagementMethod [id=" << methodId << "]");
 
     switch (methodId)
     {
@@ -185,6 +304,14 @@ Manageable::status_t Acl::ManagementMethod (uint32_t methodId, Args& /*args*/, s
         else
             status = Manageable::STATUS_USER;
         break;
+
+    case _qmf::Acl::METHOD_LOOKUP :
+        status = lookup(args, text);
+        break;
+
+    case _qmf::Acl::METHOD_LOOKUPPUBLISH :
+        status = lookupPublish(args, text);
+        break;
     }
 
     return status;
diff --git a/cpp/src/qpid/acl/Acl.h b/cpp/src/qpid/acl/Acl.h
index 77f43838de..c3451018ef 100644
--- a/cpp/src/qpid/acl/Acl.h
+++ b/cpp/src/qpid/acl/Acl.h
@@ -30,6 +30,7 @@
 #include "qmf/org/apache/qpid/acl/Acl.h"
 #include "qpid/sys/Mutex.h"
 
+#include <boost/shared_ptr.hpp>
 #include <map>
 #include <string>
 
@@ -40,9 +41,12 @@ class Broker;
 }
 
 namespace acl {
+class ConnectionCounter;
 
 struct AclValues {
-	std::string aclFile;
+    std::string aclFile;
+    uint32_t    aclMaxConnectPerUser;
+    uint32_t    aclMaxConnectPerIp;
 };
 
 
@@ -50,37 +54,56 @@ class Acl : public broker::AclModule, public RefCounted, public management::Mana
 {
 
 private:
-   acl::AclValues aclValues;
-   broker::Broker* broker;
-   bool transferAcl;
-   boost::shared_ptr<AclData> data;
-   qmf::org::apache::qpid::acl::Acl* mgmtObject; // mgnt owns lifecycle
-   qpid::management::ManagementAgent* agent;
-   mutable qpid::sys::Mutex dataLock; 
+    acl::AclValues                       aclValues;
+    broker::Broker*                      broker;
+    bool                                 transferAcl;
+    boost::shared_ptr<AclData>           data;
+    qmf::org::apache::qpid::acl::Acl*    mgmtObject; // mgnt owns lifecycle
+    qpid::management::ManagementAgent*   agent;
+    mutable qpid::sys::Mutex             dataLock;
+    boost::shared_ptr<ConnectionCounter> connectionCounter;
 
 public:
-   Acl (AclValues& av, broker::Broker& b);
-
-   void initialize();
-   
-   inline virtual bool doTransferAcl() {return transferAcl;};
-   
-   // create specilied authorise methods for cases that need faster matching as needed.
-   virtual bool authorise(const std::string& id, const Action& action, const ObjectType& objType, const std::string& name, std::map<Property, std::string>* params=0);
-   virtual bool authorise(const std::string& id, const Action& action, const ObjectType& objType, const std::string& ExchangeName,const std::string& RoutingKey);
-
-   virtual ~Acl();
+    Acl (AclValues& av, broker::Broker& b);
+
+    void reportConnectLimit(const std::string user, const std::string addr);
+
+    inline virtual bool doTransferAcl() {
+        return transferAcl;
+    };
+
+// create specilied authorise methods for cases that need faster matching as needed.
+    virtual bool authorise(
+        const std::string&               id,
+        const Action&                    action,
+        const ObjectType&                objType,
+        const std::string&               name,
+        std::map<Property, std::string>* params=0);
+
+    virtual bool authorise(
+        const std::string&               id,
+        const Action&                    action,
+        const ObjectType&                objType,
+        const std::string&               ExchangeName,
+        const std::string&               RoutingKey);
+
+    virtual ~Acl();
 private:
-   bool result(const AclResult& aclreslt, const std::string& id, const Action& action, const ObjectType& objType, const std::string& name);
-   bool readAclFile(std::string& errorText);
-   bool readAclFile(std::string& aclFile, std::string& errorText);
-   virtual qpid::management::ManagementObject* GetManagementObject(void) const;
-   virtual management::Manageable::status_t ManagementMethod (uint32_t methodId, management::Args& args, std::string& text);
+    bool result(
+        const AclResult&   aclreslt,
+        const std::string& id,
+        const Action&      action,
+        const ObjectType&  objType,
+        const std::string& name);
+    bool readAclFile(std::string& errorText);
+    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 management::Manageable::status_t ManagementMethod (uint32_t methodId, management::Args& args, std::string& text);
 
 };
 
-
-
 }} // namespace qpid::acl
 
 #endif // QPID_ACL_ACL_H
diff --git a/cpp/src/qpid/acl/AclConnectionCounter.cpp b/cpp/src/qpid/acl/AclConnectionCounter.cpp
new file mode 100644
index 0000000000..5d4e3c1544
--- /dev/null
+++ b/cpp/src/qpid/acl/AclConnectionCounter.cpp
@@ -0,0 +1,214 @@
+/*
+ *
+ * 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 "AclConnectionCounter.h"
+#include "Acl.h"
+#include "qpid/broker/Connection.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 instantiates a broker::ConnectionObserver and limits client
+// connections by counting connections per user name and per client IP address.
+//
+
+
+//
+//
+//
+ConnectionCounter::ConnectionCounter(Acl& a, uint32_t nl, uint32_t hl) :
+    acl(a), nameLimit(nl), hostLimit(hl) {}
+
+ConnectionCounter::~ConnectionCounter() {}
+
+
+//
+// limitCheckLH
+//
+// Increment the name's count in map and return a comparison against the limit.
+// called with dataLock already taken
+//
+bool ConnectionCounter::limitCheckLH(
+    connectCountsMap_t& theMap, const std::string& theName, uint32_t theLimit) {
+
+    bool result(true);
+    if (theLimit > 0) {
+        connectCountsMap_t::iterator eRef = theMap.find(theName);
+        if (eRef != theMap.end()) {
+            uint32_t count = (uint32_t)(*eRef).second + 1;
+            (*eRef).second = count;
+            result = count <= theLimit;
+        } else {
+            theMap[theName] = 1;
+        }
+    }
+    return result;
+}
+
+
+//
+// releaseLH
+//
+// Decrement the name's count in map.
+// called with dataLock already taken
+//
+void ConnectionCounter::releaseLH(
+    connectCountsMap_t& theMap, const std::string& theName, uint32_t theLimit) {
+
+    if (theLimit > 0) {
+        connectCountsMap_t::iterator eRef = theMap.find(theName);
+        if (eRef != theMap.end()) {
+            uint32_t count = (uint32_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, "ACL ConnectionCounter Connection for '" << theName
+                << "' not found in connection count pool");
+        }
+    }
+}
+
+
+//
+// connection - called during Connection's constructor
+//
+void ConnectionCounter::connection(broker::Connection& connection) {
+    QPID_LOG(trace, "ACL ConnectionCounter connection IP:" << connection.getMgmtId()
+        << ", userId:" << connection.getUserId());
+
+    Mutex::ScopedLock locker(dataLock);
+
+    connectProgressMap[connection.getMgmtId()] = C_CREATED;
+}
+
+
+//
+// opened - called when first AMQP frame is received over Connection
+//
+void ConnectionCounter::opened(broker::Connection& connection) {
+    QPID_LOG(trace, "ACL ConnectionCounter Opened IP:" << connection.getMgmtId()
+        << ", userId:" << connection.getUserId());
+
+    Mutex::ScopedLock locker(dataLock);
+
+    const std::string& userName(              connection.getUserId());
+    const std::string& hostName(getClientHost(connection.getMgmtId()));
+
+    // Bump state from CREATED to OPENED
+    (void) limitCheckLH(connectProgressMap, connection.getMgmtId(), C_OPENED);
+
+    bool nameOk = limitCheckLH(connectByNameMap, userName, nameLimit);
+    bool hostOk = limitCheckLH(connectByHostMap, hostName, hostLimit);
+
+    if (!nameOk) {
+        // User has too many
+        acl.reportConnectLimit(userName, hostName);
+        QPID_LOG(notice, "ACL ConnectionCounter User '" << userName
+            << "' exceeded maximum allowed connections");
+        throw Exception(
+            QPID_MSG("User '" << userName
+                << "' exceeded maximum allowed connections"));
+    }
+
+    if (!hostOk) {
+        // Host has too many
+        acl.reportConnectLimit(userName, hostName);
+        QPID_LOG(notice, "ACL ConnectionCounter Client host '" << hostName
+            << "' exceeded maximum allowed connections");
+        throw Exception(
+            QPID_MSG("Client host '" << hostName
+                << "' exceeded maximum allowed connections"));
+    }
+}
+
+
+//
+// closed - called during Connection's destructor
+//
+void ConnectionCounter::closed(broker::Connection& connection) {
+    QPID_LOG(trace, "ACL ConnectionCounter Closed IP:" << connection.getMgmtId()
+        << ", userId:" << connection.getUserId());
+
+    Mutex::ScopedLock locker(dataLock);
+
+    connectCountsMap_t::iterator eRef = connectProgressMap.find(connection.getMgmtId());
+    if (eRef != connectProgressMap.end()) {
+        if ((*eRef).second == C_OPENED){
+            // Normal case: connection was created and opened.
+            // Decrement in-use counts
+            releaseLH(connectByNameMap,
+                      connection.getUserId(),
+                      nameLimit);
+
+            releaseLH(connectByHostMap,
+                      getClientHost(connection.getMgmtId()),
+                      hostLimit);
+        } else {
+            // Connection was created but not opened.
+            // Don't decrement any connection counts.
+        }
+        connectProgressMap.erase(eRef);
+
+    } else {
+        // connection not found in progress map
+        QPID_LOG(notice, "ACL ConnectionCounter info for '" << connection.getMgmtId()
+            << "' not found in connection state pool");
+    }
+}
+
+
+//
+// getClientIp - given a connection's mgmtId return the client host part.
+//
+// TODO: Ideally this would be a method of the connection itself.
+//
+std::string ConnectionCounter::getClientHost(const std::string mgmtId)
+{
+    size_t hyphen = mgmtId.find('-');
+    if (std::string::npos != hyphen) {
+        size_t colon = mgmtId.find_last_of(':');
+        if (std::string::npos != colon) {
+            // trailing colon found
+            return mgmtId.substr(hyphen+1, colon - hyphen - 1);
+        } else {
+            // colon not found - use everything after hyphen
+            return mgmtId.substr(hyphen+1);
+        }
+    }
+
+    // no hyphen found - use whole string
+    assert(false);
+    return mgmtId;
+}
+
+}} // namespace qpid::ha
diff --git a/cpp/src/qpid/acl/AclConnectionCounter.h b/cpp/src/qpid/acl/AclConnectionCounter.h
new file mode 100644
index 0000000000..31d11540fd
--- /dev/null
+++ b/cpp/src/qpid/acl/AclConnectionCounter.h
@@ -0,0 +1,81 @@
+#ifndef QPID_ACL_CONNECTIONCOUNTER_H
+#define QPID_ACL_CONNECTIONCOUNTER_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/broker/ConnectionObserver.h"
+#include "qpid/sys/Mutex.h"
+#include <boost/iterator/iterator_concepts.hpp>
+
+#include <map>
+
+namespace qpid {
+
+namespace broker {
+class Connection;
+}
+
+namespace acl {
+class Acl;
+
+ /**
+ * Terminate client connections when a user tries to create 'too many'.
+ * Terminate hostIp connections when an IP host tries to create 'too many'.
+ */
+class ConnectionCounter : public broker::ConnectionObserver
+{
+private:
+    typedef std::map<std::string, uint32_t> connectCountsMap_t;
+    enum CONNECTION_PROGRESS { C_CREATED=1, C_OPENED=2 };
+
+    Acl&             acl;
+    uint32_t         nameLimit;
+    uint32_t         hostLimit;
+    qpid::sys::Mutex dataLock;
+
+    connectCountsMap_t connectProgressMap;
+    connectCountsMap_t connectByNameMap;
+    connectCountsMap_t connectByHostMap;
+
+    std::string getClientHost(const std::string mgmtId);
+
+    bool limitCheckLH(connectCountsMap_t& theMap,
+                      const std::string& theName,
+                      uint32_t theLimit);
+
+    void releaseLH(connectCountsMap_t& theMap,
+                   const std::string& theName,
+                   uint32_t theLimit);
+
+public:
+    ConnectionCounter(Acl& acl, uint32_t nl, uint32_t hl);
+    ~ConnectionCounter();
+
+    void connection(broker::Connection& connection);
+    void     opened(broker::Connection& connection);
+    void     closed(broker::Connection& connection);
+
+};
+
+}} // namespace qpid::ha
+
+#endif  /*!QPID_ACL_CONNECTIONCOUNTER_H*/
diff --git a/cpp/src/qpid/acl/AclData.cpp b/cpp/src/qpid/acl/AclData.cpp
index 06fc223a73..da7f240a9b 100644
--- a/cpp/src/qpid/acl/AclData.cpp
+++ b/cpp/src/qpid/acl/AclData.cpp
@@ -24,152 +24,302 @@
 namespace qpid {
 namespace acl {
 
-    AclData::AclData():decisionMode(qpid::acl::DENY),transferAcl(false),aclSource("UNKNOWN")
+    //
+    // constructor
+    //
+    AclData::AclData():
+        decisionMode(qpid::acl::DENY),
+        transferAcl(false),
+        aclSource("UNKNOWN")
     {
-        for (unsigned int cnt=0; cnt< qpid::acl::ACTIONSIZE; cnt++){
+        for (unsigned int cnt=0; cnt< qpid::acl::ACTIONSIZE; cnt++)
+        {
             actionList[cnt]=0;
         }
-
     }
 
+
+    //
+    // clear
+    //
     void AclData::clear ()
     {
-        for (unsigned int cnt=0; cnt< qpid::acl::ACTIONSIZE; cnt++){
-            if (actionList[cnt]){
+        for (unsigned int cnt=0; cnt< qpid::acl::ACTIONSIZE; cnt++)
+        {
+            if (actionList[cnt])
+            {
                 for (unsigned int cnt1=0; cnt1< qpid::acl::OBJECTSIZE; cnt1++)
-                    delete actionList[cnt][cnt1]; 
+                    delete actionList[cnt][cnt1];
             }
             delete[] actionList[cnt];
         }
-
     }
 
-    bool AclData::matchProp(const std::string & src, const std::string& src1)
+
+    //
+    // matchProp
+    //
+    // Compare a rule's property name with a lookup name,
+    // The rule's name may contain a trailing '*' to specify a wildcard match.
+    //
+    bool AclData::matchProp(const std::string& ruleStr,
+                            const std::string& lookupStr)
     {
-        // allow wildcard on the end of strings...
-        if (src.data()[src.size()-1]=='*') {
-            return (src.compare(0, src.size()-1, src1, 0,src.size()-1  ) == 0);
-        } else {
-            return (src.compare(src1)==0) ;
+        // allow wildcard on the end of rule strings...
+        if (ruleStr.data()[ruleStr.size()-1]=='*')
+        {
+            return ruleStr.compare(0,
+                                   ruleStr.size()-1,
+                                   lookupStr,
+                                   0,
+                                   ruleStr.size()-1  ) == 0;
+        }
+        else
+        {
+            return ruleStr.compare(lookupStr) == 0;
         }
     }
 
-    AclResult AclData::lookup(const std::string& id, const Action& action, const ObjectType& objType,
-        const std::string& name, std::map<Property, std::string>* params) {
-
-            QPID_LOG(debug, "ACL: Lookup for id:" << id << " action:" << AclHelper::getActionStr((Action) action)
-                << " objectType:" << AclHelper::getObjectTypeStr((ObjectType) objType) << " name:" << name
-                << " with params " << AclHelper::propertyMapToString(params));
-
-            AclResult aclresult = decisionMode;
-            if (actionList[action] && actionList[action][objType]) {
-                AclData::actObjItr itrRule = actionList[action][objType]->find(id);
-                if (itrRule == actionList[action][objType]->end())
-                    itrRule = actionList[action][objType]->find("*");
-
-                if (itrRule != actionList[action][objType]->end()) {
-
-                    QPID_LOG(debug, "ACL: checking the following rules for : " << itrRule->first );
-
-                    //loop the vector
-                    for (ruleSetItr i = itrRule->second.begin(); i < itrRule->second.end(); i++) {
-                        QPID_LOG(debug, "ACL: checking rule " <<  i->toString());
-                        // loop the names looking for match
-                        bool match = true;
-                        for (propertyMapItr pMItr = i->props.begin(); (pMItr != i->props.end()) && match; pMItr++) {
-                            //match name is exists first
-                            if (pMItr->first == acl::PROP_NAME) {
-                                if (matchProp(pMItr->second, name)){
-                                    QPID_LOG(debug, "ACL: name '" << name << "' matched with name '"
-                                        << pMItr->second << "' given in the rule");
-                                }else{
+
+    //
+    // lookup
+    //
+    // The ACL main business logic function of matching rules and declaring
+    // an allow or deny result.
+    //
+    AclResult AclData::lookup(
+        const std::string&               id,
+        const Action&                    action,
+        const ObjectType&                objType,
+        const std::string&               name,
+        std::map<Property, std::string>* params)
+    {
+        QPID_LOG(debug, "ACL: Lookup for id:" << id
+                    << " action:" << AclHelper::getActionStr((Action) action)
+                    << " objectType:" << AclHelper::getObjectTypeStr((ObjectType) objType)
+                    << " name:" << name
+                    << " with params " << AclHelper::propertyMapToString(params));
+
+        // A typical log looks like:
+        // ACL: Lookup for id:bob@QPID action:create objectType:queue name:q2
+        //  with params { durable=false passive=false autodelete=false
+        //  exclusive=false alternate= policytype= maxqueuesize=0
+        //  maxqueuecount=0 }
+
+        // Default result is blanket decision mode for the entire ACL list.
+        AclResult aclresult = decisionMode;
+
+        // Test for lists of rules at the intersection of the Action & Object
+        if (actionList[action] && actionList[action][objType])
+        {
+            // Find the list of rules for this actorId
+            AclData::actObjItr itrRule = actionList[action][objType]->find(id);
+
+            // If individual actorId not found then find a rule set for '*'.
+            if (itrRule == actionList[action][objType]->end())
+                itrRule = actionList[action][objType]->find("*");
+
+            if (itrRule != actionList[action][objType]->end())
+            {
+                // A list of rules exists for this actor/action/object tuple.
+                // Iterate the rule set to search for a matching rule.
+                ruleSetItr rsItr = itrRule->second.end();
+                for (int cnt = itrRule->second.size(); cnt != 0; cnt--)
+                {
+                    rsItr--;
+
+                    QPID_LOG(debug, "ACL: checking rule " <<  rsItr->toString());
+
+                    bool match = true;
+                    bool limitChecked = true;
+
+                    // Iterate this rule's properties. A 'match' is true when
+                    // all of the rule's properties are found to be satisfied
+                    // in the lookup param list. The lookup may specify things
+                    // (they usually do) that are not in the rule properties but
+                    // these things don't interfere with the rule match.
+
+                    for (specPropertyMapItr rulePropMapItr  = rsItr->props.begin();
+                                           (rulePropMapItr != rsItr->props.end()) && match;
+                                            rulePropMapItr++)
+                    {
+                        // The rule property map's NAME property is given in
+                        // the calling args and not in the param map.
+                        if (rulePropMapItr->first == acl::SPECPROP_NAME)
+                        {
+                            if (matchProp(rulePropMapItr->second, name))
+                            {
+                                QPID_LOG(debug, "ACL: lookup name '" << name
+                                    << "' matched with rule name '"
+                                    << rulePropMapItr->second << "'");
+                            }
+                            else
+                            {
+                                match = false;
+                                QPID_LOG(debug, "ACL: lookup name '" << name
+                                    << "' didn't match with rule name '"
+                                    << rulePropMapItr->second << "'");
+                            }
+                        }
+                        else
+                        {
+                            if (params)
+                            {
+                                // The rule's property map non-NAME properties
+                                //  found in the lookup's params list.
+                                // In some cases the param's index is not the same
+                                //  as rule's index.
+                                propertyMapItr lookupParamItr;
+                                switch (rulePropMapItr->first)
+                                {
+                                case acl::SPECPROP_MAXQUEUECOUNTUPPERLIMIT:
+                                case acl::SPECPROP_MAXQUEUECOUNTLOWERLIMIT:
+                                    lookupParamItr = params->find(PROP_MAXQUEUECOUNT);
+                                    break;
+
+                                case acl::SPECPROP_MAXQUEUESIZEUPPERLIMIT:
+                                case acl::SPECPROP_MAXQUEUESIZELOWERLIMIT:
+                                    lookupParamItr = params->find(PROP_MAXQUEUESIZE);
+                                    break;
+
+                                default:
+                                    lookupParamItr = params->find((Property)rulePropMapItr->first);
+                                    break;
+                                };
+
+                                if (lookupParamItr == params->end())
+                                {
+                                    // Now the rule has a specified property
+                                    // that does not exist in the caller's
+                                    // lookup params list.
+                                    // This rule does not match.
                                     match = false;
-                                    QPID_LOG(debug, "ACL: name '" << name << "' didn't match with name '"
-                                        << pMItr->second << "' given in the rule");
+                                    QPID_LOG(debug, "ACL: lookup parameter map doesn't contain the rule property '"
+                                        << AclHelper::getPropertyStr(rulePropMapItr->first) << "'");
                                 }
-                            } else if (params) { //match pMItr against params
-                                propertyMapItr paramItr = params->find(pMItr->first);
-                                if (paramItr == params->end()) {
-                                    match = false;
-                                    QPID_LOG(debug, "ACL: the given parameter map in lookup doesn't contain the property '"
-                                        << AclHelper::getPropertyStr(pMItr->first) << "'");
-                                }else if ( pMItr->first == acl::PROP_MAXQUEUECOUNT || pMItr->first == acl::PROP_MAXQUEUESIZE ) {
-                                    if ( pMItr->first == paramItr->first ) {
-
-                                        uint64_t aclMax = 0;
-                                        uint64_t paramMax = 0;
-
-                                        try{                      
-                                            aclMax = boost::lexical_cast<uint64_t>(pMItr->second);
-                                        }catch(const boost::bad_lexical_cast&){
-                                            match = false;  
-                                            QPID_LOG(error,"Error evaluating rule. " << 
-                                                "Illegal value given in ACL source <" << aclSource <<
-                                                "> for property '" <<  
-                                                AclHelper::getPropertyStr(pMItr->first) << "' : " << 
-                                                boost::lexical_cast<std::string>(pMItr->second));
-                                            break;  
-                                        } 
-
-                                        try{
-                                            paramMax = boost::lexical_cast<uint64_t>(paramItr->second);
-                                        }catch(const boost::bad_lexical_cast&){
+                                else
+                                {
+                                    // Now account for the business of rules
+                                    // whose property indexes are mismatched.
+                                    switch (rulePropMapItr->first)
+                                    {
+                                    case acl::SPECPROP_MAXQUEUECOUNTUPPERLIMIT:
+                                    case acl::SPECPROP_MAXQUEUESIZEUPPERLIMIT:
+                                        limitChecked &=
+                                            compareIntMax(
+                                                rulePropMapItr->first,
+                                                boost::lexical_cast<std::string>(rulePropMapItr->second),
+                                                boost::lexical_cast<std::string>(lookupParamItr->second));
+                                        break;
+
+                                    case acl::SPECPROP_MAXQUEUECOUNTLOWERLIMIT:
+                                    case acl::SPECPROP_MAXQUEUESIZELOWERLIMIT:
+                                        limitChecked &=
+                                            compareIntMin(
+                                                rulePropMapItr->first,
+                                                boost::lexical_cast<std::string>(rulePropMapItr->second),
+                                                boost::lexical_cast<std::string>(lookupParamItr->second));
+                                        break;
+
+                                    default:
+                                        if (matchProp(rulePropMapItr->second, lookupParamItr->second))
+                                        {
+                                            QPID_LOG(debug, "ACL: the pair("
+                                                << AclHelper::getPropertyStr(lookupParamItr->first)
+                                                << "," << lookupParamItr->second
+                                                << ") given in lookup matched the pair("
+                                                << AclHelper::getPropertyStr(rulePropMapItr->first) << ","
+                                                << rulePropMapItr->second
+                                                << ") given in the rule");
+                                        }
+                                        else
+                                        {
                                             match = false;
-                                            QPID_LOG(error,"Error evaluating rule. " <<
-                                                "Illegal value given in lookup for property '" <<  
-                                                AclHelper::getPropertyStr(pMItr->first) << "' : " << 
-                                                boost::lexical_cast<std::string>(paramItr->second));
-                                            break;                                                                
-                                        } 
-
-                                        QPID_LOG(debug, "ACL: Numeric comparison for property " <<
-                                            AclHelper::getPropertyStr(paramItr->first)  <<
-                                            " (value given in lookup = " << 
-                                            boost::lexical_cast<std::string>(paramItr->second) << 
-                                            ", value give in rule = " <<
-                                            boost::lexical_cast<std::string>(pMItr->second) << " )");
-
-                                        if (( aclMax ) && ( paramMax == 0 || paramMax > aclMax)){
-                                            match = decisionMode == qpid::acl::ALLOW ;
-                                            QPID_LOG(debug, "ACL: Limit exceeded and match=" << 
-                                                (match ? "true": "false") <<
-                                                " as decision mode is " << AclHelper::getAclResultStr(decisionMode));
+                                            QPID_LOG(debug, "ACL: the pair("
+                                                << AclHelper::getPropertyStr(lookupParamItr->first)
+                                                << "," << lookupParamItr->second
+                                                << ") given in lookup doesn't match the pair("
+                                                << AclHelper::getPropertyStr(rulePropMapItr->first)
+                                                << "," << rulePropMapItr->second
+                                                << ") given in the rule");
                                         }
-                                    }
-                                }else if (matchProp(pMItr->second, paramItr->second)) {
-                                    QPID_LOG(debug, "ACL: the pair("
-                                        << AclHelper::getPropertyStr(paramItr->first) << "," << paramItr->second
-                                        << ") given in lookup matched the pair("
-                                        << AclHelper::getPropertyStr(pMItr->first) << "," << pMItr->second << ") given in the rule"); 
-                                } else {
-                                    QPID_LOG(debug, "ACL: the pair("
-                                        << AclHelper::getPropertyStr(paramItr->first) << "," << paramItr->second
-                                        << ") given in lookup doesn't match the pair("
-                                        << AclHelper::getPropertyStr(pMItr->first) << "," << pMItr->second << ") given in the rule");
-                                    match = false;
-                                } 
+                                        break;
+                                    };
+                                }
+                            }
+                            else
+                            {
+                                // params don't exist.
                             }
                         }
-                        if (match)
+                    }
+                    if (match)
+                    {
+                        aclresult = rsItr->ruleMode;
+                        if (!limitChecked)
                         {
-                            aclresult = getACLResult(i->logOnly, i->log);
-                            QPID_LOG(debug,"Successful match, the decision is:" << AclHelper::getAclResultStr(aclresult));
-                            return aclresult;
+                            // Now a lookup matched all rule properties but one
+                            //  of the numeric limit checks has failed.
+                            // Demote allow rules to corresponding deny rules.
+                            switch (aclresult)
+                            {
+                            case acl::ALLOW:
+                                aclresult = acl::DENY;
+                                break;
+                            case acl::ALLOWLOG:
+                                aclresult = acl::DENYLOG;
+                                break;
+                            default:
+                                break;
+                            };
                         }
+                        QPID_LOG(debug,"ACL: Successful match, the decision is:"
+                            << AclHelper::getAclResultStr(aclresult));
+                        return aclresult;
+                    }
+                    else
+                    {
+                        // This rule did not match the requested lookup and
+                        // does not contribute to an ACL decision.
                     }
                 }
             }
+            else
+            {
+                // The Action-Object list has entries but not for this actorId
+                // nor for *.
+            }
+        }
+        else
+        {
+            // The Action-Object list has no entries.
+        }
 
-            QPID_LOG(debug,"No successful match, defaulting to the decision mode " << AclHelper::getAclResultStr(aclresult));
-            return aclresult;
+        QPID_LOG(debug,"ACL: No successful match, defaulting to the decision mode "
+            << AclHelper::getAclResultStr(aclresult));
+        return aclresult;
     }
 
-    AclResult AclData::lookup(const std::string& id, const Action& action, const ObjectType& objType, const std::string& /*Exchange*/ name, const std::string& RoutingKey)
+
+    //
+    // lookup
+    //
+    // The ACL main business logic function of matching rules and declaring
+    // an allow or deny result.
+    //
+    AclResult AclData::lookup(
+        const std::string&              id,
+        const Action&                   action,
+        const ObjectType&               objType,
+        const std::string& /*Exchange*/ name,
+        const std::string&              routingKey)
     {
 
-        QPID_LOG(debug, "ACL: Lookup for id:" << id << " action:" << AclHelper::getActionStr((Action) action)
-            << " objectType:" << AclHelper::getObjectTypeStr((ObjectType) objType) << " exchange name:" << name
-            << " with routing key " << RoutingKey);
+        QPID_LOG(debug, "ACL: Lookup for id:" << id
+            << " action:" << AclHelper::getActionStr((Action) action)
+            << " objectType:" << AclHelper::getObjectTypeStr((ObjectType) objType)
+            << " exchange name:" << name
+            << " with routing key " << routingKey);
 
         AclResult aclresult = decisionMode;
 
@@ -179,83 +329,183 @@ namespace acl {
             if (itrRule == actionList[action][objType]->end())
                 itrRule = actionList[action][objType]->find("*");
 
-            if (itrRule != actionList[action][objType]->end() ) {
-
-                QPID_LOG(debug, "ACL: checking the following rules for : " << itrRule->first );
-
+            if (itrRule != actionList[action][objType]->end() )
+            {
                 //loop the vector
-                for (ruleSetItr i=itrRule->second.begin(); i<itrRule->second.end(); i++) {
-                    QPID_LOG(debug, "ACL: checking rule " <<  i->toString());
+                ruleSetItr rsItr = itrRule->second.end();
+                for (int cnt = itrRule->second.size(); cnt != 0; cnt--)
+                {
+                    rsItr--;
+
+                    QPID_LOG(debug, "ACL: checking rule " <<  rsItr->toString());
 
                     // loop the names looking for match
                     bool match =true;
-                    for (propertyMapItr pMItr = i->props.begin(); (pMItr != i->props.end()) && match; pMItr++)
+                    for (specPropertyMapItr pMItr  = rsItr->props.begin();
+                                           (pMItr != rsItr->props.end()) && match;
+                                            pMItr++)
                     {
                         //match name is exists first
-                        if (pMItr->first == acl::PROP_NAME){
-                            if (matchProp(pMItr->second, name)){  							     
-                                QPID_LOG(debug, "ACL: name '" << name << "' matched with name '"
-                                    << pMItr->second << "' given in the rule");
+                        switch (pMItr->first)
+                        {
+                        case acl::SPECPROP_NAME:
+                            if (matchProp(pMItr->second, name))
+                            {
+                                QPID_LOG(debug, "ACL: lookup exchange name '"
+                                    << name << "' matched with rule name '"
+                                    << pMItr->second << "'");
 
-                            }else{
+                            }
+                            else
+                            {
                                 match= false;
-                                QPID_LOG(debug, "ACL: name '" << name << "' didn't match with name '"
-                                    << pMItr->second << "' given in the rule");
-                            }    
-                        }else if (pMItr->first == acl::PROP_ROUTINGKEY){
-                            if (matchProp(pMItr->second, RoutingKey)){  
-                                QPID_LOG(debug, "ACL: name '" << name << "' matched with routing_key '"
-                                    << pMItr->second << "' given in the rule");    
-                            }else{
+                                QPID_LOG(debug, "ACL: lookup exchange name '"
+                                    << name << "' did not match with rule name '"
+                                    << pMItr->second << "'");
+                            }
+                            break;
+
+                        case acl::SPECPROP_ROUTINGKEY:
+                            if (matchProp(pMItr->second, routingKey))
+                            {
+                                QPID_LOG(debug, "ACL: lookup key name '"
+                                    << routingKey << "' matched with rule routing key '"
+                                    << pMItr->second << "'");
+                            }
+                            else
+                            {
                                 match= false;
-                                QPID_LOG(debug, "ACL: name '" << name << "' didn't match with routing_key '"
-                                    << pMItr->second << "' given in the rule");
-                            } 
-                        }
+                                QPID_LOG(debug, "ACL: lookup key name '"
+                                    << routingKey << "' did not match with rule routing key '"
+                                    << pMItr->second << "'");
+                            }
+                            break;
+
+                        default:
+                            // Don't care
+                            break;
+                        };
                     }
                     if (match){
-                        aclresult = getACLResult(i->logOnly, i->log);
-                        QPID_LOG(debug,"Successful match, the decision is:" << AclHelper::getAclResultStr(aclresult));
+                        aclresult = rsItr->ruleMode;
+                        QPID_LOG(debug,"ACL: Successful match, the decision is:"
+                            << AclHelper::getAclResultStr(aclresult));
                         return aclresult;
                     }
                 }
             }
         }
-        QPID_LOG(debug,"No successful match, defaulting to the decision mode " << AclHelper::getAclResultStr(aclresult));
+        QPID_LOG(debug,"ACL: No successful match, defaulting to the decision mode "
+            << AclHelper::getAclResultStr(aclresult));
         return aclresult;
 
     }
 
 
-    AclResult AclData::getACLResult(bool logOnly, bool log)
+    //
+    //
+    //
+    AclData::~AclData()
+    {
+        clear();
+    }
+
+
+    //
+    // Limit check a MAX int limit
+    //
+    bool AclData::compareIntMax(const qpid::acl::SpecProperty theProperty,
+                                const std::string             theAclValue,
+                                const std::string             theLookupValue)
     {
-        switch (decisionMode)
+        uint64_t aclMax   (0);
+        uint64_t paramMax (0);
+
+        try
         {
-        case qpid::acl::ALLOWLOG:
-        case qpid::acl::ALLOW:
-            if (logOnly) return qpid::acl::ALLOWLOG;
-            if (log)
-                return qpid::acl::DENYLOG;
-            else
-                return qpid::acl::DENY;
+            aclMax = boost::lexical_cast<uint64_t>(theAclValue);
+        }
+        catch(const boost::bad_lexical_cast&)
+        {
+            assert (false);
+            return false;
+        }
+
+        try
+        {
+            paramMax = boost::lexical_cast<uint64_t>(theLookupValue);
+        }
+        catch(const boost::bad_lexical_cast&)
+        {
+            QPID_LOG(error,"ACL: Error evaluating rule. "
+                << "Illegal value given in lookup for property '"
+                << AclHelper::getPropertyStr(theProperty)
+                << "' : " << theLookupValue);
+            return false;
+        }
 
+        QPID_LOG(debug, "ACL: Numeric greater-than comparison for property "
+            << AclHelper::getPropertyStr(theProperty)
+            << " (value given in lookup = " << theLookupValue
+            << ", value give in rule = " << theAclValue << " )");
 
-        case qpid::acl::DENYLOG:
-        case qpid::acl::DENY:
-            if (logOnly) return qpid::acl::DENYLOG;
-            if (log)
-                return qpid::acl::ALLOWLOG;
-            else
-                return qpid::acl::ALLOW;
+        if (( aclMax ) && ( paramMax == 0 || paramMax > aclMax))
+        {
+            QPID_LOG(debug, "ACL: Max limit exceeded for property '"
+                << AclHelper::getPropertyStr(theProperty) << "'");
+            return false;
         }
 
-        QPID_LOG(error, "ACL Decision Failed, setting DENY");
-        return qpid::acl::DENY;
+        return true;
     }
 
-    AclData::~AclData()
+
+    //
+    // limit check a MIN int limit
+    //
+    bool AclData::compareIntMin(const qpid::acl::SpecProperty theProperty,
+                                const std::string             theAclValue,
+                                const std::string             theLookupValue)
     {
-        clear();
+        uint64_t aclMin   (0);
+        uint64_t paramMin (0);
+
+        try
+        {
+            aclMin = boost::lexical_cast<uint64_t>(theAclValue);
+        }
+        catch(const boost::bad_lexical_cast&)
+        {
+            assert (false);
+            return false;
+        }
+
+        try
+        {
+            paramMin = boost::lexical_cast<uint64_t>(theLookupValue);
+        }
+        catch(const boost::bad_lexical_cast&)
+        {
+            QPID_LOG(error,"ACL: Error evaluating rule. "
+                << "Illegal value given in lookup for property '"
+                << AclHelper::getPropertyStr(theProperty)
+                << "' : " << theLookupValue);
+            return false;
+        }
+
+        QPID_LOG(debug, "ACL: Numeric less-than comparison for property "
+            << AclHelper::getPropertyStr(theProperty)
+            << " (value given in lookup = " << theLookupValue
+            << ", value give in rule = " << theAclValue << " )");
+
+        if (( aclMin ) && ( paramMin == 0 || paramMin < aclMin))
+        {
+            QPID_LOG(debug, "ACL: Min limit exceeded for property '"
+                << AclHelper::getPropertyStr(theProperty) << "'");
+            return false;
+        }
+
+        return true;
     }
 
-}} 
+}}
diff --git a/cpp/src/qpid/acl/AclData.h b/cpp/src/qpid/acl/AclData.h
index 81125fdcbc..1c1cb3e9c6 100644
--- a/cpp/src/qpid/acl/AclData.h
+++ b/cpp/src/qpid/acl/AclData.h
@@ -33,50 +33,91 @@ class AclData {
 public:
 
     typedef std::map<qpid::acl::Property, std::string> propertyMap;
-    typedef propertyMap::const_iterator propertyMapItr;
+    typedef propertyMap::const_iterator                propertyMapItr;
+
+    typedef std::map<qpid::acl::SpecProperty, std::string> specPropertyMap;
+    typedef specPropertyMap::const_iterator                specPropertyMapItr;
+
+    //
+    // rule
+    //
+    // Created by AclReader and stored in a ruleSet vector for subsequent
+    //  run-time lookup matching and allow/deny decisions.
+    // RuleSet vectors are indexed by Action-Object-actorId so these
+    //  attributes are not part of a rule.
+    // A single ACL file entry may create many rule entries in
+    //  many ruleset vectors.
+    //
     struct rule {
 
-        bool log;
-        bool logOnly;  // this is a rule is to log only
+        int                   rawRuleNum;   // rule number in ACL file
+        qpid::acl::AclResult  ruleMode;     // combined allow/deny log/nolog
+        specPropertyMap       props;        //
 
-        // key value map
-        //??
-        propertyMap props;
 
-
-        rule (propertyMap& p):log(false),logOnly(false),props(p) {};
+        rule (int ruleNum, qpid::acl::AclResult res, specPropertyMap& p) :
+            rawRuleNum(ruleNum),
+            ruleMode(res),
+            props(p)
+            {};
 
         std::string toString () const {
             std::ostringstream ruleStr;
-            ruleStr << "[log=" << log << ", logOnly=" << logOnly << " props{";
-            for (propertyMapItr pMItr = props.begin(); pMItr != props.end(); pMItr++) {
-                ruleStr << " " << AclHelper::getPropertyStr((Property) pMItr-> first) << "=" << pMItr->second;
+            ruleStr << "[rule " << rawRuleNum
+                    << " ruleMode = " << AclHelper::getAclResultStr(ruleMode)
+                    << " props{";
+            for (specPropertyMapItr pMItr  = props.begin();
+                                    pMItr != props.end();
+                                    pMItr++) {
+                ruleStr << " "
+                        << AclHelper::getPropertyStr((SpecProperty) pMItr-> first)
+                        << "=" << pMItr->second;
             }
             ruleStr << " }]";
             return ruleStr.str();
         }
     };
-    typedef  std::vector<rule> ruleSet;
-    typedef  ruleSet::const_iterator ruleSetItr;
-    typedef  std::map<std::string, ruleSet > actionObject; // user 
-    typedef  actionObject::iterator actObjItr;
-    typedef  actionObject* aclAction;
 
-    // Action*[] -> Object*[] -> map<user -> set<Rule> >
-    aclAction* actionList[qpid::acl::ACTIONSIZE];
-    qpid::acl::AclResult decisionMode;  // determines if the rule set is a deny or allow mode. 
-    bool transferAcl;
-    std::string aclSource; 
+    typedef  std::vector<rule>               ruleSet;
+    typedef  ruleSet::const_iterator         ruleSetItr;
+    typedef  std::map<std::string, ruleSet > actionObject; // user
+    typedef  actionObject::iterator          actObjItr;
+    typedef  actionObject*                   aclAction;
 
-    AclResult lookup(const std::string& id, const Action& action, const ObjectType& objType, const std::string& name, std::map<Property, std::string>* params=0);
-    AclResult lookup(const std::string& id, const Action& action, const ObjectType& objType, const std::string& ExchangeName, const std::string& RoutingKey);
-    AclResult getACLResult(bool logOnly, bool log);
+    // Action*[] -> Object*[] -> map<user -> set<Rule> >
+    aclAction*           actionList[qpid::acl::ACTIONSIZE];
+    qpid::acl::AclResult decisionMode;  // allow/deny[-log] if no matching rule found
+    bool                 transferAcl;
+    std::string          aclSource;
+
+    AclResult lookup(
+        const std::string&               id,        // actor id
+        const Action&                    action,
+        const ObjectType&                objType,
+        const std::string&               name,      // object name
+        std::map<Property, std::string>* params=0);
+
+    AclResult lookup(
+        const std::string&               id,        // actor id
+        const Action&                    action,
+        const ObjectType&                objType,
+        const std::string&               ExchangeName,
+        const std::string&               RoutingKey);
 
     bool matchProp(const std::string & src, const std::string& src1);
     void clear ();
 
     AclData();
     virtual ~AclData();
+
+private:
+    bool compareIntMax(const qpid::acl::SpecProperty theProperty,
+                       const std::string             theAclValue,
+                       const std::string             theLookupValue);
+
+    bool compareIntMin(const qpid::acl::SpecProperty theProperty,
+                       const std::string             theAclValue,
+                       const std::string             theLookupValue);
 };
 
 }} // namespace qpid::acl
diff --git a/cpp/src/qpid/acl/AclPlugin.cpp b/cpp/src/qpid/acl/AclPlugin.cpp
index d611797c49..6c18cd2749 100644
--- a/cpp/src/qpid/acl/AclPlugin.cpp
+++ b/cpp/src/qpid/acl/AclPlugin.cpp
@@ -40,7 +40,9 @@ struct AclOptions : public Options {
 
     AclOptions(AclValues& v) : Options("ACL Options"), values(v) {
         addOptions()
-            ("acl-file", optValue(values.aclFile, "FILE"), "The policy file to load from, loaded from data dir");
+            ("acl-file",           optValue(values.aclFile, "FILE"), "The policy file to load from, loaded from data dir")
+            ("acl-max-connect-per-user", optValue(values.aclMaxConnectPerUser, "N"), "The maximum number of connections allowed per user")
+            ("acl-max-connect-per-ip"  , optValue(values.aclMaxConnectPerIp, "N"),   "The maximum number of connections allowed per host IP address");
     }
 };
 
diff --git a/cpp/src/qpid/acl/AclReader.cpp b/cpp/src/qpid/acl/AclReader.cpp
index 74358a20c1..80debf1bd1 100644
--- a/cpp/src/qpid/acl/AclReader.cpp
+++ b/cpp/src/qpid/acl/AclReader.cpp
@@ -49,7 +49,7 @@ namespace acl {
         objStatus = ALL;
     }
 
-    bool AclReader::aclRule::addProperty(const Property p, const std::string v) {
+    bool AclReader::aclRule::addProperty(const SpecProperty p, const std::string v) {
         return props.insert(propNvPair(p, v)).second;
     }
 
@@ -85,146 +85,108 @@ namespace acl {
 
     void AclReader::loadDecisionData(boost::shared_ptr<AclData> d) {
         d->clear();
-        QPID_LOG(debug, "ACL Load Rules");
-        int cnt = rules.size();
+        QPID_LOG(debug, "ACL: Load Rules");
         bool foundmode = false;
 
-        for (rlCitr i = rules.end(); cnt; cnt--) {
+        rlCitr i = rules.end();
+        for (int cnt = rules.size(); cnt; cnt--) {
             i--;
-            QPID_LOG(debug, "ACL Processing " << std::setfill(' ') << std::setw(2)
+            QPID_LOG(debug, "ACL: Processing " << std::setfill(' ') << std::setw(2)
                 << cnt << " " << (*i)->toString());
 
             if (!foundmode && (*i)->actionAll && (*i)->names.size() == 1
                 && (*((*i)->names.begin())).compare("*") == 0) {
                     d->decisionMode = (*i)->res;
-                    QPID_LOG(debug, "ACL FoundMode "
+                    QPID_LOG(debug, "ACL: FoundMode "
                         << AclHelper::getAclResultStr(d->decisionMode));
                     foundmode = true;
             } else {
-                AclData::rule rule((*i)->props);
-                bool addrule = true;
-
-                switch ((*i)->res) {
-                case qpid::acl::ALLOWLOG:
-                    rule.log = true;
-                    if (d->decisionMode == qpid::acl::ALLOW ||
-                        d->decisionMode == qpid::acl::ALLOWLOG)
-                        rule.logOnly = true;
-                    break;
-                case qpid::acl::ALLOW:
-                    if (d->decisionMode == qpid::acl::ALLOW ||
-                        d->decisionMode == qpid::acl::ALLOWLOG)
-                        addrule = false;
-                    break;
-                case qpid::acl::DENYLOG:
-                    rule.log = true;
-                    if (d->decisionMode == qpid::acl::DENY ||
-                        d->decisionMode == qpid::acl::DENYLOG)
-                        rule.logOnly = true;
-                    break;
-                case qpid::acl::DENY:
-                    if (d->decisionMode == qpid::acl::DENY ||
-                        d->decisionMode == qpid::acl::DENYLOG)
-                        addrule = false;
-                    break;
-                default:
-                    throw Exception("Invalid ACL Result loading rules.");
-                }
+                AclData::rule rule(cnt, (*i)->res, (*i)->props);
 
                 // Action -> Object -> map<user -> set<Rule> >
-                if (addrule) {
-                    std::ostringstream actionstr;
-                    for (int acnt = ((*i)->actionAll ? 0 : (*i)->action);
-                        acnt < acl::ACTIONSIZE;
-                        (*i)->actionAll ? acnt++ : acnt = acl::ACTIONSIZE) {
-
-                            if (acnt == acl::ACT_PUBLISH)
-                                d->transferAcl = true; // we have transfer ACL
-
-                            actionstr << AclHelper::getActionStr((Action) acnt) << ",";
-
-                            //find the Action, create if not exist
-                            if (d->actionList[acnt] == NULL) {
-                                d->actionList[acnt] =
-                                    new AclData::aclAction[qpid::acl::OBJECTSIZE];
-                                for (int j = 0; j < qpid::acl::OBJECTSIZE; j++)
-                                    d->actionList[acnt][j] = NULL;
-                            }
-
-                            // optimize this loop to limit to valid options only!!
-                            for (int ocnt = ((*i)->objStatus != aclRule::VALUE ? 0
-                                : (*i)->object);
-                                ocnt < acl::OBJECTSIZE;
-                            (*i)->objStatus != aclRule::VALUE ? ocnt++ : ocnt = acl::OBJECTSIZE) {
-
-                                //find the Object, create if not exist
-                                if (d->actionList[acnt][ocnt] == NULL)
-                                    d->actionList[acnt][ocnt] =
-                                    new AclData::actionObject;
-
-                                // add users and Rule to object set
-                                bool allNames = false;
-                                // check to see if names.begin is '*'
-                                if ((*(*i)->names.begin()).compare("*") == 0)
-                                    allNames = true;
-
-                                for (nsCitr itr = (allNames ? names.begin()
-                                    : (*i)->names.begin());
-                                    itr != (allNames ? names.end() : (*i)->names.end());
-                                itr++) {
-
-                                    AclData::actObjItr itrRule =
-                                        d->actionList[acnt][ocnt]->find(*itr);
-
-                                    if (itrRule == d->actionList[acnt][ocnt]->end()) {
-                                        AclData::ruleSet rSet;
-                                        rSet.push_back(rule);
-                                        d->actionList[acnt][ocnt]->insert
-                                            (make_pair(std::string(*itr), rSet));
-                                    } else {
-                                        // TODO add code to check for dead rules
-                                        // allow peter create queue name=tmp <-- dead rule!!
-                                        // allow peter create queue
-
-                                        itrRule->second.push_back(rule);
-                                    }
-                                }
-
-                            }
+                std::ostringstream actionstr;
+                for (int acnt = ((*i)->actionAll ? 0 : (*i)->action);
+                    acnt < acl::ACTIONSIZE;
+                    (*i)->actionAll ? acnt++ : acnt = acl::ACTIONSIZE) {
+
+                    if (acnt == acl::ACT_PUBLISH)
+                        d->transferAcl = true; // we have transfer ACL
+
+                    actionstr << AclHelper::getActionStr((Action) acnt) << ",";
+
+                    //find the Action, create if not exist
+                    if (d->actionList[acnt] == NULL) {
+                        d->actionList[acnt] =
+                            new AclData::aclAction[qpid::acl::OBJECTSIZE];
+                        for (int j = 0; j < qpid::acl::OBJECTSIZE; j++)
+                            d->actionList[acnt][j] = NULL;
                     }
 
-                    std::ostringstream objstr;
-                    for (int ocnt = ((*i)->objStatus != aclRule::VALUE ? 0 : (*i)->object);
+                    // TODO: optimize this loop to limit to valid options only!!
+                    for (int ocnt = ((*i)->objStatus != aclRule::VALUE ? 0
+                        : (*i)->object);
                         ocnt < acl::OBJECTSIZE;
-                        (*i)->objStatus != aclRule::VALUE ? ocnt++ : ocnt = acl::OBJECTSIZE) {
-                            objstr << AclHelper::getObjectTypeStr((ObjectType) ocnt) << ",";
+                    (*i)->objStatus != aclRule::VALUE ? ocnt++ : ocnt = acl::OBJECTSIZE) {
+
+                        //find the Object, create if not exist
+                        if (d->actionList[acnt][ocnt] == NULL)
+                            d->actionList[acnt][ocnt] =
+                                new AclData::actionObject;
+
+                        // add users and Rule to object set
+                        bool allNames = false;
+                        // check to see if names.begin is '*'
+                        if ((*(*i)->names.begin()).compare("*") == 0)
+                            allNames = true;
+
+                        for (nsCitr itr  = (allNames ? names.begin() : (*i)->names.begin());
+                                    itr != (allNames ? names.end()   : (*i)->names.end());
+                                    itr++) {
+                            AclData::actObjItr itrRule =
+                                d->actionList[acnt][ocnt]->find(*itr);
+
+                            if (itrRule == d->actionList[acnt][ocnt]->end()) {
+                                AclData::ruleSet rSet;
+                                rSet.push_back(rule);
+                                d->actionList[acnt][ocnt]->insert
+                                    (make_pair(std::string(*itr), rSet));
+                            } else {
+                                // TODO add code to check for dead rules
+                                // allow peter create queue name=tmp <-- dead rule!!
+                                // allow peter create queue
+
+                                itrRule->second.push_back(rule);
+                            }
+                        }
                     }
+                }
 
-                    bool allNames = ((*(*i)->names.begin()).compare("*") == 0);
-                    std::ostringstream userstr;
-                    for (nsCitr itr = (allNames ? names.begin() : (*i)->names.begin());
-                        itr != (allNames ? names.end() : (*i)->names.end());
-                        itr++) {
-                            userstr << *itr << ",";
-                    }
+                std::ostringstream objstr;
+                for (int ocnt = ((*i)->objStatus != aclRule::VALUE ? 0 : (*i)->object);
+                         ocnt < acl::OBJECTSIZE;
+                                 (*i)->objStatus != aclRule::VALUE ? ocnt++ : ocnt = acl::OBJECTSIZE) {
+                        objstr << AclHelper::getObjectTypeStr((ObjectType) ocnt) << ",";
+                }
 
-                    QPID_LOG(debug, "ACL: Adding actions {" <<
-                        actionstr.str().substr(0,actionstr.str().length()-1)
-                        << "} to objects {" <<
-                        objstr.str().substr(0,objstr.str().length()-1)
-                        << "} with props " <<
-                        AclHelper::propertyMapToString(&rule.props)
-                        << " for users {" <<
-                        userstr.str().substr(0,userstr.str().length()-1)
-                        << "}" );
-                } else {
-                    QPID_LOG(debug, "ACL Skipping based on Mode:"
-                        << AclHelper::getAclResultStr(d->decisionMode));
+                bool allNames = ((*(*i)->names.begin()).compare("*") == 0);
+                std::ostringstream userstr;
+                for (nsCitr itr  = (allNames ? names.begin() : (*i)->names.begin());
+                            itr != (allNames ? names.end()   : (*i)->names.end());
+                            itr++) {
+                    userstr << *itr << ",";
                 }
-            }
 
+                QPID_LOG(debug, "ACL: Adding actions {" <<
+                    actionstr.str().substr(0,actionstr.str().length()-1)
+                    << "} to objects {" <<
+                    objstr.str().substr(0,objstr.str().length()-1)
+                    << "} with props " <<
+                    AclHelper::propertyMapToString(&rule.props)
+                    << " for users {" <<
+                    userstr.str().substr(0,userstr.str().length()-1)
+                    << "}" );
+            }
         }
-
     }
 
 
@@ -277,7 +239,7 @@ namespace acl {
             }
             ifs.close();
             if (err) return -3;
-            QPID_LOG(notice, "Read ACL file \"" <<  fn << "\"");
+            QPID_LOG(notice, "ACL: Read file \"" <<  fn << "\"");
         } catch (const std::exception& e) {
             errorStream << "Unable to read ACL file \"" << fn << "\": " << e.what();
             ifs.close();
@@ -410,8 +372,8 @@ namespace acl {
 
     // Debug aid
     void AclReader::printNames() const {
-        QPID_LOG(debug, "Group list: " << groups.size() << " groups found:" );
-        std::string tmp;
+        QPID_LOG(debug, "ACL: Group list: " << groups.size() << " groups found:" );
+        std::string tmp("ACL: ");
         for (gmCitr i=groups.begin(); i!= groups.end(); i++) {
             tmp += "  \"";
             tmp += i->first;
@@ -421,10 +383,10 @@ namespace acl {
                 tmp += *j;
             }
             QPID_LOG(debug, tmp);
-            tmp.clear();
+            tmp = "ACL: ";
         }
-        QPID_LOG(debug, "Name list: " << names.size() << " names found:" );
-        tmp.clear();
+        QPID_LOG(debug, "ACL: name list: " << names.size() << " names found:" );
+        tmp = "ACL: ";
         for (nsCitr k=names.begin(); k!=names.end(); k++) {
             tmp += " ";
             tmp += *k;
@@ -501,9 +463,9 @@ namespace acl {
                         << propNvp.first << "\". (Must be name=value)";
                     return false;
                 }
-                Property prop;
+                SpecProperty prop;
                 try {
-                    prop = AclHelper::getProperty(propNvp.first);
+                    prop = AclHelper::getSpecProperty(propNvp.first);
                 } catch (...) {
                     errorStream << ACL_FORMAT_ERR_LOG_PREFIX << "Line : " << lineNumber
                         << ", Unknown property \"" << propNvp.first << "\".";
@@ -532,10 +494,10 @@ namespace acl {
 
     // Debug aid
     void AclReader::printRules() const {
-        QPID_LOG(debug, "Rule list: " << rules.size() << " ACL rules found:");
-        int cnt = 0;
+        QPID_LOG(debug, "ACL: Rule list: " << rules.size() << " ACL rules found:");
+        int cnt = 1;
         for (rlCitr i=rules.begin(); i<rules.end(); i++,cnt++) {
-            QPID_LOG(debug, "  " << std::setfill(' ') << std::setw(2) << cnt << " " << (*i)->toString());
+            QPID_LOG(debug, "ACL:   " << std::setfill(' ') << std::setw(2) << cnt << " " << (*i)->toString());
         }
     }
 
diff --git a/cpp/src/qpid/acl/AclReader.h b/cpp/src/qpid/acl/AclReader.h
index 62c6f38f37..730013f4ed 100644
--- a/cpp/src/qpid/acl/AclReader.h
+++ b/cpp/src/qpid/acl/AclReader.h
@@ -33,65 +33,71 @@ namespace qpid {
 namespace acl {
 
 class AclReader {
-    typedef std::set<std::string> nameSet;
-    typedef nameSet::const_iterator nsCitr;
-    typedef boost::shared_ptr<nameSet> nameSetPtr;
-
-    typedef std::pair<std::string, nameSetPtr> groupPair;
-    typedef std::map<std::string, nameSetPtr> groupMap;
-    typedef groupMap::const_iterator gmCitr;
-    typedef std::pair<gmCitr, bool> gmRes;
-
-    typedef std::pair<Property, std::string> propNvPair;
-    typedef std::map<Property, std::string> propMap;
-    typedef propMap::const_iterator pmCitr;
-
+    typedef std::set<std::string>               nameSet;
+    typedef nameSet::const_iterator             nsCitr;
+    typedef boost::shared_ptr<nameSet>          nameSetPtr;
+
+    typedef std::pair<std::string, nameSetPtr>  groupPair;
+    typedef std::map<std::string, nameSetPtr>   groupMap;
+    typedef groupMap::const_iterator            gmCitr;
+    typedef std::pair<gmCitr, bool>             gmRes;
+
+    typedef std::pair<SpecProperty, std::string> propNvPair;
+    typedef std::map<SpecProperty, std::string>  propMap;
+    typedef propMap::const_iterator              pmCitr;
+
+    //
+    // aclRule
+    //
+    // A temporary rule created during ACL file processing.
+    //
     class aclRule {
       public:
         enum objectStatus {NONE, VALUE, ALL};
-        AclResult res;
-        nameSet names;
-        bool actionAll; // True if action is set to keyword "all"
-        Action action; // Ignored if action is set to keyword "all"
-        objectStatus objStatus;
-        ObjectType object; // Ignored for all status values except VALUE
-        propMap props;
+
+        AclResult       res;
+        nameSet         names;
+        bool            actionAll; // True if action is set to keyword "all"
+        Action          action; // Ignored if action is set to keyword "all"
+        objectStatus    objStatus;
+        ObjectType      object; // Ignored for all status values except VALUE
+        propMap         props;
       public:
         aclRule(const AclResult r, const std::string n, const groupMap& groups); // action = "all"
         aclRule(const AclResult r, const std::string n, const groupMap& groups, const Action a);
         void setObjectType(const ObjectType o);
         void setObjectTypeAll();
-        bool addProperty(const Property p, const std::string v);
+        bool addProperty(const SpecProperty p, const std::string v);
         bool validate(const AclHelper::objectMapPtr& validationMap);
         std::string toString(); // debug aid
       private:
         void processName(const std::string& name, const groupMap& groups);
     };
-    typedef boost::shared_ptr<aclRule> aclRulePtr;
-    typedef std::vector<aclRulePtr> ruleList;
-    typedef ruleList::const_iterator rlCitr;
+    typedef boost::shared_ptr<aclRule>          aclRulePtr;
+    typedef std::vector<aclRulePtr>             ruleList;
+    typedef ruleList::const_iterator            rlCitr;
 
-    typedef std::vector<std::string> tokList;
-    typedef tokList::const_iterator tlCitr;
+    typedef std::vector<std::string>            tokList;
+    typedef tokList::const_iterator             tlCitr;
 
-    typedef std::set<std::string> keywordSet;
-    typedef keywordSet::const_iterator ksCitr;
+    typedef std::set<std::string>               keywordSet;
+    typedef keywordSet::const_iterator          ksCitr;
     typedef std::pair<std::string, std::string> nvPair; // Name-Value pair
 
-    std::string fileName;
-    int lineNumber;
-    bool contFlag;
-    std::string groupName;
-    nameSet names;
-    groupMap groups;
-    ruleList rules;
+    std::string             fileName;
+    int                     lineNumber;
+    bool                    contFlag;
+    std::string             groupName;
+    nameSet                 names;
+    groupMap                groups;
+    ruleList                rules;
     AclHelper::objectMapPtr validationMap;
-    std::ostringstream errorStream;
+    std::ostringstream      errorStream;
 
   public:
     AclReader();
     virtual ~AclReader();
-    int read(const std::string& fn, boost::shared_ptr<AclData> d);
+    int read(const std::string& fn, boost::shared_ptr<AclData> d); // return=0 for success
     std::string getError();
 
   private:
diff --git a/cpp/src/qpid/acl/AclValidator.cpp b/cpp/src/qpid/acl/AclValidator.cpp
index d5a00b005b..49bb65db4b 100644
--- a/cpp/src/qpid/acl/AclValidator.cpp
+++ b/cpp/src/qpid/acl/AclValidator.cpp
@@ -29,7 +29,7 @@
 namespace qpid {
 namespace acl {
 
-    AclValidator::IntPropertyType::IntPropertyType(int64_t i,int64_t j) : min(i), max(j){    
+    AclValidator::IntPropertyType::IntPropertyType(int64_t i,int64_t j) : min(i), max(j){
     }
 
     bool AclValidator::IntPropertyType::validate(const std::string& val) {
@@ -49,12 +49,12 @@ namespace acl {
     }
 
     std::string AclValidator::IntPropertyType::allowedValues() {
-        return "values should be between " + 
+        return "values should be between " +
             boost::lexical_cast<std::string>(min) + " and " +
             boost::lexical_cast<std::string>(max);
     }
 
-    AclValidator::EnumPropertyType::EnumPropertyType(std::vector<std::string>& allowed): values(allowed){      
+    AclValidator::EnumPropertyType::EnumPropertyType(std::vector<std::string>& allowed): values(allowed){
     }
 
     bool AclValidator::EnumPropertyType::validate(const std::string& val) {
@@ -78,24 +78,27 @@ namespace acl {
     }
 
     AclValidator::AclValidator(){
-        validators.insert(Validator(acl::PROP_MAXQUEUESIZE,
-            boost::shared_ptr<PropertyType>(
-            new IntPropertyType(0,std::numeric_limits<int64_t>::max()))
-            )
-            );
-
-        validators.insert(Validator(acl::PROP_MAXQUEUECOUNT,
-            boost::shared_ptr<PropertyType>(
-            new IntPropertyType(0,std::numeric_limits<int64_t>::max()))
-            )
-            );
+        validators.insert(Validator(acl::SPECPROP_MAXQUEUESIZELOWERLIMIT,
+                          boost::shared_ptr<PropertyType>(
+                            new IntPropertyType(0,std::numeric_limits<int64_t>::max()))));
+
+        validators.insert(Validator(acl::SPECPROP_MAXQUEUESIZEUPPERLIMIT,
+                          boost::shared_ptr<PropertyType>(
+                            new IntPropertyType(0,std::numeric_limits<int64_t>::max()))));
+
+        validators.insert(Validator(acl::SPECPROP_MAXQUEUECOUNTLOWERLIMIT,
+                                    boost::shared_ptr<PropertyType>(
+                                        new IntPropertyType(0,std::numeric_limits<int64_t>::max()))));
+
+        validators.insert(Validator(acl::SPECPROP_MAXQUEUECOUNTUPPERLIMIT,
+                                    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::PROP_POLICYTYPE,
-            boost::shared_ptr<PropertyType>(new EnumPropertyType(v))
-            )
-            );
+        validators.insert(Validator(acl::SPECPROP_POLICYTYPE,
+                          boost::shared_ptr<PropertyType>(
+                            new EnumPropertyType(v))));
 
     }
 
@@ -114,9 +117,9 @@ namespace acl {
                     if (d->actionList[cnt][cnt1]){
 
                         std::for_each(d->actionList[cnt][cnt1]->begin(),
-                            d->actionList[cnt][cnt1]->end(),
-                            boost::bind(&AclValidator::validateRuleSet, this, _1));                    
-                    }//if 
+                                      d->actionList[cnt][cnt1]->end(),
+                                      boost::bind(&AclValidator::validateRuleSet, this, _1));
+                    }//if
                 }//for
             }//if
         }//for
@@ -125,25 +128,29 @@ namespace acl {
     void AclValidator::validateRuleSet(std::pair<const std::string, qpid::acl::AclData::ruleSet>& rules){
         std::for_each(rules.second.begin(),
             rules.second.end(),
-            boost::bind(&AclValidator::validateRule, this, _1)); 
+            boost::bind(&AclValidator::validateRule, this, _1));
     }
 
     void AclValidator::validateRule(qpid::acl::AclData::rule& rule){
         std::for_each(rule.props.begin(),
             rule.props.end(),
-            boost::bind(&AclValidator::validateProperty, this, _1)); 
+            boost::bind(&AclValidator::validateProperty, this, _1));
     }
 
-    void AclValidator::validateProperty(std::pair<const qpid::acl::Property, std::string>& prop){
+    void AclValidator::validateProperty(std::pair<const qpid::acl::SpecProperty, std::string>& prop){
         ValidatorItr itr = validators.find(prop.first);
         if (itr != validators.end()){
-            QPID_LOG(debug,"Found validator for property " << itr->second->allowedValues());
+            QPID_LOG(debug,"ACL: Found validator for property '" << acl::AclHelper::getPropertyStr(itr->first)
+                     << "'. " << itr->second->allowedValues());
 
             if (!itr->second->validate(prop.second)){
-                throw Exception( prop.second + " is not a valid value for '" + 
+                QPID_LOG(debug, "ACL: Property failed validation. '" << prop.second << "' is not a valid value for '"
+                    << AclHelper::getPropertyStr(prop.first) << "'");
+
+                throw Exception( prop.second + " is not a valid value for '" +
                     AclHelper::getPropertyStr(prop.first) + "', " +
                     itr->second->allowedValues());
-            } 
+            }
         }
     }
 
diff --git a/cpp/src/qpid/acl/AclValidator.h b/cpp/src/qpid/acl/AclValidator.h
index 966e5d326b..f85c241b06 100644
--- a/cpp/src/qpid/acl/AclValidator.h
+++ b/cpp/src/qpid/acl/AclValidator.h
@@ -33,18 +33,18 @@ namespace acl {
 class AclValidator {
 
     /* Base Property */
-   class PropertyType{        
-        
+   class PropertyType{
+
         public:
             virtual ~PropertyType(){};
             virtual bool validate(const std::string& val)=0;
             virtual std::string allowedValues()=0;
    };
 
-   class IntPropertyType : public PropertyType{        
+   class IntPropertyType : public PropertyType{
             int64_t min;
             int64_t max;
-        
+
         public:
             IntPropertyType(int64_t min,int64_t max);
             virtual ~IntPropertyType (){};
@@ -53,7 +53,7 @@ class AclValidator {
    };
 
    class EnumPropertyType : public PropertyType{
-            std::vector<std::string> values;               
+            std::vector<std::string> values;
 
         public:
             EnumPropertyType(std::vector<std::string>& allowed);
@@ -61,23 +61,23 @@ class AclValidator {
             virtual bool validate(const std::string& val);
             virtual std::string allowedValues();
    };
-   
-   typedef std::pair<acl::Property,boost::shared_ptr<PropertyType> > Validator;
-   typedef std::map<acl::Property,boost::shared_ptr<PropertyType> > ValidatorMap;
+
+   typedef std::pair<acl::SpecProperty,boost::shared_ptr<PropertyType> > Validator;
+   typedef std::map<acl::SpecProperty,boost::shared_ptr<PropertyType> > ValidatorMap;
    typedef ValidatorMap::iterator ValidatorItr;
- 
+
    ValidatorMap validators;
 
 public:
 
    void validateRuleSet(std::pair<const std::string, qpid::acl::AclData::ruleSet>& rules);
    void validateRule(qpid::acl::AclData::rule& rule);
-   void validateProperty(std::pair<const qpid::acl::Property, std::string>& prop);
-   void validate(boost::shared_ptr<AclData> d);   
+   void validateProperty(std::pair<const qpid::acl::SpecProperty, std::string>& prop);
+   void validate(boost::shared_ptr<AclData> d);
    AclValidator();
    ~AclValidator();
 };
-    
+
 }} // namespace qpid::acl
 
 #endif // QPID_ACL_ACLVALIDATOR_H
diff --git a/cpp/src/qpid/acl/management-schema.xml b/cpp/src/qpid/acl/management-schema.xml
index 7f48a9be34..19fe37333c 100644
--- a/cpp/src/qpid/acl/management-schema.xml
+++ b/cpp/src/qpid/acl/management-schema.xml
@@ -23,8 +23,40 @@
     <property name="transferAcl"   type="bool"    access="RO"    desc="Any transfer ACL rules in force"/>
     <property name="lastAclLoad"   type="absTime" access="RO"    desc="Timestamp of last successful load of ACL"/>
     <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"/>
 
     <method name="reloadACLFile" desc="Reload the ACL file"/>
+
+    <!--
+    Lookup is a general object lookup
+        User Name
+        Action
+        Object
+        Object Name
+        Property Map consisting of <"name" "value"> string pairs.
+    -->
+    <method name="Lookup" desc="Lookup: user action object [objectName [propertyMap]]">
+        <arg name="userId"      dir="I" type="lstr"/>
+        <arg name="action"      dir="I" type="lstr"/>
+        <arg name="object"      dir="I" type="lstr"/>
+        <arg name="objectName"  dir="I" type="lstr"/>
+        <arg name="propertyMap" dir="I" type="map"/>
+        <arg name="result"      dir="O" type="lstr"/>
+    </method>
+
+    <!--
+    LookupPublish is a specific lookup for a PUBLISH EXCHANGE fastpath
+        User Name
+        Exchange Name
+        Routing Key
+    -->
+    <method name="LookupPublish" desc="Lookup PUBLISH EXCHANGE: user exchangeName routingKey">
+        <arg name="userId"       dir="I" type="lstr"/>
+        <arg name="exchangeName" dir="I" type="lstr"/>
+        <arg name="routingKey"   dir="I" type="lstr"/>
+        <arg name="result"       dir="O" type="lstr"/>
+    </method>
+
   </class>
 
   <eventArguments>
@@ -34,10 +66,12 @@
     <arg name="objectType" type="sstr"/>
     <arg name="reason"     type="lstr"/>
     <arg name="userId"     type="sstr"/>
+    <arg name="clientAddr" 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="fileLoaded"     sev="inform" args="userId"/>
   <event name="fileLoadFailed" sev="error"  args="userId, reason"/>