Fix AMQPListener for polling with timeout

On timeout, poll() should return None, not raise an exception (Timeout).

Add also an unit test.

Change-Id: I1ed5ae2f093841111f0b691dddb879c16d218b73

3 files changed, 21 insertions, 1 deletions

diff --git a/oslo/messaging/_cmd/zmq_receiver.py b/oslo/messaging/_cmd/zmq_receiver.py
index ae0713d..ae0713d 100755..100644
--- a/oslo/messaging/_cmd/zmq_receiver.py
+++ b/oslo/messaging/_cmd/zmq_receiver.py
diff --git a/oslo/messaging/_drivers/amqpdriver.py b/oslo/messaging/_drivers/amqpdriver.py
index 4398363..c5b2378 100644
--- a/oslo/messaging/_drivers/amqpdriver.py
+++ b/oslo/messaging/_drivers/amqpdriver.py
@@ -116,7 +116,10 @@ class AMQPListener(base.Listener):
                 timeout = deadline - time.time()
                 if timeout < 0:
                     return None
-                self.conn.consume(limit=1, timeout=timeout)
+                try:
+                    self.conn.consume(limit=1, timeout=timeout)
+                except rpc_common.Timeout:
+                    return None
             else:
                 self.conn.consume(limit=1)
 
diff --git a/tests/drivers/test_impl_rabbit.py b/tests/drivers/test_impl_rabbit.py
index f665b44..06f2af4 100644
--- a/tests/drivers/test_impl_rabbit.py
+++ b/tests/drivers/test_impl_rabbit.py
@@ -265,6 +265,23 @@ class TestSendReceive(test_utils.BaseTestCase):
 TestSendReceive.generate_scenarios()
 
 
+class TestPollAsync(test_utils.BaseTestCase):
+
+    def setUp(self):
+        super(TestPollAsync, self).setUp()
+        self.messaging_conf.transport_driver = 'rabbit'
+        self.messaging_conf.in_memory = True
+
+    def test_poll_timeout(self):
+        transport = messaging.get_transport(self.conf)
+        self.addCleanup(transport.cleanup)
+        driver = transport._driver
+        target = messaging.Target(topic='testtopic')
+        listener = driver.listen(target)
+        received = listener.poll(timeout=0.050)
+        self.assertIsNone(received)
+
+
 class TestRacyWaitForReply(test_utils.BaseTestCase):
 
     def setUp(self):