Periodically purge expired messages from queues

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

1 files changed, 40 insertions, 0 deletions

diff --git a/qpid/cpp/src/tests/ClientSessionTest.cpp b/qpid/cpp/src/tests/ClientSessionTest.cpp
index 85497ace5d..440605a2e4 100644
--- a/qpid/cpp/src/tests/ClientSessionTest.cpp
+++ b/qpid/cpp/src/tests/ClientSessionTest.cpp
@@ -42,6 +42,7 @@ using namespace qpid;
 using qpid::sys::Monitor;
 using qpid::sys::Thread;
 using qpid::sys::TIME_SEC;
+using qpid::broker::Broker;
 using std::string;
 using std::cout;
 using std::endl;
@@ -94,6 +95,8 @@ struct SimpleListener : public MessageListener
 
 struct ClientSessionFixture : public ProxySessionFixture
 {
+    ClientSessionFixture(Broker::Options opts = Broker::Options()) : ProxySessionFixture(opts) {}
+
     void declareSubscribe(const string& q="my-queue",
                           const string& dest="my-dest")
     {
@@ -282,6 +285,43 @@ QPID_AUTO_TEST_CASE(testOpenFailure) {
     BOOST_CHECK(!c.isOpen());
 }
 
+QPID_AUTO_TEST_CASE(testPeriodicExpiration) {
+    Broker::Options opts;
+    opts.queueCleanInterval = 1;
+    ClientSessionFixture fix(opts);
+    fix.session.queueDeclare(arg::queue="my-queue", arg::exclusive=true, arg::autoDelete=true);
+
+    for (uint i = 0; i < 10; i++) {        
+        Message m((boost::format("Message_%1%") % (i+1)).str(), "my-queue");        
+        if (i % 2) m.getDeliveryProperties().setTtl(500);
+        fix.session.messageTransfer(arg::content=m);
+    }
+
+    BOOST_CHECK_EQUAL(fix.session.queueQuery(string("my-queue")).getMessageCount(), 10u);
+    sleep(2);
+    BOOST_CHECK_EQUAL(fix.session.queueQuery(string("my-queue")).getMessageCount(), 5u);
+}
+
+QPID_AUTO_TEST_CASE(testExpirationOnPop) {
+    ClientSessionFixture fix;
+    fix.session.queueDeclare(arg::queue="my-queue", arg::exclusive=true, arg::autoDelete=true);
+
+    for (uint i = 0; i < 10; i++) {        
+        Message m((boost::format("Message_%1%") % (i+1)).str(), "my-queue");        
+        if (i % 2) m.getDeliveryProperties().setTtl(200);
+        fix.session.messageTransfer(arg::content=m);
+    }
+
+    ::usleep(300* 1000);
+
+    for (uint i = 0; i < 10; i++) {        
+        if (i % 2) continue;
+        Message m;
+        BOOST_CHECK(fix.subs.get(m, "my-queue", TIME_SEC));
+        BOOST_CHECK_EQUAL((boost::format("Message_%1%") % (i+1)).str(), m.getData());
+    }
+}
+
 QPID_AUTO_TEST_SUITE_END()