2f529a62bf
In Mitaka oslo messaging introduced the transport_url specific for notifications under the oslo_messaging_notifications section, but some projects still use the default transport_url defined in the DEFAULT section. Use messaging notifications transport instead of default. This patch proposes short-circuiting notification methods as much as possible to optimize code execution when Karbor has no notification transport mechanism configured. The NOTIFIER only be inited only if notifications are actually enabled. Change-Id: Id9419ddd801bf9d8ac13c4a73a91e39fd763cb2b Closes-Bug: #1719786
145 lines
4.2 KiB
Python
145 lines
4.2 KiB
Python
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
# not use this file except in compliance with the License. You may obtain
|
|
# a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
# License for the specific language governing permissions and limitations
|
|
# under the License.
|
|
|
|
__all__ = [
|
|
'init',
|
|
'cleanup',
|
|
'set_defaults',
|
|
'add_extra_exmods',
|
|
'clear_extra_exmods',
|
|
'get_allowed_exmods',
|
|
'RequestContextSerializer',
|
|
'get_client',
|
|
'get_server',
|
|
'get_notifier',
|
|
]
|
|
|
|
from oslo_config import cfg
|
|
import oslo_messaging as messaging
|
|
from oslo_messaging.rpc import dispatcher
|
|
|
|
import karbor.context
|
|
import karbor.exception
|
|
from karbor import utils
|
|
|
|
CONF = cfg.CONF
|
|
TRANSPORT = None
|
|
NOTIFICATION_TRANSPORT = None
|
|
NOTIFIER = None
|
|
|
|
ALLOWED_EXMODS = [
|
|
karbor.exception.__name__,
|
|
]
|
|
EXTRA_EXMODS = []
|
|
|
|
|
|
def init(conf):
|
|
|
|
global TRANSPORT, NOTIFICATION_TRANSPORT, NOTIFIER
|
|
exmods = get_allowed_exmods()
|
|
TRANSPORT = messaging.get_rpc_transport(conf,
|
|
allowed_remote_exmods=exmods)
|
|
NOTIFICATION_TRANSPORT = messaging.get_notification_transport(
|
|
conf,
|
|
allowed_remote_exmods=exmods)
|
|
|
|
# get_notification_transport has loaded oslo_messaging_notifications config
|
|
# group, so we can now check if notifications are actually enabled.
|
|
if utils.notifications_enabled(conf):
|
|
json_serializer = messaging.JsonPayloadSerializer()
|
|
serializer = RequestContextSerializer(json_serializer)
|
|
NOTIFIER = messaging.Notifier(NOTIFICATION_TRANSPORT,
|
|
serializer=serializer)
|
|
else:
|
|
NOTIFIER = utils.DO_NOTHING
|
|
|
|
|
|
def initialized():
|
|
return None not in [TRANSPORT, NOTIFIER]
|
|
|
|
|
|
def cleanup():
|
|
global TRANSPORT, NOTIFICATION_TRANSPORT, NOTIFIER
|
|
assert TRANSPORT is not None
|
|
|
|
TRANSPORT.cleanup()
|
|
NOTIFICATION_TRANSPORT.cleanup()
|
|
TRANSPORT = NOTIFICATION_TRANSPORT = NOTIFIER = None
|
|
|
|
|
|
def set_defaults(control_exchange):
|
|
messaging.set_transport_defaults(control_exchange)
|
|
|
|
|
|
def add_extra_exmods(*args):
|
|
EXTRA_EXMODS.extend(args)
|
|
|
|
|
|
def clear_extra_exmods():
|
|
del EXTRA_EXMODS[:]
|
|
|
|
|
|
def get_allowed_exmods():
|
|
return ALLOWED_EXMODS + EXTRA_EXMODS
|
|
|
|
|
|
class RequestContextSerializer(messaging.Serializer):
|
|
|
|
def __init__(self, base):
|
|
self._base = base
|
|
super(RequestContextSerializer, self).__init__()
|
|
|
|
def serialize_entity(self, context, entity):
|
|
if not self._base:
|
|
return entity
|
|
return self._base.serialize_entity(context, entity)
|
|
|
|
def deserialize_entity(self, context, entity):
|
|
if not self._base:
|
|
return entity
|
|
return self._base.deserialize_entity(context, entity)
|
|
|
|
def serialize_context(self, context):
|
|
_context = context.to_dict()
|
|
return _context
|
|
|
|
def deserialize_context(self, context):
|
|
return karbor.context.RequestContext.from_dict(context)
|
|
|
|
|
|
def get_client(target, version_cap=None, serializer=None):
|
|
assert TRANSPORT is not None
|
|
serializer = RequestContextSerializer(serializer)
|
|
return messaging.RPCClient(TRANSPORT,
|
|
target,
|
|
version_cap=version_cap,
|
|
serializer=serializer)
|
|
|
|
|
|
def get_server(target, endpoints, serializer=None):
|
|
assert TRANSPORT is not None
|
|
serializer = RequestContextSerializer(serializer)
|
|
access_policy = dispatcher.DefaultRPCAccessPolicy
|
|
return messaging.get_rpc_server(TRANSPORT,
|
|
target,
|
|
endpoints,
|
|
executor='eventlet',
|
|
serializer=serializer,
|
|
access_policy=access_policy)
|
|
|
|
|
|
def get_notifier(service=None, host=None, publisher_id=None):
|
|
assert NOTIFIER is not None
|
|
if not publisher_id:
|
|
publisher_id = "%s.%s" % (service, host or CONF.host)
|
|
return NOTIFIER.prepare(publisher_id=publisher_id)
|