QPID-5519: ACL property/properties for paged queues

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

4 files changed, 61 insertions, 1 deletions

diff --git a/qpid/cpp/src/qpid/acl/AclData.cpp b/qpid/cpp/src/qpid/acl/AclData.cpp
index a205ddabfc..48b5e462e5 100644
--- a/qpid/cpp/src/qpid/acl/AclData.cpp
+++ b/qpid/cpp/src/qpid/acl/AclData.cpp
@@ -160,6 +160,16 @@ bool AclData::lookupMatchRule(
                 //  as rule's index.
                 propertyMapItr lookupParamItr;
                 switch (rulePropMapItr->first) {
+                case acl::SPECPROP_MAXPAGESLOWERLIMIT:
+                case acl::SPECPROP_MAXPAGESUPPERLIMIT:
+                    lookupParamItr = params->find(PROP_MAXPAGES);
+                    break;
+
+                case acl::SPECPROP_MAXPAGEFACTORLOWERLIMIT:
+                case acl::SPECPROP_MAXPAGEFACTORUPPERLIMIT:
+                    lookupParamItr = params->find(PROP_MAXPAGEFACTOR);
+                    break;
+
                 case acl::SPECPROP_MAXQUEUECOUNTUPPERLIMIT:
                 case acl::SPECPROP_MAXQUEUECOUNTLOWERLIMIT:
                     lookupParamItr = params->find(PROP_MAXQUEUECOUNT);
@@ -201,6 +211,8 @@ bool AclData::lookupMatchRule(
                     case acl::SPECPROP_MAXQUEUESIZEUPPERLIMIT:
                     case acl::SPECPROP_MAXFILECOUNTUPPERLIMIT:
                     case acl::SPECPROP_MAXFILESIZEUPPERLIMIT:
+                    case acl::SPECPROP_MAXPAGESUPPERLIMIT:
+                    case acl::SPECPROP_MAXPAGEFACTORUPPERLIMIT:
                         limitChecked &=
                             compareInt(
                                 rulePropMapItr->first,
@@ -213,6 +225,8 @@ bool AclData::lookupMatchRule(
                     case acl::SPECPROP_MAXQUEUESIZELOWERLIMIT:
                     case acl::SPECPROP_MAXFILECOUNTLOWERLIMIT:
                     case acl::SPECPROP_MAXFILESIZELOWERLIMIT:
+                    case acl::SPECPROP_MAXPAGESLOWERLIMIT:
+                    case acl::SPECPROP_MAXPAGEFACTORLOWERLIMIT:
                         limitChecked &=
                             compareInt(
                                 rulePropMapItr->first,
diff --git a/qpid/cpp/src/qpid/acl/AclValidator.cpp b/qpid/cpp/src/qpid/acl/AclValidator.cpp
index a077667a33..89e072000e 100644
--- a/qpid/cpp/src/qpid/acl/AclValidator.cpp
+++ b/qpid/cpp/src/qpid/acl/AclValidator.cpp
@@ -110,6 +110,22 @@ namespace acl {
                                     boost::shared_ptr<PropertyType>(
                                         new IntPropertyType(0,std::numeric_limits<int64_t>::max()))));
 
+        validators.insert(Validator(acl::SPECPROP_MAXPAGESLOWERLIMIT,
+                          boost::shared_ptr<PropertyType>(
+                            new IntPropertyType(0,std::numeric_limits<int64_t>::max()))));
+
+        validators.insert(Validator(acl::SPECPROP_MAXPAGESUPPERLIMIT,
+                          boost::shared_ptr<PropertyType>(
+                            new IntPropertyType(0,std::numeric_limits<int64_t>::max()))));
+
+        validators.insert(Validator(acl::SPECPROP_MAXPAGEFACTORLOWERLIMIT,
+                          boost::shared_ptr<PropertyType>(
+                            new IntPropertyType(0,std::numeric_limits<int64_t>::max()))));
+
+        validators.insert(Validator(acl::SPECPROP_MAXPAGEFACTORUPPERLIMIT,
+                          boost::shared_ptr<PropertyType>(
+                            new IntPropertyType(0,std::numeric_limits<int64_t>::max()))));
+
         std::string policyTypes[] = {"ring", "self-destruct", "reject"};
         std::vector<std::string> v(policyTypes, policyTypes + sizeof(policyTypes) / sizeof(std::string));
         validators.insert(Validator(acl::SPECPROP_POLICYTYPE,
diff --git a/qpid/cpp/src/qpid/broker/AclModule.h b/qpid/cpp/src/qpid/broker/AclModule.h
index c01697ace9..934a11789f 100644
--- a/qpid/cpp/src/qpid/broker/AclModule.h
+++ b/qpid/cpp/src/qpid/broker/AclModule.h
@@ -77,6 +77,9 @@ namespace acl {
         PROP_SCHEMAPACKAGE,
         PROP_SCHEMACLASS,
         PROP_POLICYTYPE,
+        PROP_PAGING,
+        PROP_MAXPAGES,
+        PROP_MAXPAGEFACTOR,
         PROP_MAXQUEUESIZE,
         PROP_MAXQUEUECOUNT,
         PROP_MAXFILESIZE,
@@ -100,6 +103,7 @@ namespace acl {
         SPECPROP_SCHEMAPACKAGE   = PROP_SCHEMAPACKAGE,
         SPECPROP_SCHEMACLASS     = PROP_SCHEMACLASS,
         SPECPROP_POLICYTYPE      = PROP_POLICYTYPE,
+        SPECPROP_PAGING          = PROP_PAGING,
 
         SPECPROP_MAXQUEUESIZELOWERLIMIT,
         SPECPROP_MAXQUEUESIZEUPPERLIMIT,
@@ -108,7 +112,11 @@ namespace acl {
         SPECPROP_MAXFILESIZELOWERLIMIT,
         SPECPROP_MAXFILESIZEUPPERLIMIT,
         SPECPROP_MAXFILECOUNTLOWERLIMIT,
-        SPECPROP_MAXFILECOUNTUPPERLIMIT };
+        SPECPROP_MAXFILECOUNTUPPERLIMIT,
+        SPECPROP_MAXPAGESLOWERLIMIT,
+        SPECPROP_MAXPAGESUPPERLIMIT,
+        SPECPROP_MAXPAGEFACTORLOWERLIMIT, 
+        SPECPROP_MAXPAGEFACTORUPPERLIMIT };
 
 // AclResult  shared between ACL spec and ACL authorise interface
     enum AclResult {
@@ -229,6 +237,9 @@ namespace acl {
             if (str.compare("schemapackage") == 0) return PROP_SCHEMAPACKAGE;
             if (str.compare("schemaclass")   == 0) return PROP_SCHEMACLASS;
             if (str.compare("policytype")    == 0) return PROP_POLICYTYPE;
+            if (str.compare("paging")        == 0) return PROP_PAGING;
+            if (str.compare("maxpages")      == 0) return PROP_MAXPAGES;
+            if (str.compare("maxpagefactor") == 0) return PROP_MAXPAGEFACTOR;
             if (str.compare("maxqueuesize")  == 0) return PROP_MAXQUEUESIZE;
             if (str.compare("maxqueuecount") == 0) return PROP_MAXQUEUECOUNT;
             if (str.compare("maxfilesize")   == 0) return PROP_MAXFILESIZE;
@@ -249,6 +260,9 @@ namespace acl {
             case PROP_SCHEMAPACKAGE: return "schemapackage";
             case PROP_SCHEMACLASS:   return "schemaclass";
             case PROP_POLICYTYPE:    return "policytype";
+            case PROP_PAGING:        return "paging";
+            case PROP_MAXPAGES:      return "maxpages";
+            case PROP_MAXPAGEFACTOR: return "maxpagefactor";
             case PROP_MAXQUEUESIZE:  return "maxqueuesize";
             case PROP_MAXQUEUECOUNT: return "maxqueuecount";
             case PROP_MAXFILESIZE:   return "maxfilesize";
@@ -270,6 +284,7 @@ namespace acl {
             if (str.compare("schemapackage") == 0) return SPECPROP_SCHEMAPACKAGE;
             if (str.compare("schemaclass")   == 0) return SPECPROP_SCHEMACLASS;
             if (str.compare("policytype")    == 0) return SPECPROP_POLICYTYPE;
+            if (str.compare("paging")        == 0) return SPECPROP_PAGING;
             if (str.compare("queuemaxsizelowerlimit")   == 0) return SPECPROP_MAXQUEUESIZELOWERLIMIT;
             if (str.compare("queuemaxsizeupperlimit")   == 0) return SPECPROP_MAXQUEUESIZEUPPERLIMIT;
             if (str.compare("queuemaxcountlowerlimit")  == 0) return SPECPROP_MAXQUEUECOUNTLOWERLIMIT;
@@ -278,6 +293,10 @@ namespace acl {
             if (str.compare("filemaxsizeupperlimit")    == 0) return SPECPROP_MAXFILESIZEUPPERLIMIT;
             if (str.compare("filemaxcountlowerlimit")   == 0) return SPECPROP_MAXFILECOUNTLOWERLIMIT;
             if (str.compare("filemaxcountupperlimit")   == 0) return SPECPROP_MAXFILECOUNTUPPERLIMIT;
+            if (str.compare("pageslowerlimit")          == 0) return SPECPROP_MAXPAGESLOWERLIMIT;
+            if (str.compare("pagesupperlimit")          == 0) return SPECPROP_MAXPAGESUPPERLIMIT;
+            if (str.compare("pagefactorlowerlimit")     == 0) return SPECPROP_MAXPAGEFACTORLOWERLIMIT;
+            if (str.compare("pagefactorupperlimit")     == 0) return SPECPROP_MAXPAGEFACTORUPPERLIMIT;
             // Allow old names in ACL file as aliases for newly-named properties
             if (str.compare("maxqueuesize")             == 0) return SPECPROP_MAXQUEUESIZEUPPERLIMIT;
             if (str.compare("maxqueuecount")            == 0) return SPECPROP_MAXQUEUECOUNTUPPERLIMIT;
@@ -297,6 +316,7 @@ namespace acl {
                 case SPECPROP_SCHEMAPACKAGE: return "schemapackage";
                 case SPECPROP_SCHEMACLASS:   return "schemaclass";
                 case SPECPROP_POLICYTYPE:    return "policytype";
+                case SPECPROP_PAGING:        return "paging";
                 case SPECPROP_MAXQUEUESIZELOWERLIMIT:  return "queuemaxsizelowerlimit";
                 case SPECPROP_MAXQUEUESIZEUPPERLIMIT:  return "queuemaxsizeupperlimit";
                 case SPECPROP_MAXQUEUECOUNTLOWERLIMIT: return "queuemaxcountlowerlimit";
@@ -305,6 +325,10 @@ namespace acl {
                 case SPECPROP_MAXFILESIZEUPPERLIMIT:   return "filemaxsizeupperlimit";
                 case SPECPROP_MAXFILECOUNTLOWERLIMIT:  return "filemaxcountlowerlimit";
                 case SPECPROP_MAXFILECOUNTUPPERLIMIT:  return "filemaxcountupperlimit";
+                case SPECPROP_MAXPAGESLOWERLIMIT:      return "pageslowerlimit";
+                case SPECPROP_MAXPAGESUPPERLIMIT:      return "pageslowerlimit";
+                case SPECPROP_MAXPAGEFACTORLOWERLIMIT: return "pagefactorlowerlimit";
+                case SPECPROP_MAXPAGEFACTORUPPERLIMIT: return "pagefactorlowerlimit";
                 default: assert(false); // should never get here
             }
             return "";
@@ -381,6 +405,9 @@ namespace acl {
             p4->insert(PROP_EXCLUSIVE);
             p4->insert(PROP_AUTODELETE);
             p4->insert(PROP_POLICYTYPE);
+            p4->insert(PROP_PAGING);
+            p4->insert(PROP_MAXPAGES);
+            p4->insert(PROP_MAXPAGEFACTOR);
             p4->insert(PROP_MAXQUEUESIZE);
             p4->insert(PROP_MAXQUEUECOUNT);
 
diff --git a/qpid/cpp/src/qpid/broker/Broker.cpp b/qpid/cpp/src/qpid/broker/Broker.cpp
index 73910ec6c4..4017fdbfe3 100644
--- a/qpid/cpp/src/qpid/broker/Broker.cpp
+++ b/qpid/cpp/src/qpid/broker/Broker.cpp
@@ -1301,6 +1301,9 @@ std::pair<boost::shared_ptr<Queue>, bool> Broker::createQueue(
         params.insert(make_pair(acl::PROP_EXCLUSIVE, owner ? _TRUE : _FALSE));
         params.insert(make_pair(acl::PROP_AUTODELETE, settings.autodelete ? _TRUE : _FALSE));
         params.insert(make_pair(acl::PROP_POLICYTYPE, settings.getLimitPolicy()));
+        params.insert(make_pair(acl::PROP_PAGING, settings.paging ? _TRUE : _FALSE));
+        params.insert(make_pair(acl::PROP_MAXPAGES, boost::lexical_cast<string>(settings.maxPages)));
+        params.insert(make_pair(acl::PROP_MAXPAGEFACTOR, boost::lexical_cast<string>(settings.pageFactor)));
         params.insert(make_pair(acl::PROP_MAXQUEUECOUNT, boost::lexical_cast<string>(settings.maxDepth.getCount())));
         params.insert(make_pair(acl::PROP_MAXQUEUESIZE, boost::lexical_cast<string>(settings.maxDepth.getSize())));
         params.insert(make_pair(acl::PROP_MAXFILECOUNT, boost::lexical_cast<string>(settings.maxFileCount)));