Handle ttl in messages transfers received by the broker 7 added test for it

Moved Timer instance from DtxManager to Broker so it can be shared



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

1 files changed, 7 insertions, 1 deletions

diff --git a/cpp/src/qpid/broker/Queue.cpp b/cpp/src/qpid/broker/Queue.cpp
index ea5f7a0ba9..4382ac2e57 100644
--- a/cpp/src/qpid/broker/Queue.cpp
+++ b/cpp/src/qpid/broker/Queue.cpp
@@ -264,6 +264,12 @@ bool Queue::consumeNextMessage(QueuedMessage& m, Consumer::shared_ptr c)
             return false;
         } else {
             QueuedMessage msg = messages.front();
+            if (msg.payload->hasExpired()) {
+                QPID_LOG(debug, "Message expired from queue '" << name << "'");
+                popAndDequeue();
+                continue;
+            }
+
             if (!optimisticConsume && store && !msg.payload->isEnqueueComplete()) { 
                 QPID_LOG(debug, "Messages not ready to dispatch on queue '" << name << "'");
                 addListener(c);
@@ -294,7 +300,7 @@ bool Queue::browseNextMessage(QueuedMessage& m, Consumer::shared_ptr c)
 {
     QueuedMessage msg(this);
     while (seek(msg, c)) {
-        if (c->filter(msg.payload)) {
+        if (c->filter(msg.payload) && !msg.payload->hasExpired()) {
             if (c->accept(msg.payload)) {
                 //consumer wants the message
                 c->position = msg.position;