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 2010/08/02 11:51:11 UTC
svn commit: r981435 - in /qpid/trunk/qpid/cpp/src: qpid/broker/Broker.cpp
qpid/broker/QueueCleaner.cpp qpid/broker/QueueCleaner.h tests/QueueTest.cpp
tests/cluster_tests.py
Author: gsim
Date: Mon Aug 2 09:51:10 2010
New Revision: 981435
URL: http://svn.apache.org/viewvc?rev=981435&view=rev
Log:
Ensure that for clustered broker the queue cleaner is run on the correct thread.
Modified:
qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/QueueCleaner.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/QueueCleaner.h
qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp
qpid/trunk/qpid/cpp/src/tests/cluster_tests.py
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=981435&r1=981434&r2=981435&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp Mon Aug 2 09:51:10 2010
@@ -156,7 +156,7 @@ Broker::Broker(const Broker::Options& co
conf.replayFlushLimit*1024, // convert kb to bytes.
conf.replayHardLimit*1024),
*this),
- queueCleaner(queues, timer),
+ queueCleaner(queues, &timer),
queueEvents(poller,!conf.asyncQueueEvents),
recovery(true),
clusterUpdatee(false),
@@ -504,6 +504,7 @@ bool Broker::deferDeliveryImpl(const std
void Broker::setClusterTimer(std::auto_ptr<sys::Timer> t) {
clusterTimer = t;
+ queueCleaner.setTimer(clusterTimer.get());
}
const std::string Broker::TCP_TRANSPORT("tcp");
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/QueueCleaner.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/QueueCleaner.cpp?rev=981435&r1=981434&r2=981435&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/QueueCleaner.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/QueueCleaner.cpp Mon Aug 2 09:51:10 2010
@@ -26,17 +26,24 @@
namespace qpid {
namespace broker {
-QueueCleaner::QueueCleaner(QueueRegistry& q, sys::Timer& t) : queues(q), timer(t) {}
+QueueCleaner::QueueCleaner(QueueRegistry& q, sys::Timer* t) : queues(q), timer(t) {}
QueueCleaner::~QueueCleaner()
{
if (task) task->cancel();
}
+void QueueCleaner::setTimer(sys::Timer* t)
+{
+ timer = t;
+}
+
void QueueCleaner::start(qpid::sys::Duration p)
{
- task = new Task(*this, p);
- timer.add(task);
+ if (timer) {
+ task = new Task(*this, p);
+ timer->add(task);
+ }
}
QueueCleaner::Task::Task(QueueCleaner& p, qpid::sys::Duration d) : sys::TimerTask(d), parent(p) {}
@@ -66,7 +73,7 @@ void QueueCleaner::fired()
queues.eachQueue(collect);
std::for_each(copy.begin(), copy.end(), boost::bind(&Queue::purgeExpired, _1));
task->setupNextFire();
- timer.add(task);
+ if (timer) timer->add(task);
}
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/QueueCleaner.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/QueueCleaner.h?rev=981435&r1=981434&r2=981435&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/QueueCleaner.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/QueueCleaner.h Mon Aug 2 09:51:10 2010
@@ -35,8 +35,9 @@ class QueueRegistry;
class QueueCleaner
{
public:
- QPID_BROKER_EXTERN QueueCleaner(QueueRegistry& queues, sys::Timer& timer);
+ QPID_BROKER_EXTERN QueueCleaner(QueueRegistry& queues, sys::Timer* timer);
QPID_BROKER_EXTERN ~QueueCleaner();
+ QPID_BROKER_EXTERN void setTimer(sys::Timer* timer);
QPID_BROKER_EXTERN void start(qpid::sys::Duration period);
private:
class Task : public sys::TimerTask
@@ -50,7 +51,7 @@ class QueueCleaner
boost::intrusive_ptr<sys::TimerTask> task;
QueueRegistry& queues;
- sys::Timer& timer;
+ sys::Timer* timer;
void fired();
};
Modified: qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp?rev=981435&r1=981434&r2=981435&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp (original)
+++ qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp Mon Aug 2 09:51:10 2010
@@ -687,7 +687,7 @@ QPID_AUTO_TEST_CASE(testQueueCleaner) {
addMessagesToQueue(10, *queue, 200, 400);
BOOST_CHECK_EQUAL(queue->getMessageCount(), 10u);
- QueueCleaner cleaner(queues, timer);
+ QueueCleaner cleaner(queues, &timer);
cleaner.start(100 * qpid::sys::TIME_MSEC);
::usleep(300*1000);
BOOST_CHECK_EQUAL(queue->getMessageCount(), 5u);
Modified: qpid/trunk/qpid/cpp/src/tests/cluster_tests.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/cluster_tests.py?rev=981435&r1=981434&r2=981435&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/cluster_tests.py (original)
+++ qpid/trunk/qpid/cpp/src/tests/cluster_tests.py Mon Aug 2 09:51:10 2010
@@ -185,6 +185,35 @@ acl allow all all
broker1.ready()
broker2.ready()
+ def test_queue_cleaner(self):
+ """ Regression test to ensure that cleanup of expired messages works correctly """
+ cluster = self.cluster(2, args=["--queue-purge-interval", 3])
+
+ s0 = cluster[0].connect().session()
+ sender = s0.sender("my-lvq; {create: always, node:{x-declare:{arguments:{'qpid.last_value_queue':1}}}}")
+ #send 10 messages that will all expire and be cleaned up
+ for i in range(1, 10):
+ msg = Message("message-%s" % i)
+ msg.properties["qpid.LVQ_key"] = "a"
+ msg.ttl = 0.1
+ sender.send(msg)
+ #wait for queue cleaner to run
+ time.sleep(3)
+
+ #test all is ok by sending and receiving a message
+ msg = Message("non-expiring")
+ msg.properties["qpid.LVQ_key"] = "b"
+ sender.send(msg)
+ s0.connection.close()
+ s1 = cluster[1].connect().session()
+ m = s1.receiver("my-lvq", capacity=1).fetch(timeout=1)
+ s1.acknowledge()
+ self.assertEqual("non-expiring", m.content)
+ s1.connection.close()
+
+ for b in cluster: b.ready() # Make sure all brokers still running.
+
+
class LongTests(BrokerTest):
"""Tests that can run for a long time if -DDURATION=<minutes> is set"""
def duration(self):
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org