Merge "Update rmq topics to better reflect their purpose"

senlin/cmd/conductor.py

@@ -39,7 +39,7 @@ def main():
     from senlin.conductor import service as conductor
 
     profiler.setup('senlin-conductor', cfg.CONF.host)
-    srv = conductor.ConductorService(cfg.CONF.host, consts.ENGINE_TOPIC)
+    srv = conductor.ConductorService(cfg.CONF.host, consts.CONDUCTOR_TOPIC)
     launcher = service.launch(cfg.CONF, srv,
                               workers=cfg.CONF.conductor.workers,
                               restart_method='mutate')

senlin/cmd/engine.py

@@ -40,7 +40,7 @@ def main():
 
     profiler.setup('senlin-engine', cfg.CONF.host)
     srv = engine.EngineService(cfg.CONF.host,
-                               consts.DISPATCHER_TOPIC)
+                               consts.ENGINE_TOPIC)
     launcher = service.launch(cfg.CONF, srv,
                               workers=cfg.CONF.engine.workers,
                               restart_method='mutate')

senlin/common/consts.py

@@ -13,15 +13,15 @@
 from oslo_log import log as logging
 
 RPC_ATTRS = (
+    CONDUCTOR_TOPIC,
     ENGINE_TOPIC,
-    DISPATCHER_TOPIC,
     HEALTH_MANAGER_TOPIC,
     RPC_API_VERSION_BASE,
     RPC_API_VERSION,
 ) = (
+    'senlin-conductor',
     'senlin-engine',
-    'engine-dispatcher',
+    'senlin-health-manager',
-    'engine-health-mgr',
     '1.0',
     '1.1',
 )

senlin/conductor/service.py

@@ -87,7 +87,7 @@ class ConductorService(service.Service):
             self.service_name, host, topic,
             threads=CONF.conductor.threads
         )
-        self.dispatcher_topic = consts.DISPATCHER_TOPIC
+        self.dispatcher_topic = consts.ENGINE_TOPIC
         self.health_mgr_topic = consts.HEALTH_MANAGER_TOPIC
 
         # The following are initialized here and will be assigned in start()

senlin/engine/dispatcher.py

@@ -36,7 +36,7 @@ def notify(method, engine_id=None, **kwargs):
     :param method: remote method to call
     :param engine_id: dispatcher to notify; None implies broadcast
     """
-    client = messaging.get_rpc_client(consts.DISPATCHER_TOPIC, cfg.CONF.host)
+    client = messaging.get_rpc_client(consts.ENGINE_TOPIC, cfg.CONF.host)
 
     if engine_id:
         # Notify specific dispatcher identified by engine_id

senlin/rpc/client.py

@@ -32,7 +32,7 @@ class EngineClient(object):
 
     def __init__(self):
         serializer = object_base.VersionedObjectSerializer()
-        self._client = messaging.get_rpc_client(consts.ENGINE_TOPIC,
+        self._client = messaging.get_rpc_client(consts.CONDUCTOR_TOPIC,
                                                 cfg.CONF.host,
                                                 serializer=serializer)
 

senlin/tests/unit/api/shared.py

@@ -49,7 +49,7 @@ class ControllerTest(object):
         super(ControllerTest, self).__init__(*args, **kwargs)
 
         cfg.CONF.set_default('host', 'server.test')
-        self.topic = consts.ENGINE_TOPIC
+        self.topic = consts.CONDUCTOR_TOPIC
         self.api_version = '1.0'
         self.project = 'PROJ'
         self.mock_enforce = None

senlin/tests/unit/cmd/test_conductor.py

@@ -50,7 +50,7 @@ class TestConductor(base.SenlinTestCase):
         mock_profiler_setup.assert_called_once()
 
         mock_service.assert_called_once_with(
-            'hostname', consts.ENGINE_TOPIC
+            'hostname', consts.CONDUCTOR_TOPIC
         )
 
         mock_launch.assert_called_once_with(

senlin/tests/unit/cmd/test_engine.py

@@ -50,7 +50,7 @@ class TestEngine(base.SenlinTestCase):
         mock_profiler_setup.assert_called_once()
 
         mock_service.assert_called_once_with(
-            'hostname', consts.DISPATCHER_TOPIC
+            'hostname', consts.ENGINE_TOPIC
         )
 
         mock_launch.assert_called_once_with(

senlin/tests/unit/engine/test_service.py

@@ -65,7 +65,7 @@ class TestEngine(base.SenlinTestCase):
 
         self.service_id = '4db0a14c-dc10-4131-8ed6-7573987ce9b0'
         self.tg = mock.Mock()
-        self.topic = consts.DISPATCHER_TOPIC
+        self.topic = consts.ENGINE_TOPIC
 
         self.tg = mock.Mock()
         self.svc = service.EngineService('HOST', self.topic)
@@ -160,7 +160,7 @@ class TestEngine(base.SenlinTestCase):
 
         dispatcher.notify('METHOD')
 
-        mock_rpc.assert_called_once_with(consts.DISPATCHER_TOPIC, 'HOSTNAME')
+        mock_rpc.assert_called_once_with(consts.ENGINE_TOPIC, 'HOSTNAME')
         mock_client = mock_rpc.return_value
         mock_client.prepare.assert_called_once_with(fanout=True)
 
@@ -178,7 +178,7 @@ class TestEngine(base.SenlinTestCase):
         result = dispatcher.notify('METHOD', 'FAKE_ENGINE')
 
         self.assertTrue(result)
-        mock_rpc.assert_called_once_with(consts.DISPATCHER_TOPIC, 'HOSTNAME')
+        mock_rpc.assert_called_once_with(consts.ENGINE_TOPIC, 'HOSTNAME')
         mock_client = mock_rpc.return_value
         mock_client.prepare.assert_called_once_with(server='FAKE_ENGINE')
 
@@ -196,7 +196,7 @@ class TestEngine(base.SenlinTestCase):
         result = dispatcher.notify('METHOD')
 
         self.assertFalse(result)
-        mock_rpc.assert_called_once_with(consts.DISPATCHER_TOPIC, 'HOSTNAME')
+        mock_rpc.assert_called_once_with(consts.ENGINE_TOPIC, 'HOSTNAME')
         mock_client.prepare.assert_called_once_with(fanout=True)
 
         mock_context.cast.assert_called_once_with(mock.ANY, 'METHOD')