QPID-3603: HA don't replicate excluseive, auto-delete, non-timeout queues.

Such queues don't need to be replicated because they are destroyed when
as the owning session disconnects, so won't survive a failover.

This eliminsates managment subscriptio queues from replication.

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@1339268 13f79535-47bb-0310-9956-ffa450edef68

2 files changed, 26 insertions, 6 deletions

diff --git a/cpp/src/qpid/ha/BrokerReplicator.cpp b/cpp/src/qpid/ha/BrokerReplicator.cpp
index ee6b6e42fb..60ec4ea59f 100644
--- a/cpp/src/qpid/ha/BrokerReplicator.cpp
+++ b/cpp/src/qpid/ha/BrokerReplicator.cpp
@@ -73,6 +73,8 @@ const string ARGS("args");
 const string ARGUMENTS("arguments");
 const string AUTODEL("autoDel");
 const string AUTODELETE("autoDelete");
+const string EXCL("excl");
+const string EXCLUSIVE("exclusive");
 const string BIND("bind");
 const string UNBIND("unbind");
 const string BINDING("binding");
@@ -278,7 +280,9 @@ void BrokerReplicator::route(Deliverable& msg) {
 void BrokerReplicator::doEventQueueDeclare(Variant::Map& values) {
     string name = values[QNAME].asString();
     Variant::Map argsMap = asMapVoid(values[ARGS]);
-    if (!haBroker.replicateLevel(argsMap)) return; // Not a replicated queue.
+    if (!isReplicated(
+            values[ARGS].asMap(), values[AUTODEL].asBool(), values[EXCL].asBool()))
+        return;
     if (values[DISP] == CREATED && haBroker.replicateLevel(argsMap)) {
         framing::FieldTable args;
         amqp_0_10::translate(argsMap, args);
@@ -286,19 +290,20 @@ void BrokerReplicator::doEventQueueDeclare(Variant::Map& values) {
         // The queue was definitely created on the primary.
         if (broker.getQueues().find(name)) {
             broker.getQueues().destroy(name);
-            QPID_LOG(warning, logPrefix << "queue declare event, replaced exsiting: " << name);
+            QPID_LOG(warning, logPrefix << "queue declare event, replaced exsiting: "
+                     << name);
         }
         std::pair<boost::shared_ptr<Queue>, bool> result =
             broker.createQueue(
                 name,
                 values[DURABLE].asBool(),
                 values[AUTODEL].asBool(),
-                0 /*i.e. no owner regardless of exclusivity on master*/,
+                0, // no owner regardless of exclusivity on primary
                 values[ALTEX].asString(),
                 args,
                 values[USER].asString(),
                 values[RHOST].asString());
-        assert(result.second);
+        assert(result.second);  // Should be true since we destroyed existing queue above
         QPID_LOG(debug, logPrefix << "queue declare event: " << name);
         startQueueReplicator(result.first);
     }
@@ -420,7 +425,10 @@ void BrokerReplicator::doEventUnbind(Variant::Map& values) {
 
 void BrokerReplicator::doResponseQueue(Variant::Map& values) {
     Variant::Map argsMap(asMapVoid(values[ARGUMENTS]));
-    if (!haBroker.replicateLevel(argsMap)) return;
+    if (!isReplicated(values[ARGUMENTS].asMap(),
+                      values[AUTODELETE].asBool(),
+                      values[EXCLUSIVE].asBool()))
+        return;
     framing::FieldTable args;
     amqp_0_10::translate(argsMap, args);
     string name(values[NAME].asString());
@@ -522,6 +530,17 @@ void BrokerReplicator::doResponseHaBroker(Variant::Map& values) {
     }
 }
 
+namespace {
+const std::string AUTO_DELETE_TIMEOUT("qpid.auto_delete_timeout");
+}
+
+bool BrokerReplicator::isReplicated(
+    const Variant::Map& args, bool autodelete, bool exclusive)
+{
+    bool ignore = autodelete && exclusive && args.find(AUTO_DELETE_TIMEOUT) == args.end();
+    return haBroker.replicateLevel(args) && !ignore;
+}
+
 void BrokerReplicator::startQueueReplicator(const boost::shared_ptr<Queue>& queue)
 {
     if (haBroker.replicateLevel(queue->getSettings()) == ALL) {
diff --git a/cpp/src/qpid/ha/BrokerReplicator.h b/cpp/src/qpid/ha/BrokerReplicator.h
index da9802268a..d2fd23e63d 100644
--- a/cpp/src/qpid/ha/BrokerReplicator.h
+++ b/cpp/src/qpid/ha/BrokerReplicator.h
@@ -88,7 +88,8 @@ class BrokerReplicator : public broker::Exchange
     void doResponseHaBroker(types::Variant::Map& values);
 
     QueueReplicatorPtr findQueueReplicator(const std::string& qname);
-    void  startQueueReplicator(const boost::shared_ptr<broker::Queue>&);
+    void startQueueReplicator(const boost::shared_ptr<broker::Queue>&);
+    bool isReplicated(const types::Variant::Map& args, bool autodelete, bool exclusive);
     void ready();
 
     LogPrefix logPrefix;