Author: jonathan
Date: Fri Oct 8 17:11:06 2010
New Revision: 1005908
URL: http://svn.apache.org/viewvc?rev=1005908&view=rev
Log:
Registers the amq.failover exchange in the management exchange.
Modified:
qpid/trunk/qpid/cpp/src/qpid/broker/ExchangeRegistry.cpp
qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp
qpid/trunk/qpid/cpp/src/qpid/cluster/FailoverExchange.cpp
qpid/trunk/qpid/cpp/src/qpid/cluster/FailoverExchange.h
qpid/trunk/qpid/cpp/src/tests/cluster_tests.py
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/ExchangeRegistry.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/ExchangeRegistry.cpp?rev=1005908&r1=1005907&r2=1005908&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/ExchangeRegistry.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/ExchangeRegistry.cpp Fri Oct 8
17:11:06 2010
@@ -45,7 +45,7 @@ pair<Exchange::shared_ptr, bool> Exchang
if (i == exchanges.end()) {
Exchange::shared_ptr exchange;
- if(type == TopicExchange::typeName){
+ if (type == TopicExchange::typeName){
exchange = Exchange::shared_ptr(new TopicExchange(name, durable,
args, parent, broker));
}else if(type == DirectExchange::typeName){
exchange = Exchange::shared_ptr(new DirectExchange(name, durable,
args, parent, broker));
Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp?rev=1005908&r1=1005907&r2=1005908&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp Fri Oct 8 17:11:06 2010
@@ -282,7 +282,7 @@ Cluster::Cluster(const ClusterSettings&
broker.setClusterTimer(std::auto_ptr<sys::Timer>(timer));
// Failover exchange provides membership updates to clients.
- failoverExchange.reset(new FailoverExchange(this));
+ failoverExchange.reset(new FailoverExchange(broker.GetVhostObject(),
&broker));
broker.getExchanges().registerExchange(failoverExchange);
// Update exchange is used during updates to replicate messages
Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/FailoverExchange.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/FailoverExchange.cpp?rev=1005908&r1=1005907&r2=1005908&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/FailoverExchange.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/FailoverExchange.cpp Fri Oct 8
17:11:06 2010
@@ -38,11 +38,11 @@ using namespace std;
using namespace broker;
using namespace framing;
-const string FailoverExchange::TYPE_NAME("amq.failover");
-
-FailoverExchange::FailoverExchange(management::Manageable* parent) :
Exchange(TYPE_NAME, parent) {
+const string FailoverExchange::typeName("amq.failover");
+
+FailoverExchange::FailoverExchange(management::Manageable* parent, Broker* b)
: Exchange(typeName, parent, b ) {
if (mgmtExchange != 0)
- mgmtExchange->set_type(TYPE_NAME);
+ mgmtExchange->set_type(typeName);
}
void FailoverExchange::setUrls(const vector<Url>& u) {
@@ -58,7 +58,7 @@ void FailoverExchange::updateUrls(const
boost::bind(&FailoverExchange::sendUpdate, this, _1));
}
-string FailoverExchange::getType() const { return TYPE_NAME; }
+string FailoverExchange::getType() const { return typeName; }
bool FailoverExchange::bind(Queue::shared_ptr queue, const string&, const
framing::FieldTable*) {
Lock l(lock);
@@ -77,7 +77,7 @@ bool FailoverExchange::isBound(Queue::sh
}
void FailoverExchange::route(Deliverable&, const string& , const
framing::FieldTable* ) {
- QPID_LOG(warning, "Message received by exchange " << TYPE_NAME << "
ignoring");
+ QPID_LOG(warning, "Message received by exchange " << typeName << "
ignoring");
}
void FailoverExchange::sendUpdate(const Queue::shared_ptr& queue) {
@@ -88,16 +88,17 @@ void FailoverExchange::sendUpdate(const
array.add(boost::shared_ptr<Str16Value>(new Str16Value(i->str())));
const ProtocolVersion v;
boost::intrusive_ptr<Message> msg(new Message);
- AMQFrame command(MessageTransferBody(v, TYPE_NAME, 1, 0));
+ AMQFrame command(MessageTransferBody(v, typeName, 1, 0));
command.setLastSegment(false);
msg->getFrames().append(command);
AMQHeaderBody header;
header.get<MessageProperties>(true)->setContentLength(0);
-
header.get<MessageProperties>(true)->getApplicationHeaders().setArray(TYPE_NAME,
array);
+
header.get<MessageProperties>(true)->getApplicationHeaders().setArray(typeName,
array);
AMQFrame headerFrame(header);
headerFrame.setFirstSegment(false);
msg->getFrames().append(headerFrame);
DeliverableMessage(msg).deliverTo(queue);
}
+
}} // namespace cluster
Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/FailoverExchange.h
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/FailoverExchange.h?rev=1005908&r1=1005907&r2=1005908&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/FailoverExchange.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/FailoverExchange.h Fri Oct 8 17:11:06
2010
@@ -38,9 +38,9 @@ namespace cluster {
class FailoverExchange : public broker::Exchange
{
public:
- static const std::string TYPE_NAME;
+ static const std::string typeName;
- FailoverExchange(management::Manageable* parent);
+ FailoverExchange(management::Manageable* parent, broker::Broker* b);
/** Set the URLs but don't send an update.*/
void setUrls(const std::vector<Url>&);
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=1005908&r1=1005907&r2=1005908&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/cluster_tests.py (original)
+++ qpid/trunk/qpid/cpp/src/tests/cluster_tests.py Fri Oct 8 17:11:06 2010
@@ -18,7 +18,7 @@
# under the License.
#
-import os, signal, sys, time, imp, re
+import os, signal, sys, time, imp, re, subprocess
from qpid import datatypes, messaging
from qpid.brokertest import *
from qpid.harness import Skipped
@@ -214,6 +214,15 @@ acl allow all all
for b in cluster: b.ready() # Make sure all brokers still running.
+ def test_amqfailover_visible(self):
+ """Verify that the amq.failover exchange can be seen by
+ QMF-based tools - regression test for BZ615300."""
+ broker1 = self.cluster(1)[0]
+ broker2 = self.cluster(1)[0]
+ qs = subprocess.Popen(["qpid-stat", "-e", broker1.host_port()],
stdout=subprocess.PIPE)
+ out = qs.communicate()[0]
+ assert out.find("amq.failover") > 0
+
class LongTests(BrokerTest):
"""Tests that can run for a long time if -DDURATION=<minutes> is set"""
def duration(self):
@@ -525,4 +534,3 @@ class StoreTests(BrokerTest):
self.assertEqual(c.get_message("q").content, "x")
b = cluster.start("b")
self.assertEqual(c.get_message("q").content, "y")
-
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscr...@qpid.apache.org
