Ensure that for clustered broker the queue cleaner is run on the correct thread.

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@981435 13f79535-47bb-0310-9956-ffa450edef68

5 files changed, 46 insertions, 8 deletions

diff --git a/cpp/src/qpid/broker/Broker.cpp b/cpp/src/qpid/broker/Broker.cpp
index 1a8bed1be0..5e0ca90e69 100644
--- a/cpp/src/qpid/broker/Broker.cpp
+++ b/cpp/src/qpid/broker/Broker.cpp
@@ -156,7 +156,7 @@ Broker::Broker(const Broker::Options& conf) :
             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::string& ,
 
 void Broker::setClusterTimer(std::auto_ptr<sys::Timer> t) {
     clusterTimer = t;
+    queueCleaner.setTimer(clusterTimer.get());
 }
 
 const std::string Broker::TCP_TRANSPORT("tcp");
diff --git a/cpp/src/qpid/broker/QueueCleaner.cpp b/cpp/src/qpid/broker/QueueCleaner.cpp
index ed98468490..a462fe72ea 100644
--- a/cpp/src/qpid/broker/QueueCleaner.cpp
+++ b/cpp/src/qpid/broker/QueueCleaner.cpp
@@ -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);
 }
 
 
diff --git a/cpp/src/qpid/broker/QueueCleaner.h b/cpp/src/qpid/broker/QueueCleaner.h
index 11c2d180ac..8eae0afaaf 100644
--- a/cpp/src/qpid/broker/QueueCleaner.h
+++ b/cpp/src/qpid/broker/QueueCleaner.h
@@ -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();
 };
diff --git a/cpp/src/tests/QueueTest.cpp b/cpp/src/tests/QueueTest.cpp
index 80c69ac386..adc5c0bee7 100644
--- a/cpp/src/tests/QueueTest.cpp
+++ b/cpp/src/tests/QueueTest.cpp
@@ -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);
diff --git a/cpp/src/tests/cluster_tests.py b/cpp/src/tests/cluster_tests.py
index 06a8dcee50..57f3727a13 100755
--- a/cpp/src/tests/cluster_tests.py
+++ b/cpp/src/tests/cluster_tests.py
@@ -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):