¡@
OpenStack Study: rpc.py**** CubicPower OpenStack Study ****
# Copyright 2013 Red Hat, Inc. # # 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', 'TRANSPORT_ALIASES', ] from oslo.config import cfg from oslo import messaging import cinder.context import cinder.exception from cinder.openstack.common import jsonutils CONF = cfg.CONF TRANSPORT = None NOTIFIER = None ALLOWED_EXMODS = [ cinder.exception.__name__, ] EXTRA_EXMODS = [] # NOTE(flaper87): The cinder.openstack.common.rpc entries are # for backwards compat with Havana rpc_backend configuration # values. The cinder.rpc entries are for compat with Folsom values. TRANSPORT_ALIASES = { 'cinder.openstack.common.rpc.impl_kombu': 'rabbit', 'cinder.openstack.common.rpc.impl_qpid': 'qpid', 'cinder.openstack.common.rpc.impl_zmq': 'zmq', 'cinder.rpc.impl_kombu': 'rabbit', 'cinder.rpc.impl_qpid': 'qpid', 'cinder.rpc.impl_zmq': 'zmq', }
**** CubicPower OpenStack Study **** def init(conf): global TRANSPORT, NOTIFIER exmods = get_allowed_exmods() TRANSPORT = messaging.get_transport(conf, allowed_remote_exmods=exmods, aliases=TRANSPORT_ALIASES) serializer = RequestContextSerializer(JsonPayloadSerializer()) NOTIFIER = messaging.Notifier(TRANSPORT, serializer=serializer)
**** CubicPower OpenStack Study **** def initialized(): return None not in [TRANSPORT, NOTIFIER]
**** CubicPower OpenStack Study **** def cleanup(): global TRANSPORT, NOTIFIER assert TRANSPORT is not None assert NOTIFIER is not None TRANSPORT.cleanup() TRANSPORT = NOTIFIER = None
**** CubicPower OpenStack Study **** def set_defaults(control_exchange): messaging.set_transport_defaults(control_exchange)
**** CubicPower OpenStack Study **** def add_extra_exmods(*args): EXTRA_EXMODS.extend(args)
**** CubicPower OpenStack Study **** def clear_extra_exmods(): del EXTRA_EXMODS[:]
**** CubicPower OpenStack Study **** def get_allowed_exmods(): return ALLOWED_EXMODS + EXTRA_EXMODS
**** CubicPower OpenStack Study **** class JsonPayloadSerializer(messaging.NoOpSerializer): @staticmethod
**** CubicPower OpenStack Study **** def serialize_entity(context, entity): return jsonutils.to_primitive(entity, convert_instances=True)
**** CubicPower OpenStack Study **** class RequestContextSerializer(messaging.Serializer):
**** CubicPower OpenStack Study **** def __init__(self, base): self._base = base
**** CubicPower OpenStack Study **** def serialize_entity(self, context, entity): if not self._base: return entity return self._base.serialize_entity(context, entity)
**** CubicPower OpenStack Study **** def deserialize_entity(self, context, entity): if not self._base: return entity return self._base.deserialize_entity(context, entity)
**** CubicPower OpenStack Study **** def serialize_context(self, context): return context.to_dict()
**** CubicPower OpenStack Study **** def deserialize_context(self, context): return cinder.context.RequestContext.from_dict(context) def get_transport_url(url_str=None): return messaging.TransportURL.parse(CONF, url_str, TRANSPORT_ALIASES) 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) return messaging.get_rpc_server(TRANSPORT, target, endpoints, executor='eventlet', serializer=serializer) 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)
**** CubicPower OpenStack Study **** def get_transport_url(url_str=None): return messaging.TransportURL.parse(CONF, url_str, TRANSPORT_ALIASES)
**** CubicPower OpenStack Study **** 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)
**** CubicPower OpenStack Study **** def get_server(target, endpoints, serializer=None): assert TRANSPORT is not None serializer = RequestContextSerializer(serializer) return messaging.get_rpc_server(TRANSPORT, target, endpoints, executor='eventlet', serializer=serializer)
**** CubicPower OpenStack Study **** 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)
|