QPID-3087: Fixes to store interaction changes

* don't create queue on recovery
* ensure laternate exchange is set before creating store record for queue



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

6 files changed, 48 insertions, 25 deletions

diff --git a/qpid/cpp/src/qpid/broker/Broker.cpp b/qpid/cpp/src/qpid/broker/Broker.cpp
index 4e7e78c961..695943854d 100644
--- a/qpid/cpp/src/qpid/broker/Broker.cpp
+++ b/qpid/cpp/src/qpid/broker/Broker.cpp
@@ -790,13 +790,8 @@ std::pair<boost::shared_ptr<Queue>, bool> Broker::createQueue(
         if (!alternate) throw framing::NotFoundException(QPID_MSG("Alternate exchange does not exist: " << alternateExchange));
     }
 
-    std::pair<Queue::shared_ptr, bool> result = queues.declare(name, durable, autodelete, owner, arguments);
+    std::pair<Queue::shared_ptr, bool> result = queues.declare(name, durable, autodelete, owner, alternate, arguments);
     if (result.second) {
-        if (alternate) {
-            result.first->setAlternateExchange(alternate);
-            alternate->incAlternateUsers();
-        }
-
         //add default binding:
         result.first->bind(exchanges.getDefault(), name);
 
diff --git a/qpid/cpp/src/qpid/broker/Queue.cpp b/qpid/cpp/src/qpid/broker/Queue.cpp
index d18b0fcda3..09b8256b54 100644
--- a/qpid/cpp/src/qpid/broker/Queue.cpp
+++ b/qpid/cpp/src/qpid/broker/Queue.cpp
@@ -725,7 +725,7 @@ void Queue::create(const FieldTable& _settings)
     if (store) {
         store->create(*this, _settings);
     }
-    configure(_settings);
+    configureImpl(_settings);
 }
 
 
@@ -750,9 +750,14 @@ int getIntegerSetting(const qpid::framing::FieldTable& settings, const std::stri
     }
 }
 
-void Queue::configure(const FieldTable& _settings, bool recovering)
+void Queue::configure(const FieldTable& _settings)
 {
+    settings = _settings;
+    configureImpl(settings);
+}
 
+void Queue::configureImpl(const FieldTable& _settings)
+{
     eventMode = _settings.getAsInt(qpidQueueEventGeneration);
     if (eventMode && broker) {
         broker->getQueueEvents().observe(*this, eventMode == ENQUEUE_ONLY);
@@ -819,9 +824,6 @@ void Queue::configure(const FieldTable& _settings, bool recovering)
         mgmtObject->set_arguments(ManagementAgent::toMap(_settings));
     }
 
-    if ( isDurable() && ! getPersistenceId() && ! recovering )
-      store->create(*this, _settings);
-
     QueueFlowLimit::observe(*this, _settings);
 }
 
@@ -919,9 +921,10 @@ Queue::shared_ptr Queue::decode ( QueueRegistry& queues, Buffer& buffer, bool re
 {
     string name;
     buffer.getShortString(name);
-    std::pair<Queue::shared_ptr, bool> result = queues.declare(name, true);
-    buffer.get(result.first->settings);
-    result.first->configure(result.first->settings, recovering );
+    FieldTable settings;
+    buffer.get(settings);
+    boost::shared_ptr<Exchange> alternate;
+    std::pair<Queue::shared_ptr, bool> result = queues.declare(name, true, false, 0, alternate, settings, recovering);
     if (result.first->policy.get() && buffer.available() >= result.first->policy->encodedSize()) {
         buffer.get ( *(result.first->policy) );
     }
diff --git a/qpid/cpp/src/qpid/broker/Queue.h b/qpid/cpp/src/qpid/broker/Queue.h
index 331c9eaa4e..7597ec74ce 100644
--- a/qpid/cpp/src/qpid/broker/Queue.h
+++ b/qpid/cpp/src/qpid/broker/Queue.h
@@ -149,6 +149,7 @@ class Queue : public boost::enable_shared_from_this<Queue>,
     QueuedMessage getFront();
     void forcePersistent(QueuedMessage& msg);
     int getEventMode();
+    void configureImpl(const qpid::framing::FieldTable& settings);
 
     inline void mgntEnqStats(const boost::intrusive_ptr<Message>& msg)
     {
@@ -192,11 +193,17 @@ class Queue : public boost::enable_shared_from_this<Queue>,
 
     QPID_BROKER_EXTERN bool dispatch(Consumer::shared_ptr);
 
+    /**
+     * Used to configure a new queue and create a persistent record
+     * for it in store if required.
+     */
     void create(const qpid::framing::FieldTable& settings);
 
-    // "recovering" means we are doing a MessageStore recovery.
-    QPID_BROKER_EXTERN void configure(const qpid::framing::FieldTable& settings,
-                                      bool recovering = false);
+    /**
+     * Used to reconfigure a recovered queue (does not create
+     * persistent record in store).
+     */
+    QPID_BROKER_EXTERN void configure(const qpid::framing::FieldTable& settings);
     void destroyed();
     QPID_BROKER_EXTERN void bound(const std::string& exchange,
                                   const std::string& key,
diff --git a/qpid/cpp/src/qpid/broker/QueueRegistry.cpp b/qpid/cpp/src/qpid/broker/QueueRegistry.cpp
index c44b11e4bf..135a3543d9 100644
--- a/qpid/cpp/src/qpid/broker/QueueRegistry.cpp
+++ b/qpid/cpp/src/qpid/broker/QueueRegistry.cpp
@@ -21,6 +21,7 @@
 #include "qpid/broker/Queue.h"
 #include "qpid/broker/QueueRegistry.h"
 #include "qpid/broker/QueueEvents.h"
+#include "qpid/broker/Exchange.h"
 #include "qpid/log/Statement.h"
 #include <sstream>
 #include <assert.h>
@@ -37,7 +38,12 @@ QueueRegistry::~QueueRegistry(){}
 std::pair<Queue::shared_ptr, bool>
 QueueRegistry::declare(const string& declareName, bool durable, 
                        bool autoDelete, const OwnershipToken* owner,
-                       const qpid::framing::FieldTable& arguments)
+                       boost::shared_ptr<Exchange> alternate,
+                       const qpid::framing::FieldTable& arguments,
+                       bool recovering/*true if this declare is a
+                                        result of recovering queue
+                                        definition from persistente
+                                        record*/)
 {
     RWlock::ScopedWlock locker(lock);
     string name = declareName.empty() ? generateName() : declareName;
@@ -46,8 +52,17 @@ QueueRegistry::declare(const string& declareName, bool durable,
 
     if (i == queues.end()) {
         Queue::shared_ptr queue(new Queue(name, autoDelete, durable ? store : 0, owner, parent, broker));
-        //apply settings & create persistent record if required
-        queue->create(arguments);
+        if (alternate) {
+            queue->setAlternateExchange(alternate);//need to do this *before* create
+            alternate->incAlternateUsers();
+        }
+        if (!recovering) {
+            //apply settings & create persistent record if required
+            queue->create(arguments);
+        } else {
+            //i.e. recovering a queue for which we already have a persistent record
+            queue->configure(arguments);
+        }
         queues[name] = queue;
         if (lastNode) queue->setLastNodeFailure();
 
diff --git a/qpid/cpp/src/qpid/broker/QueueRegistry.h b/qpid/cpp/src/qpid/broker/QueueRegistry.h
index 90ee924ba4..8a32a64f05 100644
--- a/qpid/cpp/src/qpid/broker/QueueRegistry.h
+++ b/qpid/cpp/src/qpid/broker/QueueRegistry.h
@@ -35,6 +35,7 @@ namespace broker {
 
 class Queue;
 class QueueEvents;
+class Exchange;
 class OwnershipToken;
 class Broker;
 class MessageStore;
@@ -62,7 +63,9 @@ class QueueRegistry {
         bool durable = false,
         bool autodelete = false, 
         const OwnershipToken* owner = 0,
-        const qpid::framing::FieldTable& args = framing::FieldTable());
+        boost::shared_ptr<Exchange> alternateExchange = boost::shared_ptr<Exchange>(),
+        const qpid::framing::FieldTable& args = framing::FieldTable(),
+        bool recovering = false);
 
     /**
      * Destroy the named queue.
diff --git a/qpid/cpp/src/tests/QueueTest.cpp b/qpid/cpp/src/tests/QueueTest.cpp
index e4e9897195..fd30a98ac0 100644
--- a/qpid/cpp/src/tests/QueueTest.cpp
+++ b/qpid/cpp/src/tests/QueueTest.cpp
@@ -637,7 +637,7 @@ QPID_AUTO_TEST_CASE(testLVQRecover){
 
     Queue::shared_ptr queue1(new Queue("my-queue", true, &testStore));
     intrusive_ptr<Message> received;
-    queue1->configure(args);
+    queue1->create(args);
 
     intrusive_ptr<Message> msg1 = create_message("e", "A");
     intrusive_ptr<Message> msg2 = create_message("e", "A");
@@ -709,9 +709,9 @@ QPID_AUTO_TEST_CASE(testMultiQueueLastNode){
     args.setPersistLastNode();
 
     Queue::shared_ptr queue1(new Queue("queue1", true, &testStore ));
-    queue1->configure(args);
+    queue1->create(args);
     Queue::shared_ptr queue2(new Queue("queue2", true, &testStore ));
-    queue2->configure(args);
+    queue2->create(args);
 
     intrusive_ptr<Message> msg1 = create_message("e", "A");
 
@@ -797,7 +797,7 @@ not requeued to the store.
 
     Queue::shared_ptr queue1(new Queue("my-queue", true, &testStore));
     intrusive_ptr<Message> received;
-    queue1->configure(args);
+    queue1->create(args);
 
     // check requeue 1
     intrusive_ptr<Message> msg1 = create_message("e", "C");