You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by gs...@apache.org on 2011/02/28 14:57:14 UTC
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
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