qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From acon...@apache.org
Subject svn commit: r1233642 - in /qpid/branches/qpid-3603-2/qpid/cpp/src: qpid/ha/WiringReplicator.cpp tests/ha_tests.py
Date Thu, 19 Jan 2012 23:03:12 GMT
Author: aconway
Date: Thu Jan 19 23:03:12 2012
New Revision: 1233642

URL: http://svn.apache.org/viewvc?rev=1233642&view=rev
Log:
QPID-3603: Minor refactor.

Modified:
    qpid/branches/qpid-3603-2/qpid/cpp/src/qpid/ha/WiringReplicator.cpp
    qpid/branches/qpid-3603-2/qpid/cpp/src/tests/ha_tests.py

Modified: qpid/branches/qpid-3603-2/qpid/cpp/src/qpid/ha/WiringReplicator.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3603-2/qpid/cpp/src/qpid/ha/WiringReplicator.cpp?rev=1233642&r1=1233641&r2=1233642&view=diff
==============================================================================
--- qpid/branches/qpid-3603-2/qpid/cpp/src/qpid/ha/WiringReplicator.cpp (original)
+++ qpid/branches/qpid-3603-2/qpid/cpp/src/qpid/ha/WiringReplicator.cpp Thu Jan 19 23:03:12
2012
@@ -31,6 +31,9 @@
 #include "qmf/org/apache/qpid/broker/EventQueueDelete.h"
 #include "qmf/org/apache/qpid/broker/EventSubscribe.h"
 
+namespace qpid {
+namespace ha {
+
 using qmf::org::apache::qpid::broker::EventBind;
 using qmf::org::apache::qpid::broker::EventExchangeDeclare;
 using qmf::org::apache::qpid::broker::EventExchangeDelete;
@@ -38,50 +41,49 @@ using qmf::org::apache::qpid::broker::Ev
 using qmf::org::apache::qpid::broker::EventQueueDelete;
 using qmf::org::apache::qpid::broker::EventSubscribe;
 
-namespace qpid {
-namespace ha {
-
+using std::string;
 using types::Variant;
 using namespace broker;
 
 namespace{
 
-const std::string QPID_REPLICATE("qpid.replicate");
-const std::string ALL("all");
-const std::string WIRING("wiring");
-
-const std::string CLASS_NAME("_class_name");
-const std::string PACKAGE_NAME("_package_name");
-const std::string VALUES("_values");
-const std::string EVENT("_event");
-const std::string SCHEMA_ID("_schema_id");
-const std::string QUERY_RESPONSE("_query_response");
-
-const std::string ARGUMENTS("arguments");
-const std::string QUEUE("queue");
-const std::string EXCHANGE("exchange");
-const std::string BIND("bind");
-const std::string ARGS("args");
-const std::string DURABLE("durable");
-const std::string QNAME("qName");
-const std::string AUTODEL("autoDel");
-const std::string ALTEX("altEx");
-const std::string USER("user");
-const std::string RHOST("rhost");
-const std::string EXTYPE("exType");
-const std::string EXNAME("exName");
-const std::string AUTODELETE("autoDelete");
-const std::string NAME("name");
-const std::string TYPE("type");
-const std::string DISP("disp");
-const std::string CREATED("created");
-
-
-const std::string QMF_OPCODE("qmf.opcode");
-const std::string QMF_CONTENT("qmf.content");
-const std::string QMF2("qmf2");
+const string QPID_REPLICATE("qpid.replicate");
+const string ALL("all");
+const string WIRING("wiring");
+
+const string CLASS_NAME("_class_name");
+const string PACKAGE_NAME("_package_name");
+const string VALUES("_values");
+const string EVENT("_event");
+const string SCHEMA_ID("_schema_id");
+const string QUERY_RESPONSE("_query_response");
+
+const string ARGUMENTS("arguments");
+const string QUEUE("queue");
+const string EXCHANGE("exchange");
+const string BIND("bind");
+const string ARGS("args");
+const string DURABLE("durable");
+const string QNAME("qName");
+const string AUTODEL("autoDel");
+const string ALTEX("altEx");
+const string USER("user");
+const string RHOST("rhost");
+const string EXTYPE("exType");
+const string EXNAME("exName");
+const string AUTODELETE("autoDelete");
+const string NAME("name");
+const string TYPE("type");
+const string DISP("disp");
+const string CREATED("created");
+const string KEY("key");
+
+
+const string QMF_OPCODE("qmf.opcode");
+const string QMF_CONTENT("qmf.content");
+const string QMF2("qmf2");
 
-const std::string QPID_WIRING_REPLICATOR("qpid.wiring-replicator");
+const string QPID_WIRING_REPLICATOR("qpid.wiring-replicator");
 
 
 bool isQMFv2(const Message& message)
@@ -95,7 +97,7 @@ template <class T> bool match(Variant::M
     return T::match(schema[CLASS_NAME], schema[PACKAGE_NAME]);
 }
 
-bool isReplicated(const std::string& value) {
+bool isReplicated(const string& value) {
     return value == ALL || value == WIRING;
 }
 bool isReplicated(const framing::FieldTable& f) {
@@ -109,20 +111,22 @@ bool isReplicated(const Variant::Map& m)
 } // namespace
 
 
-WiringReplicator::WiringReplicator(const std::string& name, Broker& b) : Exchange(name),
broker(b) {}
+WiringReplicator::WiringReplicator(const string& name, Broker& b) : Exchange(name),
broker(b) {}
 
 WiringReplicator::~WiringReplicator() {}
 
-void WiringReplicator::route(Deliverable& msg, const std::string& /*key*/, const
framing::FieldTable* headers) {
+void WiringReplicator::route(Deliverable& msg, const string& /*key*/, const framing::FieldTable*
headers) {
     try {
         // FIXME aconway 2011-11-21: outer error handling, e.g. for decoding error.
         if (!isQMFv2(msg.getMessage()) || !headers)
             throw Exception("Unexpected message, not QMF2 event or query response.");
-        // FIXME aconway 2011-11-21: string constants
-        if (headers->getAsString(QMF_CONTENT) == EVENT) { //decode as list
-            std::string content = msg.getMessage().getFrames().getContent();
-            Variant::List list;
-            amqp_0_10::ListCodec::decode(content, list);
+        // decode as list
+        string content = msg.getMessage().getFrames().getContent();
+        Variant::List list;
+        amqp_0_10::ListCodec::decode(content, list);
+
+        QPID_LOG(critical, "FIXME WiringReplicator message: " << list);
+        if (headers->getAsString(QMF_CONTENT) == EVENT) {
             for (Variant::List::iterator i = list.begin(); i != list.end(); ++i) {
                 // FIXME aconway 2011-11-18: should be iterating list?
                 Variant::Map& map = list.front().asMap();
@@ -133,16 +137,14 @@ void WiringReplicator::route(Deliverable
                 else if (match<EventExchangeDeclare>(schema)) doEventExchangeDeclare(values);
                 else if (match<EventExchangeDelete>(schema)) doEventExchangeDelete(values);
                 else if (match<EventBind>(schema)) doEventBind(values);
+                // FIXME aconway 2011-11-21: handle unbind & all other events.
                 else if (match<EventSubscribe>(schema)) {} // Deliberately ignored.
                 else throw(Exception(QPID_MSG("WiringReplicator received unexpected event,
schema=" << schema)));
             }
         } else if (headers->getAsString(QMF_OPCODE) == QUERY_RESPONSE) {
-            //decode as list
-            std::string content = msg.getMessage().getFrames().getContent();
-            Variant::List list;
-            amqp_0_10::ListCodec::decode(content, list);
+            QPID_LOG(critical, "FIXME WiringReplicator response: " << list);
             for (Variant::List::iterator i = list.begin(); i != list.end(); ++i) {
-                std::string type = i->asMap()[SCHEMA_ID].asMap()[CLASS_NAME];
+                string type = i->asMap()[SCHEMA_ID].asMap()[CLASS_NAME];
                 Variant::Map& values = i->asMap()[VALUES].asMap();
                 if (isReplicated(values[ARGUMENTS].asMap())) {
                     framing::FieldTable args;
@@ -162,7 +164,7 @@ void WiringReplicator::route(Deliverable
 }
 
 void WiringReplicator::doEventQueueDeclare(Variant::Map& values) {
-    std::string name = values[QNAME].asString();
+    string name = values[QNAME].asString();
     if (values[DISP] == CREATED && isReplicated(values[ARGS].asMap())) {
         QPID_LOG(debug, "Creating replicated queue " << name);
         framing::FieldTable args;
@@ -182,7 +184,7 @@ void WiringReplicator::doEventQueueDecla
 }
 
 void WiringReplicator::doEventQueueDelete(Variant::Map& values) {
-    std::string name = values[QNAME].asString();
+    string name = values[QNAME].asString();
     boost::shared_ptr<Queue> queue = broker.getQueues().find(name);
     if (queue && isReplicated(queue->getSettings())) {
         QPID_LOG(debug, "Deleting replicated queue " << name);
@@ -194,10 +196,11 @@ void WiringReplicator::doEventQueueDelet
 }
 
 void WiringReplicator::doEventExchangeDeclare(Variant::Map& values) {
-    if (values[DISP] == CREATED && isReplicated(values[ARGS].asMap())) {
-        std::string name = values[EXNAME].asString();
+    Variant::Map argsMap(values[ARGS].asMap());
+    if (values[DISP] == CREATED && isReplicated(argsMap)) {
+        string name = values[EXNAME].asString();
         framing::FieldTable args;
-        amqp_0_10::translate(values[ARGS].asMap(), args);
+        amqp_0_10::translate(argsMap, args);
         QPID_LOG(debug, "Creating replicated exchange " << name);
         if (!broker.createExchange(
                 name,
@@ -213,11 +216,11 @@ void WiringReplicator::doEventExchangeDe
 }
 
 void WiringReplicator::doEventExchangeDelete(Variant::Map& values) {
-    std::string name = values[EXNAME].asString();
+    string name = values[EXNAME].asString();
     try {
         boost::shared_ptr<Exchange> exchange = broker.getExchanges().get(name);
         if (exchange && isReplicated(exchange->getArgs())) {
-            QPID_LOG(warning, "Deleting replicated exchange " << name);
+            QPID_LOG(debug, "Deleting replicated exchange " << name);
             broker.deleteExchange(
                 name,
                 values[USER].asString(),
@@ -226,9 +229,22 @@ void WiringReplicator::doEventExchangeDe
     } catch (const framing::NotFoundException&) {}
 }
 
-void WiringReplicator::doEventBind(Variant::Map&) {
-    QPID_LOG(error, "FIXME WiringReplicator: Not yet implemented - replicate bindings.");
-    // FIXME aconway 2011-11-18: only replicated binds of replicated q to replicated ex.
+void WiringReplicator::doEventBind(Variant::Map& values) {
+    QPID_LOG(critical, "FIXME doEventBind " << values);
+    try {
+        boost::shared_ptr<Exchange> exchange = broker.getExchanges().get(values[EXNAME].asString());
+        boost::shared_ptr<Queue> queue = broker.getQueues().find(values[QNAME].asString());
+        // We only replicated a binds for a replicated queue to replicated exchange.
+        if (isReplicated(exchange->getArgs()) && isReplicated(queue->getSettings()))
{
+            framing::FieldTable args;
+            amqp_0_10::translate(args, values[ARGS].asMap());
+            string key = values[KEY].asString();
+            QPID_LOG(debug, "Replicated binding exchange=" << exchange->getName()
+                     << " queue=" << queue->getName()
+                     << " key=" << key);
+            exchange->bind(queue, key, &args);
+        }
+    } catch (const framing::NotFoundException&) {} // Ignore unreplicated queue or exchange.
 }
 
 void WiringReplicator::doResponseQueue(Variant::Map& values) {
@@ -260,11 +276,12 @@ void WiringReplicator::doResponseExchang
     }
 }
 
-void WiringReplicator::doResponseBind(Variant::Map& ) {
-    QPID_LOG(error, "FIXME WiringReplicator: Not yet implemented - catch-up replicate bindings.");
+void WiringReplicator::doResponseBind(Variant::Map& values) {
+    QPID_LOG(critical, "FIXME doResponseBind " << values);
+    throw Exception("FIXME WiringReplicator: Not yet implemented - catch-up replicate bindings.");
 }
 
-boost::shared_ptr<Exchange> WiringReplicator::create(const std::string& target,
Broker& broker)
+boost::shared_ptr<Exchange> WiringReplicator::create(const string& target, Broker&
broker)
 {
     boost::shared_ptr<Exchange> exchange;
     if (isWiringReplicatorDestination(target)) {
@@ -274,18 +291,18 @@ boost::shared_ptr<Exchange> WiringReplic
     return exchange;
 }
 
-bool WiringReplicator::isWiringReplicatorDestination(const std::string& target)
+bool WiringReplicator::isWiringReplicatorDestination(const string& target)
 {
     return target == QPID_WIRING_REPLICATOR;
 }
 
-bool WiringReplicator::bind(boost::shared_ptr<Queue>, const std::string&, const
framing::FieldTable*) { return false; }
-bool WiringReplicator::unbind(boost::shared_ptr<Queue>, const std::string&, const
framing::FieldTable*) { return false; }
-bool WiringReplicator::isBound(boost::shared_ptr<Queue>, const std::string* const,
const framing::FieldTable* const) { return false; }
+bool WiringReplicator::bind(boost::shared_ptr<Queue>, const string&, const framing::FieldTable*)
{ return false; }
+bool WiringReplicator::unbind(boost::shared_ptr<Queue>, const string&, const framing::FieldTable*)
{ return false; }
+bool WiringReplicator::isBound(boost::shared_ptr<Queue>, const string* const, const
framing::FieldTable* const) { return false; }
 
-const std::string WiringReplicator::typeName(QPID_WIRING_REPLICATOR);
+const string WiringReplicator::typeName(QPID_WIRING_REPLICATOR);
 
-std::string WiringReplicator::getType() const
+string WiringReplicator::getType() const
 {
     return typeName;
 }

Modified: qpid/branches/qpid-3603-2/qpid/cpp/src/tests/ha_tests.py
URL: http://svn.apache.org/viewvc/qpid/branches/qpid-3603-2/qpid/cpp/src/tests/ha_tests.py?rev=1233642&r1=1233641&r2=1233642&view=diff
==============================================================================
--- qpid/branches/qpid-3603-2/qpid/cpp/src/tests/ha_tests.py (original)
+++ qpid/branches/qpid-3603-2/qpid/cpp/src/tests/ha_tests.py Thu Jan 19 23:03:12 2012
@@ -88,16 +88,16 @@ class ShortTests(BrokerTest):
 #         self.assert_browse(s, "q01", ["01", "04", "e01"])
 #         self.assert_browse(s, "q02", []) # wiring only
 #         self.assert_missing(s,"q03")
-#         s.sender("e01").send(Message("e01")) # Verify bind
-#         self.assert_browse(s, "q02", ["e01"])
+        s.sender("e01").send(Message("e01")) # Verify bind
+        self.assert_browse(s, "q02", ["e01"])
 
         for a in ["q1", "q2", "e1"]: self.wait(s,a)
         # FIXME aconway 2011-11-18: replicate messages
 #         self.assert_browse(s, "q1", ["1", "4", "e1"])
 #         self.assert_browse(s, "q2", []) # wiring only
 #         self.assert_missing(s,"q3")
-#         s.sender("e1").send(Message("e1")) # Verify bind
-#         self.assert_browse(s, "q2", ["e1"])
+        s.sender("e1").send(Message("e1")) # Verify bind
+        self.assert_browse(s, "q2", ["e1"])
 
 
 if __name__ == "__main__":



---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project:      http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org


Mime
View raw message