You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ac...@apache.org on 2011/04/15 22:40:07 UTC
svn commit: r1092804 - in /qpid/trunk/qpid/cpp/src: qpid/broker/Exchange.cpp
tests/brokertest.py tests/cluster_tests.py
Author: aconway
Date: Fri Apr 15 20:40:06 2011
New Revision: 1092804
URL: http://svn.apache.org/viewvc?rev=1092804&view=rev
Log:
Revert "QPID-3208: Exchanges make best effort to route messages if there is an error."
This reverts commit r1092765 which introduced test failures in make check.
Modified:
qpid/trunk/qpid/cpp/src/qpid/broker/Exchange.cpp
qpid/trunk/qpid/cpp/src/tests/brokertest.py
qpid/trunk/qpid/cpp/src/tests/cluster_tests.py
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=1092804&r1=1092803&r2=1092804&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Exchange.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Exchange.cpp Fri Apr 15 20:40:06 2011
@@ -29,7 +29,6 @@
#include "qpid/framing/MessageProperties.h"
#include "qpid/framing/reply_exceptions.h"
#include "qpid/broker/DeliverableMessage.h"
-#include "stdexcept"
using namespace qpid::broker;
using namespace qpid::framing;
@@ -71,44 +70,6 @@ Exchange::PreRoute::~PreRoute(){
}
}
-namespace {
-/** Store information about an exception to be thrown later.
- * If multiple exceptions are stored, save the first of the "most severe"
- * exceptions, SESSION is les sever than CONNECTION etc.
- */
-class ExInfo {
- public:
- enum Type { NONE, SESSION, CONNECTION, OTHER };
-
- ExInfo(string ex) : type(NONE), code(0), exchange(ex) {}
-
- void store(Type type_, const char* what_, int code_, const boost::shared_ptr<Queue>& queue) {
- QPID_LOG(error, "Exchange " << exchange << " cannot deliver to queue "
- << queue->getName() << ": " << what_);
- if (type < type_) { // Replace less severe error
- type = type_;
- what = what_;
- code = code_;
- }
- }
-
- void raise() {
- switch (type) {
- case NONE: break;
- case SESSION: throw qpid::SessionException(qpid::framing::execution::ErrorCode(code), what);
- case CONNECTION: throw qpid::ConnectionException(qpid::framing::connection::CloseCode(code), what);
- case OTHER: throw std::runtime_error(what);
- }
- }
-
- private:
- Type type;
- string what;
- int code;
- string exchange;
-};
-}
-
void Exchange::doRoute(Deliverable& msg, ConstBindingList b)
{
int count = 0;
@@ -119,25 +80,11 @@ void Exchange::doRoute(Deliverable& msg,
msg.getMessage().blockContentRelease();
}
-
- ExInfo error(getName()); // Save errors to throw at the end.
for(std::vector<Binding::shared_ptr>::const_iterator i = b->begin(); i != b->end(); i++, count++) {
- try {
- msg.deliverTo((*i)->queue);
- if ((*i)->mgmtBinding != 0)
- (*i)->mgmtBinding->inc_msgMatched();
- }
- catch (const SessionException& e) {
- error.store(ExInfo::SESSION, e.what(), int(e.code), (*i)->queue);
- }
- catch (const ConnectionException& e) {
- error.store(ExInfo::CONNECTION, e.what(), int(e.code), (*i)->queue);
- }
- catch (const std::exception& e) {
- error.store(ExInfo::OTHER, e.what(), 0, (*i)->queue);
- }
+ msg.deliverTo((*i)->queue);
+ if ((*i)->mgmtBinding != 0)
+ (*i)->mgmtBinding->inc_msgMatched();
}
- error.raise();
}
if (mgmtExchange != 0)
Modified: qpid/trunk/qpid/cpp/src/tests/brokertest.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/brokertest.py?rev=1092804&r1=1092803&r2=1092804&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/brokertest.py (original)
+++ qpid/trunk/qpid/cpp/src/tests/brokertest.py Fri Apr 15 20:40:06 2011
@@ -484,24 +484,18 @@ class BrokerTest(TestCase):
cluster = Cluster(self, count, args, expect=expect, wait=wait)
return cluster
- def browse(self, session, queue, timeout=0):
+ def assert_browse(self, session, queue, expect_contents, timeout=0):
"""Assert that the contents of messages on queue (as retrieved
using session and timeout) exactly match the strings in
expect_contents"""
+
r = session.receiver("%s;{mode:browse}"%(queue))
+ actual_contents = []
try:
- contents = []
- try:
- while True: contents.append(r.fetch(timeout=timeout).content)
- except messaging.Empty: pass
- finally: pass #FIXME aconway 2011-04-14: r.close()
- return contents
-
- def assert_browse(self, session, queue, expect_contents, timeout=0):
- """Assert that the contents of messages on queue (as retrieved
- using session and timeout) exactly match the strings in
- expect_contents"""
- actual_contents = self.browse(session, queue, timeout)
+ for c in expect_contents: actual_contents.append(r.fetch(timeout=timeout).content)
+ while True: actual_contents.append(r.fetch(timeout=0).content) # Check for extra messages.
+ except messaging.Empty: pass
+ r.close()
self.assertEqual(expect_contents, actual_contents)
def join(thread, timeout=10):
Modified: qpid/trunk/qpid/cpp/src/tests/cluster_tests.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/cluster_tests.py?rev=1092804&r1=1092803&r2=1092804&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/cluster_tests.py (original)
+++ qpid/trunk/qpid/cpp/src/tests/cluster_tests.py Fri Apr 15 20:40:06 2011
@@ -449,36 +449,6 @@ acl allow all all
cluster.start()
verify(cluster[1])
- def test_binding_order(self):
- """Regression test for binding order inconsistency in cluster"""
- cluster = self.cluster(1)
- c0 = cluster[0].connect()
- s0 = c0.session()
- # Declare multiple queues bound to same key on amq.topic
- def declare(q,max=0):
- if max: declare = 'x-declare:{arguments:{"qpid.max_count":%d}}'%max
- else: declare = 'x-declare:{}'
- bind='x-bindings:[{queue:%s,key:key,exchange:"amq.topic"}]'%(q)
- s0.sender("%s;{create:always,node:{%s,%s}}" % (q,declare,bind))
- declare('d',max=4) # Only one with a limit
- for q in ['c', 'b','a']: declare(q)
- # Add a cluster member, send enough messages to exceed the max count
- cluster.start()
- try:
- s = s0.sender('amq.topic/key')
- for m in xrange(1,6): s.send(Message(str(m)))
- self.fail("Expected capacity exceeded exception")
- except messaging.exceptions.TargetCapacityExceeded: pass
- c1 = cluster[1].connect()
- s1 = c1.session()
- s0 = c0.session() # Old session s0 is broken by exception.
- # Verify queue contents are consistent.
- for q in ['a','b','c','d']:
- self.assertEqual(self.browse(s0, q), self.browse(s1, q))
- # Verify queue contents are "best effort"
- for q in ['a','b','c']: self.assert_browse(s1,q,[str(n) for n in xrange(1,6)])
- self.assert_browse(s1,'d',[str(n) for n in xrange(1,5)])
-
class LongTests(BrokerTest):
"""Tests that can run for a long time if -DDURATION=<minutes> is set"""
def duration(self):
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org