Author: kpvdr
Date: Thu Oct 29 20:41:22 2009
New Revision: 831082
URL: http://svn.apache.org/viewvc?rev=831082&view=rev
Log:
Fixed problem of queue alternate-exchange property not being persisted on
persistent queues, and on recovery this property is lost. No tests exist as yet
for this.
Modified:
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/RecoveryManagerImpl.cpp
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=831082&r1=831081&r2=831082&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp Thu Oct 29 20:41:22 2009
@@ -917,6 +917,7 @@
void Queue::encode(Buffer& buffer) const
{
buffer.putShortString(name);
+ buffer.putShortString(alternateExchange.get() ?
alternateExchange->getName() : std::string(""));
buffer.put(settings);
if (policy.get()) {
buffer.put(*policy);
@@ -925,7 +926,9 @@
uint32_t Queue::encodedSize() const
{
- return name.size() + 1/*short string size octet*/ + settings.encodedSize()
+ return name.size() + 1/*short string size octet*/
+ + (alternateExchange.get() ? alternateExchange->getName().size() : 0)
+ 1 /* short string */
+ + settings.encodedSize()
+ (policy.get() ? (*policy).encodedSize() : 0);
}
@@ -933,7 +936,10 @@
{
string name;
buffer.getShortString(name);
+ string altExch;
+ buffer.getShortString(altExch);
std::pair<Queue::shared_ptr, bool> result = queues.declare(name, true);
+ result.first->alternateExchangeName.assign(altExch);
buffer.get(result.first->settings);
result.first->configure(result.first->settings, recovering );
if (result.first->policy.get() && buffer.available() >=
result.first->policy->encodedSize()) {
@@ -1048,8 +1054,17 @@
eventMgr = &mgr;
}
-void Queue::recoveryComplete()
+void Queue::recoveryComplete(ExchangeRegistry& exchanges)
{
+ // set the alternate exchange
+ if (!alternateExchangeName.empty()) {
+ try {
+ Exchange::shared_ptr ae = exchanges.get(alternateExchangeName);
+ setAlternateExchange(ae);
+ } catch (const NotFoundException&) {
+ QPID_LOG(warning, "Could not set alternate exchange \"" <<
alternateExchangeName << "\" on queue \"" << name << "\": exchange does not
exist.");
+ }
+ }
//process any pending dequeues
for_each(pendingDequeues.begin(), pendingDequeues.end(),
boost::bind(&Queue::dequeue, this, (TransactionContext*) 0, _1));
pendingDequeues.clear();
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=831082&r1=831081&r2=831082&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h Thu Oct 29 20:41:22 2009
@@ -97,6 +97,7 @@
std::auto_ptr<QueuePolicy> policy;
bool policyExceeded;
QueueBindings bindings;
+ std::string alternateExchangeName;
boost::shared_ptr<Exchange> alternateExchange;
framing::SequenceNumber sequence;
qmf::org::apache::qpid::broker::Queue* mgmtObject;
@@ -332,7 +333,7 @@
/**
* Notify queue that recovery has completed.
*/
- void recoveryComplete();
+ void recoveryComplete(ExchangeRegistry& exchanges);
// For cluster update
QueueListeners& getListeners();
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/RecoveryManagerImpl.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/RecoveryManagerImpl.cpp?rev=831082&r1=831081&r2=831082&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/RecoveryManagerImpl.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/RecoveryManagerImpl.cpp Thu Oct 29
20:41:22 2009
@@ -155,8 +155,8 @@
void RecoveryManagerImpl::recoveryComplete()
{
- //notify all queues
- queues.eachQueue(boost::bind(&Queue::recoveryComplete, _1));
+ //notify all queues and exchanges
+ queues.eachQueue(boost::bind(&Queue::recoveryComplete, _1,
boost::ref(exchanges)));
exchanges.eachExchange(boost::bind(&Exchange::recoveryComplete, _1,
boost::ref(exchanges)));
}
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscr...@qpid.apache.org
