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@703521 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 25 insertions, 3 deletions

diff --git a/qpid/python/tests_0-10/message.py b/qpid/python/tests_0-10/message.py
index e4dc5566bd..38f87d8526 100644
--- a/qpid/python/tests_0-10/message.py
+++ b/qpid/python/tests_0-10/message.py
@@ -23,7 +23,7 @@ from qpid.datatypes import Message, RangedSet
 from qpid.session import SessionException
 
 from qpid.content import Content
-
+from time import sleep
 
 class MessageTests(TestBase010):
     """Tests for 'methods' on the amqp message 'class'"""
@@ -369,7 +369,7 @@ class MessageTests(TestBase010):
             session.message_transfer(message=Message(session.delivery_properties(routing_key="q"), "abcdefgh"))
 
         #each message is currently interpreted as requiring msg_size bytes of credit
-        msg_size = 21
+        msg_size = 27
 
         #set byte credit to finite amount (less than enough for all messages)
         session.message_flow(unit = session.credit_unit.byte, value = msg_size*5, destination = "c")
@@ -438,7 +438,7 @@ class MessageTests(TestBase010):
             session.message_transfer(message=Message(session.delivery_properties(routing_key="q"), "abcdefgh"))
 
         #each message is currently interpreted as requiring msg_size bytes of credit
-        msg_size = 19
+        msg_size = 27
 
         #set byte credit to finite amount (less than enough for all messages)
         session.message_flow(unit = session.credit_unit.byte, value = msg_size*5, destination = "c")
@@ -810,6 +810,28 @@ class MessageTests(TestBase010):
         msg = messages.get()
         assert msg.body == "test"
 
+    def test_ttl(self):
+        q = "test_ttl"
+        session = self.session
+
+        session.queue_declare(queue=q, exclusive=True, auto_delete=True)
+
+        dp = session.delivery_properties(routing_key=q, ttl=500)#expire in half a second
+        session.message_transfer(message=Message(dp, "first"))
+
+        dp = session.delivery_properties(routing_key=q, ttl=300000)#expire in fives minutes
+        session.message_transfer(message=Message(dp, "second"))
+
+        d = "msgs"
+        session.message_subscribe(queue=q, destination=d)
+        messages = session.incoming(d)
+        sleep(1)
+        session.message_flow(unit = session.credit_unit.message, value=2, destination=d)
+        session.message_flow(unit = session.credit_unit.byte, value=0xFFFFFFFF, destination=d)
+        assert messages.get(timeout=1).body == "second"
+        self.assertEmpty(messages)
+
+
     def assertDataEquals(self, session, msg, expected):
         self.assertEquals(expected, msg.body)