You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by as...@apache.org on 2013/03/04 22:08:39 UTC
svn commit: r1452524 - in /qpid/trunk/qpid/cpp/src/qpid: amqp/descriptors.h
broker/amqp/Filter.cpp broker/amqp/Filter.h broker/amqp/Outgoing.cpp
broker/amqp/Outgoing.h broker/amqp/Session.cpp
messaging/amqp/ReceiverContext.cpp
Author: astitcher
Date: Mon Mar 4 21:08:36 2013
New Revision: 1452524
URL: http://svn.apache.org/r1452524
Log:
QPID-4558: Selectors for C++ broker
- Added in amqp 1.0 support that uses a filter for the selector
- This change requires at least qpid-proton 0.4 (or a lot of warning
messages are produced by the broker)
Modified:
qpid/trunk/qpid/cpp/src/qpid/amqp/descriptors.h
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Filter.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Filter.h
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Outgoing.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Outgoing.h
qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.cpp
qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.cpp
Modified: qpid/trunk/qpid/cpp/src/qpid/amqp/descriptors.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/amqp/descriptors.h?rev=1452524&r1=1452523&r2=1452524&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/amqp/descriptors.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/amqp/descriptors.h Mon Mar 4 21:08:36 2013
@@ -77,7 +77,8 @@ const Descriptor SASL_OUTCOME(SASL_OUTCO
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-direct-binding:string");
+const std::string LEGACY_TOPIC_FILTER_SYMBOL("apache.org:legacy-amqp-topic-binding:string");
+const std::string QPID_SELECTOR_FILTER_SYMBOL("qpid.apache.org:selector-filter:string");
const uint64_t LEGACY_DIRECT_FILTER_CODE(0x0000468C00000000ULL);
const uint64_t LEGACY_TOPIC_FILTER_CODE(0x0000468C00000001ULL);
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Filter.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Filter.cpp?rev=1452524&r1=1452523&r2=1452524&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Filter.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Filter.cpp Mon Mar 4 21:08:36 2013
@@ -61,6 +61,8 @@ void Filter::onStringValue(const qpid::a
if (descriptor->match(qpid::amqp::filters::LEGACY_TOPIC_FILTER_SYMBOL, qpid::amqp::filters::LEGACY_TOPIC_FILTER_CODE)
|| descriptor->match(qpid::amqp::filters::LEGACY_DIRECT_FILTER_SYMBOL, qpid::amqp::filters::LEGACY_DIRECT_FILTER_CODE)) {
setSubjectFilter(filter);
+ } else if (descriptor->match(qpid::amqp::filters::QPID_SELECTOR_FILTER_SYMBOL, 0)) {
+ setSelectorFilter(filter);
} else {
QPID_LOG(notice, "Skipping unrecognised string filter with key " << filter.key << " and descriptor " << filter.descriptor);
}
@@ -89,6 +91,26 @@ void Filter::setSubjectFilter(const Stri
}
}
+bool Filter::hasSelectorFilter() const
+{
+ return !selectorFilter.value.empty();
+}
+
+std::string Filter::getSelectorFilter() const
+{
+ return selectorFilter.value;
+}
+
+
+void Filter::setSelectorFilter(const StringFilter& filter)
+{
+ if (hasSelectorFilter()) {
+ QPID_LOG(notice, "Skipping filter with key " << filter.key << "; selector filter already set");
+ } else {
+ selectorFilter = filter;
+ }
+}
+
void Filter::bind(boost::shared_ptr<Exchange> exchange, boost::shared_ptr<Queue> queue)
{
subjectFilter.bind(exchange, queue);
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Filter.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Filter.h?rev=1452524&r1=1452523&r2=1452524&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Filter.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Filter.h Mon Mar 4 21:08:36 2013
@@ -40,6 +40,8 @@ class Filter : qpid::amqp::MapReader
void write(pn_data_t*);
bool hasSubjectFilter() const;
std::string getSubjectFilter() const;
+ bool hasSelectorFilter() const;
+ std::string getSelectorFilter() const;
void bind(boost::shared_ptr<Exchange> exchange, boost::shared_ptr<Queue> queue);
private:
struct StringFilter
@@ -55,8 +57,10 @@ class Filter : qpid::amqp::MapReader
void onStringValue(const qpid::amqp::CharSequence& key, const qpid::amqp::CharSequence& value, const qpid::amqp::Descriptor* descriptor);
void setSubjectFilter(const StringFilter&);
+ void setSelectorFilter(const StringFilter&);
StringFilter subjectFilter;
+ StringFilter selectorFilter;
};
}}} // namespace qpid::broker::amqp
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Outgoing.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Outgoing.cpp?rev=1452524&r1=1452523&r2=1452524&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Outgoing.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Outgoing.cpp Mon Mar 4 21:08:36 2013
@@ -22,6 +22,7 @@
#include "qpid/broker/amqp/Header.h"
#include "qpid/broker/amqp/Translation.h"
#include "qpid/broker/Queue.h"
+#include "qpid/broker/Selector.h"
#include "qpid/broker/TopicKeyNode.h"
#include "qpid/sys/OutputControl.h"
#include "qpid/amqp/MessageEncoder.h"
@@ -170,6 +171,11 @@ void Outgoing::setSubjectFilter(const st
subjectFilter = f;
}
+void Outgoing::setSelectorFilter(const std::string& f)
+{
+ selector.reset(new Selector(f));
+}
+
namespace {
bool match(TokenIterator& filter, TokenIterator& target)
@@ -213,7 +219,8 @@ bool match(const std::string& filter, co
bool Outgoing::filter(const qpid::broker::Message& m)
{
- return subjectFilter.empty() || subjectFilter == m.getRoutingKey() || match(subjectFilter, m.getRoutingKey());
+ return (subjectFilter.empty() || subjectFilter == m.getRoutingKey() || match(subjectFilter, m.getRoutingKey()))
+ && (!selector || selector->filter(m));
}
void Outgoing::cancel() {}
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Outgoing.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Outgoing.h?rev=1452524&r1=1452523&r2=1452524&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Outgoing.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Outgoing.h Mon Mar 4 21:08:36 2013
@@ -24,7 +24,9 @@
#include "qpid/broker/amqp/Message.h"
#include "qpid/broker/amqp/ManagedOutgoingLink.h"
#include "qpid/broker/Consumer.h"
+
#include <boost/shared_ptr.hpp>
+#include <boost/scoped_ptr.hpp>
#include <boost/enable_shared_from_this.hpp>
extern "C" {
#include <proton/engine.h>
@@ -37,6 +39,7 @@ class OutputControl;
namespace broker {
class Broker;
class Queue;
+class Selector;
namespace amqp {
class ManagedSession;
template <class T>
@@ -61,6 +64,7 @@ class Outgoing : public qpid::broker::Co
public:
Outgoing(Broker&,boost::shared_ptr<Queue> q, pn_link_t* l, ManagedSession&, qpid::sys::OutputControl& o, bool topic);
void setSubjectFilter(const std::string&);
+ void setSelectorFilter(const std::string&);
void init();
bool dispatch();
void write(const char* data, size_t size);
@@ -102,6 +106,7 @@ class Outgoing : public qpid::broker::Co
int outstanding;
std::vector<char> buffer;
std::string subjectFilter;
+ boost::scoped_ptr<Selector> selector;
};
}}} // namespace qpid::broker::amqp
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.cpp?rev=1452524&r1=1452523&r2=1452524&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp/Session.cpp Mon Mar 4 21:08:36 2013
@@ -31,6 +31,7 @@
#include "qpid/broker/FanOutExchange.h"
#include "qpid/broker/Message.h"
#include "qpid/broker/Queue.h"
+#include "qpid/broker/Selector.h"
#include "qpid/broker/TopicExchange.h"
#include "qpid/broker/amqp/Filter.h"
#include "qpid/broker/amqp/NodeProperties.h"
@@ -132,6 +133,9 @@ void Session::attach(pn_link_t* link)
if (filter.hasSubjectFilter()) {
q->setSubjectFilter(filter.getSubjectFilter());
}
+ if (filter.hasSelectorFilter()) {
+ q->setSelectorFilter(filter.getSelectorFilter());
+ }
senders[link] = q;
} else if (node.exchange) {
QueueSettings settings(false, true);
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.cpp?rev=1452524&r1=1452523&r2=1452524&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.cpp Mon Mar 4 21:08:36 2013
@@ -120,6 +120,20 @@ void ReceiverContext::configure(pn_termi
helper.setNodeProperties(source);
}
+ // Look specifically for qpid.selector link property and add a filter for it
+ qpid::types::Variant::Map::const_iterator i = helper.getLinkProperties().find("qpid.selector");
+ if (i!=helper.getLinkProperties().end()) {
+ pn_data_t* filter = pn_terminus_filter(source);
+ pn_data_put_map(filter);
+ pn_data_enter(filter);
+ pn_data_put_symbol(filter, convert("qpid.selector"));
+ pn_data_put_described(filter);
+ pn_data_enter(filter);
+ pn_data_put_symbol(filter, convert(qpid::amqp::filters::QPID_SELECTOR_FILTER_SYMBOL));
+ pn_data_put_string(filter, convert(i->second));
+ pn_data_exit(filter);
+ pn_data_exit(filter);
+ }
if (!address.getSubject().empty()) {
//filter:
pn_data_t* filter = pn_terminus_filter(source);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org