Added initialize_service_hook for rpc.Service.

This fixes the collector part of the bug 1075463.

Declaring a consumer topic on the same rpc connection after the consume
thread has started would result the eventlet raise RuntimeError
exception. So all the declaring work should be done before calling
rpc.conn.consume_in_thread().

If the manager of a rpc.Service wants to declare topic consumers other
than the default ones created by rpc.Service.start(), it could define
the following hook in the manager class:

  def initialize_service_hook(self, service):
    #
    # Do initialization work after rpc connection is created and before
    # starting consuming thread.
    # Params:
    #   service: handle to the rpc.Service instance.

Change-Id: I80001c32ee4e51e394fed827c91ad5e1eb0f94dc

2 files changed, 23 insertions, 0 deletions

diff --git a/openstack/common/rpc/service.py b/openstack/common/rpc/service.py
index e572956..6b56ebb 100644
--- a/openstack/common/rpc/service.py
+++ b/openstack/common/rpc/service.py
@@ -57,6 +57,11 @@ class Service(service.Service):
 
         self.conn.create_consumer(self.topic, dispatcher, fanout=True)
 
+        # Hook to allow the manager to do other initializations after
+        # the rpc connection is created.
+        if callable(getattr(self.manager, 'initialize_service_hook', None)):
+            self.manager.initialize_service_hook(self)
+
         # Consume from all consumers in a thread
         self.conn.consume_in_thread()
 
diff --git a/tests/unit/rpc/test_service.py b/tests/unit/rpc/test_service.py
index 7dc20d7..9293d3e 100644
--- a/tests/unit/rpc/test_service.py
+++ b/tests/unit/rpc/test_service.py
@@ -30,6 +30,18 @@ class FakeService(service.Service):
         return self.method_result
 
 
+class FakeHookService(FakeService):
+    def __init__(self, host, topic):
+        super(FakeService, self).__init__(host, topic)
+        self.hooked = False
+
+    def initialize_service_hook(self, service):
+        self.hooked = True
+
+    def test_hook(self):
+        return self.hooked
+
+
 class RpcServiceManagerTestCase(utils.BaseTestCase):
     """Test cases for Services"""
     def setUp(self):
@@ -45,3 +57,9 @@ class RpcServiceManagerTestCase(utils.BaseTestCase):
         serv.start()
         self.assertEqual(serv.test_method(), 'manager')
         serv.stop()
+
+    def test_hook_default(self):
+        serv = FakeHookService('test-host', 'test-topic')
+        serv.start()
+        self.assertTrue(serv.test_hook())
+        serv.stop()