diff options
-rw-r--r-- | openstack/common/rpc/impl_zmq.py | 1 | ||||
-rw-r--r-- | tests/unit/rpc/test_zmq.py | 21 |
2 files changed, 22 insertions, 0 deletions
diff --git a/openstack/common/rpc/impl_zmq.py b/openstack/common/rpc/impl_zmq.py index 33d5cb6..a5ddccf 100644 --- a/openstack/common/rpc/impl_zmq.py +++ b/openstack/common/rpc/impl_zmq.py @@ -383,6 +383,7 @@ class ZmqBaseReactor(ConsumerBase): LOG.info(_("In reactor registered")) def consume_in_thread(self): + @excutils.forever_retry_uncaught_exceptions def _consume(sock): LOG.info(_("Consuming socket")) while True: diff --git a/tests/unit/rpc/test_zmq.py b/tests/unit/rpc/test_zmq.py index 11d3fcd..888dbb0 100644 --- a/tests/unit/rpc/test_zmq.py +++ b/tests/unit/rpc/test_zmq.py @@ -112,6 +112,27 @@ class _RpcZmqBaseTestCase(common.BaseRpcTestCase): finally: self.topic_nested = tmp_topic + def test_service_consume_thread_unexpected_exceptions(self): + def my_TopicConsumer_consume(myself, *args, **kwargs): + self.consume_calls += 1 + # see if it can sustain three failures + if self.consume_calls < 3: + raise Exception('unexpected exception') + else: + self.orig_ZmqReactor_consume(myself, *args, **kwargs) + + self.consume_calls = 0 + self.orig_ZmqReactor_consume = impl_zmq.ZmqReactor.consume + self.stubs.Set(impl_zmq.ZmqReactor, 'consume', + my_TopicConsumer_consume) + + value = 42 + result = self.rpc.call(FLAGS, self.context, self.topic, + {"method": "echo", + "args": {"value": value}}) + self.assertEqual(value, result) + self.stubs.UnsetAll() + class RpcZmqBaseTopicTestCase(_RpcZmqBaseTestCase): """Base topic RPC ZMQ test case. |