You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by tr...@apache.org on 2010/02/01 17:10:34 UTC
svn commit: r905322 - in /qpid/trunk/qpid/cpp/src: qpid/broker/ qpid/sys/
tests/
Author: tross
Date: Mon Feb 1 16:10:33 2010
New Revision: 905322
URL: http://svn.apache.org/viewvc?rev=905322&view=rev
Log:
QPID-2348 - [C++] The HeadersExchange does not support federation
Applied patch from Sam Joyce
Added:
qpid/trunk/qpid/cpp/src/tests/headers_federation.py
qpid/trunk/qpid/cpp/src/tests/run_headers_federation_tests
Modified:
qpid/trunk/qpid/cpp/src/qpid/broker/Bridge.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/Bridge.h
qpid/trunk/qpid/cpp/src/qpid/broker/Exchange.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/Exchange.h
qpid/trunk/qpid/cpp/src/qpid/broker/HeadersExchange.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/HeadersExchange.h
qpid/trunk/qpid/cpp/src/qpid/sys/CopyOnWriteArray.h
qpid/trunk/qpid/cpp/src/tests/federation.py
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Bridge.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Bridge.cpp?rev=905322&r1=905321&r2=905322&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Bridge.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Bridge.cpp Mon Feb 1 16:10:33 2010
@@ -273,13 +273,19 @@
}
void Bridge::propagateBinding(const string& key, const string& tagList,
- const string& op, const string& origin)
+ const string& op, const string& origin,
+ qpid::framing::FieldTable* extra_args)
{
const string& localTag = link->getBroker()->getFederationTag();
const string& peerTag = connState->getFederationPeerTag();
if (tagList.find(peerTag) == tagList.npos) {
FieldTable bindArgs;
+ if (extra_args) {
+ for (qpid::framing::FieldTable::ValueMap::iterator i=extra_args->begin(); i != extra_args->end(); ++i) {
+ bindArgs.insert((*i));
+ }
+ }
string newTagList(tagList + string(tagList.empty() ? "" : ",") + localTag);
bindArgs.setString(qpidFedOp, op);
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Bridge.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Bridge.h?rev=905322&r1=905321&r2=905322&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Bridge.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Bridge.h Mon Feb 1 16:10:33 2010
@@ -73,7 +73,7 @@
static Bridge::shared_ptr decode(LinkRegistry& links, framing::Buffer& buffer);
// Exchange::DynamicBridge methods
- void propagateBinding(const std::string& key, const std::string& tagList, const std::string& op, const std::string& origin);
+ void propagateBinding(const std::string& key, const std::string& tagList, const std::string& op, const std::string& origin, qpid::framing::FieldTable* extra_args=0);
void sendReorigin();
void ioThreadPropagateBinding(const string& queue, const string& exchange, const string& key, framing::FieldTable args);
bool containsLocalTag(const std::string& tagList) const;
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Exchange.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Exchange.cpp?rev=905322&r1=905321&r2=905322&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Exchange.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Exchange.cpp Mon Feb 1 16:10:33 2010
@@ -296,14 +296,14 @@
{
}
-void Exchange::propagateFedOp(const string& routingKey, const string& tags, const string& op, const string& origin)
+void Exchange::propagateFedOp(const string& routingKey, const string& tags, const string& op, const string& origin, qpid::framing::FieldTable* extra_args)
{
Mutex::ScopedLock l(bridgeLock);
string myOp(op.empty() ? fedOpBind : op);
for (std::vector<DynamicBridge*>::iterator iter = bridgeVector.begin();
iter != bridgeVector.end(); iter++)
- (*iter)->propagateBinding(routingKey, tags, op, origin);
+ (*iter)->propagateBinding(routingKey, tags, op, origin, extra_args);
}
Exchange::Binding::Binding(const string& _key, Queue::shared_ptr _queue, Exchange* parent,
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Exchange.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Exchange.h?rev=905322&r1=905321&r2=905322&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Exchange.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Exchange.h Mon Feb 1 16:10:33 2010
@@ -163,7 +163,7 @@
class DynamicBridge {
public:
virtual ~DynamicBridge() {}
- virtual void propagateBinding(const std::string& key, const std::string& tagList, const std::string& op, const std::string& origin) = 0;
+ virtual void propagateBinding(const std::string& key, const std::string& tagList, const std::string& op, const std::string& origin, qpid::framing::FieldTable* extra_args=0) = 0;
virtual void sendReorigin() = 0;
virtual bool containsLocalTag(const std::string& tagList) const = 0;
virtual const std::string& getLocalTag() const = 0;
@@ -185,7 +185,8 @@
QPID_BROKER_EXTERN virtual void handleHelloRequest();
void propagateFedOp(const std::string& routingKey, const std::string& tags,
- const std::string& op, const std::string& origin);
+ const std::string& op, const std::string& origin,
+ qpid::framing::FieldTable* extra_args=0);
};
}}
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/HeadersExchange.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/HeadersExchange.cpp?rev=905322&r1=905321&r2=905322&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/HeadersExchange.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/HeadersExchange.cpp Mon Feb 1 16:10:33 2010
@@ -37,10 +37,21 @@
using namespace qpid::broker;
namespace {
+ const std::string x_match("x-match");
+ // possible values for x-match
const std::string all("all");
const std::string any("any");
- const std::string x_match("x-match");
const std::string empty;
+
+ // federation related args and values
+ const std::string qpidFedOp("qpid.fed.op");
+ const std::string qpidFedTags("qpid.fed.tags");
+ const std::string qpidFedOrigin("qpid.fed.origin");
+
+ const std::string fedOpBind("B");
+ const std::string fedOpUnbind("U");
+ const std::string fedOpReorigin("R");
+ const std::string fedOpHello("H");
}
HeadersExchange::HeadersExchange(const string& _name, Manageable* _parent, Broker* b) :
@@ -68,37 +79,106 @@
return empty;
}
if (!what->convertsTo<std::string>()) {
- throw InternalErrorException(QPID_MSG("Invalid x-match value binding to headers exchange."));
+ throw InternalErrorException(QPID_MSG("Invalid x-match binding format to headers exchange. Must be a string [\"all\" or \"any\"]"));
}
return what->get<std::string>();
}
-bool HeadersExchange::bind(Queue::shared_ptr queue, const string& bindingKey, const FieldTable* args){
- std::string what = getMatch(args);
- if (what != all && what != any)
- throw InternalErrorException(QPID_MSG("Invalid x-match value binding to headers exchange."));
+bool HeadersExchange::bind(Queue::shared_ptr queue, const string& bindingKey, const FieldTable* args)
+{
+ string fedOp(fedOpBind);
+ string fedTags;
+ string fedOrigin;
+ if (args) {
+ fedOp = args->getAsString(qpidFedOp);
+ fedTags = args->getAsString(qpidFedTags);
+ fedOrigin = args->getAsString(qpidFedOrigin);
+ }
+ bool propagate = false;
+
+ // The federation args get propagated directly, so we need to identify
+ // the non feteration args in case a federated propagate is needed
+ FieldTable extra_args;
+ getNonFedArgs(args, extra_args);
+
+ if (fedOp.empty() || fedOp == fedOpBind) {
+ // x-match arg MUST be present for a bind call
+ std::string x_match_value = getMatch(args);
- Binding::shared_ptr binding (new Binding (bindingKey, queue, this, *args));
- if (bindings.add_unless(binding, MatchArgs(queue, args))) {
- if (mgmtExchange != 0) {
- mgmtExchange->inc_bindingCount();
+ if (x_match_value != all && x_match_value != any) {
+ throw InternalErrorException(QPID_MSG("Invalid or missing x-match value binding to headers exchange. Must be a string [\"all\" or \"any\"]"));
+ }
+
+ {
+ Mutex::ScopedLock l(lock);
+ Binding::shared_ptr binding (new Binding (bindingKey, queue, this, *args));
+ BoundKey bk(binding);
+ if (bindings.add_unless(bk, MatchArgs(queue, args))) {
+ propagate = bk.fedBinding.addOrigin(fedOrigin);
+ if (mgmtExchange != 0) {
+ mgmtExchange->inc_bindingCount();
+ }
+ } else {
+ return false;
+ }
+ } // lock dropped
+
+ } else if (fedOp == fedOpUnbind) {
+ Mutex::ScopedLock l(lock);
+
+ FedUnbindModifier modifier(fedOrigin);
+ bindings.modify_if(MatchKey(queue, bindingKey), modifier);
+ propagate = modifier.shouldPropagate;
+ if (modifier.shouldUnbind) {
+ unbind(queue, bindingKey, args);
+ }
+
+ } else if (fedOp == fedOpReorigin) {
+ Bindings::ConstPtr p = bindings.snapshot();
+ if (p.get())
+ {
+ Mutex::ScopedLock l(lock);
+ for (std::vector<BoundKey>::const_iterator i = p->begin(); i != p->end(); ++i)
+ {
+ if ((*i).fedBinding.hasLocal()) {
+ propagateFedOp( (*i).binding->key, string(), fedOpBind, string());
+ }
+ }
}
- routeIVE();
- return true;
- } else {
- return false;
}
+ routeIVE();
+ if (propagate) {
+ FieldTable * prop_args = (extra_args.count() != 0 ? &extra_args : 0);
+ propagateFedOp(bindingKey, fedTags, fedOp, fedOrigin, prop_args);
+ }
+
+ return true;
}
bool HeadersExchange::unbind(Queue::shared_ptr queue, const string& bindingKey, const FieldTable*){
- if (bindings.remove_if(MatchKey(queue, bindingKey))) {
- if (mgmtExchange != 0) {
- mgmtExchange->dec_bindingCount();
+ bool propagate = false;
+ {
+ Mutex::ScopedLock l(lock);
+
+ FedUnbindModifier modifier;
+ MatchKey match_key(queue, bindingKey);
+ bindings.modify_if(match_key, modifier);
+ propagate = modifier.shouldPropagate;
+ if (modifier.shouldUnbind) {
+ if (bindings.remove_if(match_key)) {
+ if (mgmtExchange != 0) {
+ mgmtExchange->dec_bindingCount();
+ }
+ } else {
+ return false;
+ }
}
- return true;
- } else {
- return false;
}
+
+ if (propagate) {
+ propagateFedOp(bindingKey, string(), fedOpUnbind, string());
+ }
+ return true;
}
@@ -117,13 +197,12 @@
PreRoute pr(msg, this);
- ConstBindingList p = bindings.snapshot();
BindingList b(new std::vector<boost::shared_ptr<qpid::broker::Exchange::Binding> >);
- if (p.get())
- {
- for (std::vector<Binding::shared_ptr>::const_iterator i = p->begin(); i != p->end(); ++i) {
- if (match((*i)->args, *args)) {
- b->push_back(*i);
+ Bindings::ConstPtr p = bindings.snapshot();
+ if (p.get()) {
+ for (std::vector<BoundKey>::const_iterator i = p->begin(); i != p->end(); ++i) {
+ if (match((*i).binding->args, *args)) {
+ b->push_back((*i).binding);
}
}
}
@@ -135,8 +214,8 @@
{
Bindings::ConstPtr p = bindings.snapshot();
if (p.get()){
- for (std::vector<Binding::shared_ptr>::const_iterator i = p->begin(); i != p->end(); ++i) {
- if ( (!args || equal((*i)->args, *args)) && (!queue || (*i)->queue == queue)) {
+ for (std::vector<BoundKey>::const_iterator i = p->begin(); i != p->end(); ++i) {
+ if ( (!args || equal((*i).binding->args, *args)) && (!queue || (*i).binding->queue == queue)) {
return true;
}
}
@@ -144,6 +223,26 @@
return false;
}
+void HeadersExchange::getNonFedArgs(const FieldTable* args, FieldTable& nonFedArgs)
+{
+ if (!args)
+ {
+ return;
+ }
+
+ for (qpid::framing::FieldTable::ValueMap::const_iterator i=args->begin(); i != args->end(); ++i)
+ {
+ const string & name(i->first);
+ if (name == qpidFedOp ||
+ name == qpidFedTags ||
+ name == qpidFedOrigin)
+ {
+ continue;
+ }
+ nonFedArgs.insert((*i));
+ }
+}
+
HeadersExchange::~HeadersExchange() {}
const std::string HeadersExchange::typeName("headers");
@@ -206,15 +305,37 @@
return true;
}
+//---------
HeadersExchange::MatchArgs::MatchArgs(Queue::shared_ptr q, const qpid::framing::FieldTable* a) : queue(q), args(a) {}
-bool HeadersExchange::MatchArgs::operator()(Exchange::Binding::shared_ptr b)
+
+bool HeadersExchange::MatchArgs::operator()(BoundKey & bk)
{
- return b->queue == queue && b->args == *args;
+ return bk.binding->queue == queue && bk.binding->args == *args;
}
+//---------
HeadersExchange::MatchKey::MatchKey(Queue::shared_ptr q, const std::string& k) : queue(q), key(k) {}
-bool HeadersExchange::MatchKey::operator()(Exchange::Binding::shared_ptr b)
+bool HeadersExchange::MatchKey::operator()(BoundKey & bk)
+{
+ return bk.binding->queue == queue && bk.binding->key == key;
+}
+
+//----------
+HeadersExchange::FedUnbindModifier::FedUnbindModifier(string & origin) : fedOrigin(origin), shouldUnbind(false), shouldPropagate(false) {}
+HeadersExchange::FedUnbindModifier::FedUnbindModifier() : shouldUnbind(false), shouldPropagate(false) {}
+
+bool HeadersExchange::FedUnbindModifier::operator()(BoundKey & bk)
{
- return b->queue == queue && b->key == key;
+ if ("" == fedOrigin) {
+ shouldPropagate = bk.fedBinding.delOrigin();
+ } else {
+ shouldPropagate = bk.fedBinding.delOrigin(fedOrigin);
+ }
+ if (bk.fedBinding.count() == 0)
+ {
+ shouldUnbind = true;
+ }
+ return true;
}
+
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/HeadersExchange.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/HeadersExchange.h?rev=905322&r1=905321&r2=905322&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/HeadersExchange.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/HeadersExchange.h Mon Feb 1 16:10:33 2010
@@ -33,30 +33,52 @@
namespace broker {
-class HeadersExchange : public virtual Exchange {
- typedef std::pair<qpid::framing::FieldTable, Binding::shared_ptr> HeaderMap;
- typedef qpid::sys::CopyOnWriteArray<Binding::shared_ptr> Bindings;
+class HeadersExchange : public virtual Exchange {
+
+ struct BoundKey
+ {
+ Binding::shared_ptr binding;
+ FedBinding fedBinding;
+ BoundKey(Binding::shared_ptr binding_) : binding(binding_) {}
+ };
struct MatchArgs
{
const Queue::shared_ptr queue;
const qpid::framing::FieldTable* args;
MatchArgs(Queue::shared_ptr q, const qpid::framing::FieldTable* a);
- bool operator()(Exchange::Binding::shared_ptr b);
+ bool operator()(BoundKey & bk);
};
+
struct MatchKey
{
- const Queue::shared_ptr queue;
+ const Queue::shared_ptr queue;
const std::string& key;
MatchKey(Queue::shared_ptr q, const std::string& k);
- bool operator()(Exchange::Binding::shared_ptr b);
+ bool operator()(BoundKey & bk);
};
+ struct FedUnbindModifier
+ {
+ string fedOrigin;
+ bool shouldUnbind;
+ bool shouldPropagate;
+ FedUnbindModifier();
+ FedUnbindModifier(string & origin);
+ bool operator()(BoundKey & bk);
+ };
+
+ typedef qpid::sys::CopyOnWriteArray<BoundKey> Bindings;
+
Bindings bindings;
qpid::sys::Mutex lock;
static std::string getMatch(const framing::FieldTable* args);
+ protected:
+ void getNonFedArgs(const framing::FieldTable* args,
+ framing::FieldTable& nonFedArgs);
+
public:
static const std::string typeName;
@@ -85,6 +107,8 @@
QPID_BROKER_EXTERN virtual ~HeadersExchange();
+ virtual bool supportsDynamicBinding() { return true; }
+
static QPID_BROKER_EXTERN bool match(const qpid::framing::FieldTable& bindArgs, const qpid::framing::FieldTable& msgArgs);
static bool equal(const qpid::framing::FieldTable& bindArgs, const qpid::framing::FieldTable& msgArgs);
};
Modified: qpid/trunk/qpid/cpp/src/qpid/sys/CopyOnWriteArray.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/sys/CopyOnWriteArray.h?rev=905322&r1=905321&r2=905322&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/sys/CopyOnWriteArray.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/sys/CopyOnWriteArray.h Mon Feb 1 16:10:33 2010
@@ -103,6 +103,24 @@
return false;
}
+ template <class TestFn, class ModifierFn>
+ bool modify_if(TestFn f, ModifierFn & m)
+ {
+ if (!array)
+ return false;
+ {
+ Mutex::ScopedLock l(lock);
+ if (std::find_if(array->begin(), array->end(), f) != array->end())
+ {
+ ArrayPtr copy(new std::vector<T>(*array));
+ m(*std::find_if(copy->begin(), copy->end(), f));
+ array = copy;
+ return true;
+ }
+ }
+ return false;
+ }
+
template <class F>
F for_each(F f)
{
Modified: qpid/trunk/qpid/cpp/src/tests/federation.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/federation.py?rev=905322&r1=905321&r2=905322&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/federation.py (original)
+++ qpid/trunk/qpid/cpp/src/tests/federation.py Mon Feb 1 16:10:33 2010
@@ -585,6 +585,159 @@
self.verify_cleanup()
+ def test_dynamic_headers(self):
+ session = self.session
+ r_conn = self.connect(host=self.remote_host(), port=self.remote_port())
+ r_session = r_conn.session("test_dynamic_headers")
+
+ session.exchange_declare(exchange="fed.headers", type="headers")
+ r_session.exchange_declare(exchange="fed.headers", type="headers")
+
+ self.startQmf()
+ qmf = self.qmf
+
+ broker = qmf.getObjects(_class="broker")[0]
+ result = broker.connect(self.remote_host(), self.remote_port(), False, "PLAIN", "guest", "guest", "tcp")
+ self.assertEqual(result.status, 0)
+
+ link = qmf.getObjects(_class="link")[0]
+ result = link.bridge(False, "fed.headers", "fed.headers", "", "", "", False, False, True, 0)
+ self.assertEqual(result.status, 0)
+ bridge = qmf.getObjects(_class="bridge")[0]
+ sleep(5)
+
+ session.queue_declare(queue="fed1", exclusive=True, auto_delete=True)
+ session.exchange_bind(queue="fed1", exchange="fed.headers", binding_key="key1", arguments={'x-match':'any', 'class':'first'})
+ self.subscribe(queue="fed1", destination="f1")
+ queue = session.incoming("f1")
+
+ props = r_session.message_properties(application_headers={'class':'first'})
+ for i in range(1, 11):
+ r_session.message_transfer(destination="fed.headers", message=Message(props, "Message %d" % i))
+
+ for i in range(1, 11):
+ msg = queue.get(timeout=5)
+ content = msg.body
+ self.assertEqual("Message %d" % i, msg.body)
+ try:
+ extra = queue.get(timeout=1)
+ self.fail("Got unexpected message in queue: " + extra.body)
+ except Empty: None
+
+ result = bridge.close()
+ self.assertEqual(result.status, 0)
+ result = link.close()
+ self.assertEqual(result.status, 0)
+
+ self.verify_cleanup()
+
+ def test_dynamic_headers_reorigin(self):
+ session = self.session
+ r_conn = self.connect(host=self.remote_host(), port=self.remote_port())
+ r_session = r_conn.session("test_dynamic_headers_reorigin")
+
+ session.exchange_declare(exchange="fed.headers_reorigin", type="headers")
+ r_session.exchange_declare(exchange="fed.headers_reorigin", type="headers")
+
+ session.exchange_declare(exchange="fed.headers_reorigin_2", type="headers")
+ r_session.exchange_declare(exchange="fed.headers_reorigin_2", type="headers")
+
+ self.startQmf()
+ qmf = self.qmf
+ broker = qmf.getObjects(_class="broker")[0]
+ result = broker.connect(self.remote_host(), self.remote_port(), False, "PLAIN", "guest", "guest", "tcp")
+ self.assertEqual(result.status, 0)
+
+ session.queue_declare(queue="fed2", exclusive=True, auto_delete=True)
+ session.exchange_bind(queue="fed2", exchange="fed.headers_reorigin_2", binding_key="key2", arguments={'x-match':'any', 'class':'second'})
+ self.subscribe(queue="fed2", destination="f2")
+ queue2 = session.incoming("f2")
+
+ link = qmf.getObjects(_class="link")[0]
+ result = link.bridge(False, "fed.headers_reorigin", "fed.headers_reorigin", "", "", "", False, False, True, 0)
+ self.assertEqual(result.status, 0)
+ result = link.bridge(False, "fed.headers_reorigin_2", "fed.headers_reorigin_2", "", "", "", False, False, True, 0)
+ self.assertEqual(result.status, 0)
+
+ bridge = qmf.getObjects(_class="bridge")[0]
+ bridge2 = qmf.getObjects(_class="bridge")[1]
+ sleep(5)
+
+ session.queue_declare(queue="fed1", exclusive=True, auto_delete=True)
+ session.exchange_bind(queue="fed1", exchange="fed.headers_reorigin", binding_key="key1", arguments={'x-match':'any', 'class':'first'})
+ self.subscribe(queue="fed1", destination="f1")
+ queue = session.incoming("f1")
+
+ props = r_session.message_properties(application_headers={'class':'first'})
+ for i in range(1, 11):
+ r_session.message_transfer(destination="fed.headers_reorigin", message=Message(props, "Message %d" % i))
+
+ for i in range(1, 11):
+ msg = queue.get(timeout=5)
+ self.assertEqual("Message %d" % i, msg.body)
+ try:
+ extra = queue.get(timeout=1)
+ self.fail("Got unexpected message in queue: " + extra.body)
+ except Empty: None
+
+ result = bridge.close()
+ self.assertEqual(result.status, 0)
+
+ # Extra test: don't explicitly close() bridge2. When the link is closed,
+ # it should clean up bridge2 automagically. verify_cleanup() will detect
+ # if bridge2 isn't cleaned up and will fail the test.
+ #
+ #result = bridge2.close()
+ #self.assertEqual(result.status, 0)
+ result = link.close()
+ self.assertEqual(result.status, 0)
+
+ self.verify_cleanup()
+
+ def test_dynamic_headers_unbind(self):
+ session = self.session
+ r_conn = self.connect(host=self.remote_host(), port=self.remote_port())
+ r_session = r_conn.session("test_dynamic_headers_unbind")
+
+ session.exchange_declare(exchange="fed.headers_unbind", type="headers")
+ r_session.exchange_declare(exchange="fed.headers_unbind", type="headers")
+
+ self.startQmf()
+ qmf = self.qmf
+
+ broker = qmf.getObjects(_class="broker")[0]
+ result = broker.connect(self.remote_host(), self.remote_port(), False, "PLAIN", "guest", "guest", "tcp")
+ self.assertEqual(result.status, 0)
+
+ link = qmf.getObjects(_class="link")[0]
+ result = link.bridge(False, "fed.headers_unbind", "fed.headers_unbind", "", "", "", False, False, True, 0)
+ self.assertEqual(result.status, 0)
+ bridge = qmf.getObjects(_class="bridge")[0]
+ sleep(5)
+
+ session.queue_declare(queue="fed1", exclusive=True, auto_delete=True)
+ queue = qmf.getObjects(_class="queue", name="fed1")[0]
+ queue.update()
+ self.assertEqual(queue.bindingCount, 1,
+ "bindings not accounted for (expected 1, got %d)" % queue.bindingCount)
+
+ session.exchange_bind(queue="fed1", exchange="fed.headers_unbind", binding_key="key1", arguments={'x-match':'any', 'class':'first'})
+ queue.update()
+ self.assertEqual(queue.bindingCount, 2,
+ "bindings not accounted for (expected 2, got %d)" % queue.bindingCount)
+
+ session.exchange_unbind(queue="fed1", exchange="fed.headers_unbind", binding_key="key1")
+ queue.update()
+ self.assertEqual(queue.bindingCount, 1,
+ "bindings not accounted for (expected 1, got %d)" % queue.bindingCount)
+
+ result = bridge.close()
+ self.assertEqual(result.status, 0)
+ result = link.close()
+ self.assertEqual(result.status, 0)
+
+ self.verify_cleanup()
+
def getProperty(self, msg, name):
for h in msg.headers:
if hasattr(h, name): return getattr(h, name)
Added: qpid/trunk/qpid/cpp/src/tests/headers_federation.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/headers_federation.py?rev=905322&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/headers_federation.py (added)
+++ qpid/trunk/qpid/cpp/src/tests/headers_federation.py Mon Feb 1 16:10:33 2010
@@ -0,0 +1,99 @@
+#!/usr/bin/env python
+#
+# 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.
+#
+
+import sys
+from qpid.testlib import TestBase010
+from qpid.datatypes import Message
+from qpid.queue import Empty
+from time import sleep
+
+class HeadersFederationTests(TestBase010):
+
+ def remote_host(self):
+ return self.defines.get("remote-host", "localhost")
+
+ def remote_port(self):
+ return int(self.defines["remote-port"])
+
+ def verify_cleanup(self):
+ attempts = 0
+ total = len(self.qmf.getObjects(_class="bridge")) + len(self.qmf.getObjects(_class="link"))
+ while total > 0:
+ attempts += 1
+ if attempts >= 10:
+ self.fail("Bridges and links didn't clean up")
+ return
+ sleep(1)
+ total = len(self.qmf.getObjects(_class="bridge")) + len(self.qmf.getObjects(_class="link"))
+
+ def test_dynamic_headers_unbind(self):
+ session = self.session
+ r_conn = self.connect(host=self.remote_host(), port=self.remote_port())
+ r_session = r_conn.session("test_dynamic_headers_unbind")
+
+ session.exchange_declare(exchange="fed.headers_unbind", type="headers")
+ r_session.exchange_declare(exchange="fed.headers_unbind", type="headers")
+
+ self.startQmf()
+ qmf = self.qmf
+
+ broker = qmf.getObjects(_class="broker")[0]
+ result = broker.connect(self.remote_host(), self.remote_port(), False, "PLAIN", "guest", "guest", "tcp")
+ self.assertEqual(result.status, 0)
+
+ link = qmf.getObjects(_class="link")[0]
+ result = link.bridge(False, "fed.headers_unbind", "fed.headers_unbind", "", "", "", False, False, True, 0)
+ self.assertEqual(result.status, 0)
+ bridge = qmf.getObjects(_class="bridge")[0]
+ sleep(5)
+
+ session.queue_declare(queue="fed1", exclusive=True, auto_delete=True)
+ queue = qmf.getObjects(_class="queue", name="fed1")[0]
+ queue.update()
+ self.assertEqual(queue.bindingCount, 1,
+ "bindings not accounted for (expected 1, got %d)" % queue.bindingCount)
+
+ session.exchange_bind(queue="fed1", exchange="fed.headers_unbind", binding_key="key1", arguments={'x-match':'any', 'class':'first'})
+ queue.update()
+ self.assertEqual(queue.bindingCount, 2,
+ "bindings not accounted for (expected 2, got %d)" % queue.bindingCount)
+
+ session.exchange_unbind(queue="fed1", exchange="fed.headers_unbind", binding_key="key1")
+ queue.update()
+ self.assertEqual(queue.bindingCount, 1,
+ "bindings not accounted for (expected 1, got %d)" % queue.bindingCount)
+
+ result = bridge.close()
+ self.assertEqual(result.status, 0)
+ result = link.close()
+ self.assertEqual(result.status, 0)
+
+ self.verify_cleanup()
+
+ def getProperty(self, msg, name):
+ for h in msg.headers:
+ if hasattr(h, name): return getattr(h, name)
+ return None
+
+ def getAppHeader(self, msg, name):
+ headers = self.getProperty(msg, "application_headers")
+ if headers:
+ return headers[name]
+ return None
Added: qpid/trunk/qpid/cpp/src/tests/run_headers_federation_tests
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/run_headers_federation_tests?rev=905322&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/run_headers_federation_tests (added)
+++ qpid/trunk/qpid/cpp/src/tests/run_headers_federation_tests Mon Feb 1 16:10:33 2010
@@ -0,0 +1,49 @@
+#!/bin/sh
+
+#
+# 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.
+#
+
+# Run the federation tests for the Headers Exchange.
+
+source ./test_env.sh
+
+trap stop_brokers INT TERM QUIT
+
+start_brokers() {
+ ../qpidd --daemon --port 0 --no-data-dir --no-module-dir --auth no > qpidd.port
+ LOCAL_PORT=`cat qpidd.port`
+ ../qpidd --daemon --port 0 --no-data-dir --no-module-dir --auth no > qpidd.port
+ REMOTE_PORT=`cat qpidd.port`
+}
+
+stop_brokers() {
+ $QPIDD_EXEC --no-module-dir -q --port $LOCAL_PORT
+ $QPIDD_EXEC --no-module-dir -q --port $REMOTE_PORT
+}
+
+if test -d ${PYTHON_DIR} ; then
+ start_brokers
+ echo "Running HeadersExchange federation tests using brokers on ports $LOCAL_PORT $REMOTE_PORT"
+ $QPID_PYTHON_TEST -m headers_federation -b localhost:$LOCAL_PORT -Dremote-port=$REMOTE_PORT $@
+ RETCODE=$?
+ stop_brokers
+ if test x$RETCODE != x0; then
+ echo "FAIL federation tests"; exit 1;
+ fi
+fi
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org