qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From g...@apache.org
Subject svn commit: r1075331 - in /qpid/trunk/qpid/cpp/src: qpid/broker/Broker.cpp qpid/broker/Queue.cpp qpid/broker/Queue.h qpid/broker/QueueRegistry.cpp qpid/broker/QueueRegistry.h tests/QueueTest.cpp
Date Mon, 28 Feb 2011 13:57:14 GMT
Author: gsim
Date: Mon Feb 28 13:57:13 2011
New Revision: 1075331

URL: http://svn.apache.org/viewvc?rev=1075331&view=rev
Log:
QPID-3087: Fixes to store interaction changes

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


Modified:
    qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp
    qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp
    qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h
    qpid/trunk/qpid/cpp/src/qpid/broker/QueueRegistry.cpp
    qpid/trunk/qpid/cpp/src/qpid/broker/QueueRegistry.h
    qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp

Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp?rev=1075331&r1=1075330&r2=1075331&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp Mon Feb 28 13:57:13 2011
@@ -790,13 +790,8 @@ std::pair<boost::shared_ptr<Queue>, bool
         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);
 

Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp?rev=1075331&r1=1075330&r2=1075331&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp Mon Feb 28 13:57:13 2011
@@ -725,7 +725,7 @@ void Queue::create(const FieldTable& _se
     if (store) {
         store->create(*this, _settings);
     }
-    configure(_settings);
+    configureImpl(_settings);
 }
 
 
@@ -750,9 +750,14 @@ int getIntegerSetting(const qpid::framin
     }
 }
 
-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& 
         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 ( QueueR
 {
     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) );
     }

Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h?rev=1075331&r1=1075330&r2=1075331&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h Mon Feb 28 13:57:13 2011
@@ -149,6 +149,7 @@ class Queue : public boost::enable_share
     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_share
 
     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,

Modified: qpid/trunk/qpid/cpp/src/qpid/broker/QueueRegistry.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/QueueRegistry.cpp?rev=1075331&r1=1075330&r2=1075331&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/QueueRegistry.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/QueueRegistry.cpp Mon Feb 28 13:57:13 2011
@@ -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& dec
 
     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();
 

Modified: qpid/trunk/qpid/cpp/src/qpid/broker/QueueRegistry.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/QueueRegistry.h?rev=1075331&r1=1075330&r2=1075331&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/QueueRegistry.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/QueueRegistry.h Mon Feb 28 13:57:13 2011
@@ -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.

Modified: qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp?rev=1075331&r1=1075330&r2=1075331&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp (original)
+++ qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp Mon Feb 28 13:57:13 2011
@@ -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(testMultiQueueLastNo
     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");



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


Mime
View raw message