QPID-6391: add support for no-local filter

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

3 files changed, 20 insertions, 1 deletions

diff --git a/qpid/cpp/src/qpid/amqp/descriptors.h b/qpid/cpp/src/qpid/amqp/descriptors.h
index 29c626edc2..276514e2e5 100644
--- a/qpid/cpp/src/qpid/amqp/descriptors.h
+++ b/qpid/cpp/src/qpid/amqp/descriptors.h
@@ -85,12 +85,14 @@ namespace filters {
 const std::string LEGACY_DIRECT_FILTER_SYMBOL("apache.org:legacy-amqp-direct-binding:string");
 const std::string LEGACY_TOPIC_FILTER_SYMBOL("apache.org:legacy-amqp-topic-binding:string");
 const std::string LEGACY_HEADERS_FILTER_SYMBOL("apache.org:legacy-amqp-headers-binding:map");
+const std::string NO_LOCAL_FILTER_SYMBOL("apache.org:no-local-filter:list");
 const std::string SELECTOR_FILTER_SYMBOL("apache.org:selector-filter:string");
 const std::string XQUERY_FILTER_SYMBOL("apache.org:xquery-filter:string");
 
 const uint64_t LEGACY_DIRECT_FILTER_CODE(0x0000468C00000000ULL);
 const uint64_t LEGACY_TOPIC_FILTER_CODE(0x0000468C00000001ULL);
 const uint64_t LEGACY_HEADERS_FILTER_CODE(0x0000468C00000002ULL);
+const uint64_t NO_LOCAL_FILTER_CODE(0x0000468C00000003ULL);
 const uint64_t SELECTOR_FILTER_CODE(0x0000468C00000004ULL);
 const uint64_t XQUERY_FILTER_CODE(0x0000468C00000005ULL);
 }
diff --git a/qpid/cpp/src/qpid/broker/amqp/Filter.cpp b/qpid/cpp/src/qpid/broker/amqp/Filter.cpp
index b7b29e004d..ab456f4355 100644
--- a/qpid/cpp/src/qpid/broker/amqp/Filter.cpp
+++ b/qpid/cpp/src/qpid/broker/amqp/Filter.cpp
@@ -46,7 +46,7 @@ const std::string DEFAULT_XQUERY_FILTER("default-xquery-filter");
 const std::string DEFAULT_XQUERY_VALUE("true()");
 const std::string WILDCARD("#");
 }
-Filter::Filter() : inHeadersMap(false) {}
+Filter::Filter() : inHeadersMap(false), nolocal(false) {}
 
 void Filter::read(pn_data_t* data)
 {
@@ -88,6 +88,9 @@ void Filter::onStringValue(const qpid::amqp::CharSequence& key, const qpid::amqp
                 setFilter(selectorFilter, filter);
             } else if (descriptor->match(qpid::amqp::filters::XQUERY_FILTER_SYMBOL, qpid::amqp::filters::XQUERY_FILTER_CODE)) {
                 setFilter(xqueryFilter, filter);
+            } else if (descriptor && descriptor->match(qpid::amqp::filters::NO_LOCAL_FILTER_SYMBOL, qpid::amqp::filters::NO_LOCAL_FILTER_CODE)) {
+                //though the no-local-filter is define to be a list, the JMS client sends it as a string
+                nolocal = true;
             } else {
                 QPID_LOG(notice, "Skipping unrecognised string filter with key " << filter.key << " and descriptor " << filter.descriptor);
             }
@@ -155,6 +158,14 @@ void Filter::onDoubleValue(const qpid::amqp::CharSequence& key, double value, co
     headersFilter.value[std::string(key.data, key.size)] = value;
 }
 
+bool Filter::onStartListValue(const qpid::amqp::CharSequence& /*key*/, uint32_t /*count*/, const qpid::amqp::Descriptor* descriptor)
+{
+    if (descriptor && descriptor->match(qpid::amqp::filters::NO_LOCAL_FILTER_SYMBOL, qpid::amqp::filters::NO_LOCAL_FILTER_CODE)) {
+        nolocal = true;
+    }
+    return false;
+}
+
 bool Filter::onStartMapValue(const qpid::amqp::CharSequence& key, uint32_t /*count*/, const qpid::amqp::Descriptor* descriptor)
 {
     if (inHeadersMap) {
@@ -234,6 +245,10 @@ void Filter::configure(QueueSettings& settings)
         settings.filter = getSelectorFilter();
         active.push_back(&selectorFilter);
     }
+    if (nolocal) {
+        settings.noLocal = true;
+        QPID_LOG(notice, "No local filter set");
+    }
 }
 
 std::string Filter::getBindingKey(boost::shared_ptr<Exchange> exchange) const
diff --git a/qpid/cpp/src/qpid/broker/amqp/Filter.h b/qpid/cpp/src/qpid/broker/amqp/Filter.h
index c246fb9ede..af942535ef 100644
--- a/qpid/cpp/src/qpid/broker/amqp/Filter.h
+++ b/qpid/cpp/src/qpid/broker/amqp/Filter.h
@@ -101,6 +101,7 @@ class Filter : private qpid::amqp::MapReader
     void onLongValue(const qpid::amqp::CharSequence&, int64_t, const qpid::amqp::Descriptor*);
     void onFloatValue(const qpid::amqp::CharSequence&, float, const qpid::amqp::Descriptor*);
     void onDoubleValue(const qpid::amqp::CharSequence&, double, const qpid::amqp::Descriptor*);
+    bool onStartListValue(const qpid::amqp::CharSequence& key, uint32_t count, const qpid::amqp::Descriptor* descriptor);
     bool onStartMapValue(const qpid::amqp::CharSequence& key, uint32_t count, const qpid::amqp::Descriptor* descriptor);
     void onEndMapValue(const qpid::amqp::CharSequence& key, uint32_t count, const qpid::amqp::Descriptor* descriptor);
     void setFilter(StringFilter&, const StringFilter&);
@@ -120,6 +121,7 @@ class Filter : private qpid::amqp::MapReader
     MapFilter headersFilter;
     std::vector<FilterBase*> active;
     bool inHeadersMap;
+    bool nolocal;
 };
 }}} // namespace qpid::broker::amqp