Define self.client in MessagingCore

Currently self.client is referenced within MessagingCore,
but no definition is made in its constructor. Additionally
self.client is defined in children classes of MessagingCore.
This patchset defines self.client in the constructor of
MessagingCore and removes the redefinition in its children.

-self.client lazily loaded

Co-Authored-By: v-francoise <Vincent.FRANCOISE@b-com.com>
Change-Id: I14525a175bf1ebde3d2636024ad2f2219c79d6e1
Closes-Bug: #1521636

watcher/common/messaging/messaging_core.py

@@ -16,58 +16,100 @@
 
 from oslo_config import cfg
 from oslo_log import log
-from watcher.common.messaging.events.event_dispatcher import \
-    EventDispatcher
-from watcher.common.messaging.messaging_handler import \
-    MessagingHandler
-from watcher.common.rpc import RequestContextSerializer
+import oslo_messaging as om
 
-from watcher.objects.base import WatcherObjectSerializer
+from watcher.common.messaging.events import event_dispatcher as dispatcher
+from watcher.common.messaging import messaging_handler
+from watcher.common import rpc
+
+from watcher.objects import base
 
 LOG = log.getLogger(__name__)
 CONF = cfg.CONF
 
 
-class MessagingCore(EventDispatcher):
+class MessagingCore(dispatcher.EventDispatcher):
 
     API_VERSION = '1.0'
 
-    def __init__(self, publisher_id, topic_control, topic_status,
+    def __init__(self, publisher_id, conductor_topic, status_topic,
                  api_version=API_VERSION):
         super(MessagingCore, self).__init__()
-        self.serializer = RequestContextSerializer(WatcherObjectSerializer())
+        self.serializer = rpc.RequestContextSerializer(
+            base.WatcherObjectSerializer())
         self.publisher_id = publisher_id
         self.api_version = api_version
-        self.topic_control = self.build_topic(topic_control)
-        self.topic_status = self.build_topic(topic_status)
 
-    def build_topic(self, topic_name):
-        return MessagingHandler(self.publisher_id, topic_name, self,
-                                self.api_version, self.serializer)
+        self.conductor_topic = conductor_topic
+        self.status_topic = status_topic
+        self.conductor_topic_handler = self.build_topic_handler(
+            conductor_topic)
+        self.status_topic_handler = self.build_topic_handler(status_topic)
+
+        self._conductor_client = None
+        self._status_client = None
+
+    @property
+    def conductor_client(self):
+        if self._conductor_client is None:
+            transport = om.get_transport(CONF)
+            target = om.Target(
+                topic=self.conductor_topic,
+                version=self.API_VERSION,
+            )
+            self._conductor_client = om.RPCClient(
+                transport, target, serializer=self.serializer)
+        return self._conductor_client
+
+    @conductor_client.setter
+    def conductor_client(self, c):
+        self.conductor_client = c
+
+    @property
+    def status_client(self):
+        if self._status_client is None:
+            transport = om.get_transport(CONF)
+            target = om.Target(
+                topic=self.status_topic,
+                version=self.API_VERSION,
+            )
+            self._status_client = om.RPCClient(
+                transport, target, serializer=self.serializer)
+        return self._status_client
+
+    @status_client.setter
+    def status_client(self, c):
+        self.status_client = c
+
+    def build_topic_handler(self, topic_name):
+        return messaging_handler.MessagingHandler(
+            self.publisher_id, topic_name, self,
+            self.api_version, self.serializer)
 
     def connect(self):
         LOG.debug("Connecting to '%s' (%s)",
                   CONF.transport_url, CONF.rpc_backend)
-        self.topic_control.start()
-        self.topic_status.start()
+        self.conductor_topic_handler.start()
+        self.status_topic_handler.start()
 
     def disconnect(self):
         LOG.debug("Disconnecting from '%s' (%s)",
                   CONF.transport_url, CONF.rpc_backend)
-        self.topic_control.stop()
-        self.topic_status.stop()
+        self.conductor_topic_handler.stop()
+        self.status_topic_handler.stop()
 
     def publish_control(self, event, payload):
-        return self.topic_control.publish_event(event, payload)
+        return self.conductor_topic_handler.publish_event(event, payload)
 
     def publish_status(self, event, payload, request_id=None):
-        return self.topic_status.publish_event(event, payload, request_id)
+        return self.status_topic_handler.publish_event(
+            event, payload, request_id)
 
     def get_version(self):
         return self.api_version
 
     def check_api_version(self, context):
-        api_manager_version = self.client.call(
+        api_manager_version = self.conductor_client.call(
             context.to_dict(), 'check_api_version',
             api_version=self.api_version)
         return api_manager_version

watcher/common/messaging/messaging_handler.py

@@ -38,11 +38,11 @@ CONF = cfg.CONF
 
 class MessagingHandler(threading.Thread):
 
-    def __init__(self, publisher_id, topic_watcher, endpoint, version,
+    def __init__(self, publisher_id, topic_name, endpoint, version,
                  serializer=None):
         super(MessagingHandler, self).__init__()
         self.publisher_id = publisher_id
-        self.topic_watcher = topic_watcher
+        self.topic_name = topic_name
         self.__endpoints = []
         self.__serializer = serializer
         self.__version = version
@@ -72,7 +72,7 @@ class MessagingHandler(threading.Thread):
         return om.Notifier(
             self.__transport,
             publisher_id=self.publisher_id,
-            topic=self.topic_watcher,
+            topic=self.topic_name,
             serializer=serializer
         )
 
@@ -87,7 +87,7 @@ class MessagingHandler(threading.Thread):
             self.__notifier = self.build_notifier()
             if len(self.__endpoints):
                 target = om.Target(
-                    topic=self.topic_watcher,
+                    topic=self.topic_name,
                     # For compatibility, we can override it with 'host' opt
                     server=CONF.host or socket.getfqdn(),
                     version=self.__version,
@@ -101,7 +101,7 @@ class MessagingHandler(threading.Thread):
             LOG.error(_LE("Messaging configuration error"))
 
     def run(self):
-        LOG.debug("configure MessagingHandler for %s" % self.topic_watcher)
+        LOG.debug("configure MessagingHandler for %s" % self.topic_name)
         self._configure()
         if len(self.__endpoints) > 0:
             LOG.debug("Starting up server")

watcher/common/messaging/notification_handler.py

@@ -18,17 +18,16 @@ import eventlet
 from oslo_log import log
 import oslo_messaging as messaging
 
-from watcher.common.messaging.utils.observable import \
-    Observable
+from watcher.common.messaging.utils import observable
 
 
 eventlet.monkey_patch()
 LOG = log.getLogger(__name__)
 
 
-class NotificationHandler(Observable):
+class NotificationHandler(observable.Observable):
     def __init__(self, publisher_id):
-        Observable.__init__(self)
+        super(NotificationHandler, self).__init__()
         self.publisher_id = publisher_id
 
     def info(self, ctx, publisher_id, event_type, payload, metadata):

watcher/common/messaging/utils/observable.py

@@ -14,19 +14,14 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from oslo_log import log
-
-from watcher.common.messaging.utils.synchronization import \
-    Synchronization
-
-LOG = log.getLogger(__name__)
+from watcher.common.messaging.utils import synchronization
 
 
-class Observable(Synchronization):
+class Observable(synchronization.Synchronization):
     def __init__(self):
+        super(Observable, self).__init__()
         self.__observers = []
         self.changed = 0
-        Synchronization.__init__(self)
 
     def set_changed(self):
         self.changed = 1