yardstick/common/messaging/producer.py

   1 # Copyright (c) 2018 Intel Corporation
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #      http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14
  15 import abc
  16 import logging
  17 import os
  18
  19 from oslo_config import cfg
  20 import oslo_messaging
  21 import six
  22
  23 from yardstick.common import messaging
  24
  25
  26 LOG = logging.getLogger(__name__)
  27
  28
  29 @six.add_metaclass(abc.ABCMeta)
  30 class MessagingProducer(object):
  31     """Abstract class to implement a MQ producer
  32
  33     This abstract class allows a class implementing this interface to publish
  34     messages in a message queue.
  35     """
  36
  37     def __init__(self, topic, pid=os.getpid(), fanout=True):
  38         """Init function.
  39
  40         :param topic: (string) MQ exchange topic
  41         :param pid: (int) PID of the process implementing this MQ Notifier
  42         :param fanout: (bool) MQ clients may request that a copy of the message
  43                        be delivered to all servers listening on a topic by
  44                        setting fanout to ``True``, rather than just one of them
  45         :returns: `MessagingNotifier` class object
  46         """
  47         self._topic = topic
  48         self._pid = pid
  49         self._fanout = fanout
  50         self._transport = oslo_messaging.get_rpc_transport(
  51             cfg.CONF, url=messaging.TRANSPORT_URL)
  52         self._target = oslo_messaging.Target(topic=topic, fanout=fanout,
  53                                              server=messaging.SERVER)
  54         self._notifier = oslo_messaging.RPCClient(self._transport,
  55                                                   self._target)
  56
  57     def send_message(self, method, payload):
  58         """Send a cast message, that will invoke a method without blocking.
  59
  60         The cast() method is used to invoke an RPC method that does not return
  61         a value.  cast() RPC requests may be broadcast to all Servers listening
  62         on a given topic by setting the fanout Target property to ``True``.
  63
  64         :param methos: (string) method name, that must be implemented in the
  65                        consumer endpoints
  66         :param payload: (subclass `Payload`) payload content
  67         """
  68         self._notifier.cast({'pid': self._pid},
  69                             method,
  70                             **payload.obj_to_dict())