Author: cctrieloff
Date: Fri Jul 31 16:52:59 2009
New Revision: 799658
URL: http://svn.apache.org/viewvc?rev=799658&view=rev
Log:
handle fail setting last-node-standing with unit test, still needs system test
Modified:
qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp
qpid/trunk/qpid/cpp/src/tests/QueueTest.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=799658&r1=799657&r2=799658&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp Fri Jul 31 16:52:59 2009
@@ -677,21 +677,25 @@
{
if (persistLastNode){
Mutex::ScopedLock locker(messageLock);
- for ( Messages::iterator i = messages.begin(); i != messages.end(); ++i
) {
- if (lastValueQueue) checkLvqReplace(*i);
- // don't force a message twice to disk.
- if(!i->payload->isStoredOnQueue(shared_from_this())) {
- i->payload->forcePersistent();
- if (i->payload->isForcedPersistent() ){
- enqueue(0, i->payload);
+ try {
+ for ( Messages::iterator i = messages.begin(); i != messages.end();
++i ) {
+ if (lastValueQueue) checkLvqReplace(*i);
+ // don't force a message twice to disk.
+ if(!i->payload->isStoredOnQueue(shared_from_this())) {
+ i->payload->forcePersistent();
+ if (i->payload->isForcedPersistent() ){
+ enqueue(0, i->payload);
+ }
}
- }
- }
+ }
+ } catch (const std::exception& e) {
+ // Could not go into last node standing (for example journal not
large enough)
+ QPID_LOG(error, "Unable to fail to last node standing for queue: "
<< name << " : " << e.what());
+ }
inLastNodeFailure = true;
}
}
-
// return true if store exists,
bool Queue::enqueue(TransactionContext* ctxt, boost::intrusive_ptr<Message>
msg)
{
Modified: qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp?rev=799658&r1=799657&r2=799658&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp (original)
+++ qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp Fri Jul 31 16:52:59 2009
@@ -19,6 +19,7 @@
*
*/
#include "unit_test.h"
+#include "test_tools.h"
#include "qpid/Exception.h"
#include "qpid/broker/Broker.h"
#include "qpid/broker/Queue.h"
@@ -275,11 +276,13 @@
uint enqCnt;
uint deqCnt;
+ bool error;
virtual void dequeue(TransactionContext*,
const boost::intrusive_ptr<PersistableMessage>& /*msg*/,
const PersistableQueue& /*queue*/)
{
+ if (error) throw Exception("Dequeue error test");
deqCnt++;
}
@@ -287,10 +290,16 @@
const boost::intrusive_ptr<PersistableMessage>& /*msg*/,
const PersistableQueue& /* queue */)
{
+ if (error) throw Exception("Enqueue error test");
enqCnt++;
}
- TestMessageStoreOC() : NullMessageStore(),enqCnt(0),deqCnt(0) {}
+ void createError()
+ {
+ error=true;
+ }
+
+ TestMessageStoreOC() : NullMessageStore(),enqCnt(0),deqCnt(0),error(false)
{}
~TestMessageStoreOC(){}
};
@@ -689,6 +698,30 @@
}
-QPID_AUTO_TEST_SUITE_END()
+QPID_AUTO_TEST_CASE(testLastNodeJournalError){
+/*
+simulate store excption going into last node standing
+
+*/
+ TestMessageStoreOC testStore;
+ client::QueueOptions args;
+ // set queue mode
+ args.setPersistLastNode();
+
+ Queue::shared_ptr queue1(new Queue("my-queue", true, &testStore));
+ intrusive_ptr<Message> received;
+ queue1->configure(args);
+
+ // check requeue 1
+ intrusive_ptr<Message> msg1 = create_message("e", "C");
+
+ queue1->deliver(msg1);
+ testStore.createError();
+
+ ScopedSuppressLogging sl; // Suppress messages for expected errors.
+ queue1->setLastNodeFailure();
+ BOOST_CHECK_EQUAL(testStore.enqCnt, 0u);
+
+}QPID_AUTO_TEST_SUITE_END()
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscr...@qpid.apache.org
