3aca3f7745
Now that all preparations are done, actually port the code to use
oslo.messaging. This patch does as little as possible. Follow up patches
that refactor and cleanup the code and configuration files, will be
merged later. The reason for this is to make the patch as slim as
possible, to make review process more smooth and concentrated.
Details:
* neutron/common/rpc.py:
- added init() and cleanup() to set global RPC layer state.
- added utility functions: get_server(), get_client(), get_notifier()
that wrap up oslo.messaging API a bit, enforcing eventlet executor
and setting serializer, among other things.
- removed PluginRpcDispatcher, instead introduced PluginRpcSerializer
to use as a default serializer for API callbacks.
* neutron/common/rpc_compat.py:
- emulated incubator RPC layer behaviour thru previously introduced
stub classes (RpcCallback, RpcProxy, ...) using new oslo.messaging
API.
- switched to using new oslo.messaging exception types.
* neutron/service.py:
- expect multiple RPC listeners that are of MessageHandlingServer
type, not GreenThread.
* neutron/common/config.py:
- initialize RPC layer in init()
* setup.cfg:
- added entry points for old notifier drivers to retain backward
compatibility.
* neutron/tests/...:
- introduced fake_notifier to replace impl_fake.
- faked out consume_in_thread() to avoid starting RPC listeners when
running unit tests.
- used 'fake' transport driver.
- made sure neutron.test.* exceptions are caught.
- initialize and clean up RPC layer for each test case.
* Ported all affected code from using neutron.openstack.common.notifier
API to oslo.messaging.Notifier.
* rpc.set_defaults() was renamed to rpc.set_transport_defaults()
* other changes not worth mentioning here.
blueprint oslo-messaging
DocImpact
Change-Id: I5a91c34df6e300f2dc46217b1b16352fcc3039fc
137 lines
4.1 KiB
Python
137 lines
4.1 KiB
Python
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
|
|
|
# Copyright (c) 2012 OpenStack Foundation.
|
|
# All Rights Reserved.
|
|
#
|
|
# 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.
|
|
|
|
from oslo.config import cfg
|
|
from oslo import messaging
|
|
from oslo.messaging import serializer as om_serializer
|
|
|
|
from neutron.common import exceptions
|
|
from neutron import context
|
|
from neutron.openstack.common import log as logging
|
|
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
|
|
|
|
TRANSPORT = None
|
|
NOTIFIER = None
|
|
|
|
ALLOWED_EXMODS = [
|
|
exceptions.__name__,
|
|
]
|
|
EXTRA_EXMODS = []
|
|
|
|
|
|
TRANSPORT_ALIASES = {
|
|
'neutron.openstack.common.rpc.impl_fake': 'fake',
|
|
'neutron.openstack.common.rpc.impl_qpid': 'qpid',
|
|
'neutron.openstack.common.rpc.impl_kombu': 'rabbit',
|
|
'neutron.openstack.common.rpc.impl_zmq': 'zmq',
|
|
'neutron.rpc.impl_fake': 'fake',
|
|
'neutron.rpc.impl_qpid': 'qpid',
|
|
'neutron.rpc.impl_kombu': 'rabbit',
|
|
'neutron.rpc.impl_zmq': 'zmq',
|
|
}
|
|
|
|
|
|
def init(conf):
|
|
global TRANSPORT, NOTIFIER
|
|
exmods = get_allowed_exmods()
|
|
TRANSPORT = messaging.get_transport(conf,
|
|
allowed_remote_exmods=exmods,
|
|
aliases=TRANSPORT_ALIASES)
|
|
NOTIFIER = messaging.Notifier(TRANSPORT)
|
|
|
|
|
|
def cleanup():
|
|
global TRANSPORT, NOTIFIER
|
|
assert TRANSPORT is not None
|
|
assert NOTIFIER is not None
|
|
TRANSPORT.cleanup()
|
|
TRANSPORT = NOTIFIER = None
|
|
|
|
|
|
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
|
|
|
|
|
|
def get_client(target, version_cap=None, serializer=None):
|
|
assert TRANSPORT is not None
|
|
serializer = PluginRpcSerializer(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 = PluginRpcSerializer(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 cfg.CONF.host)
|
|
return NOTIFIER.prepare(publisher_id=publisher_id)
|
|
|
|
|
|
class PluginRpcSerializer(om_serializer.Serializer):
|
|
"""This serializer is used to convert RPC common context into
|
|
Neutron Context.
|
|
"""
|
|
def __init__(self, base):
|
|
super(PluginRpcSerializer, self).__init__()
|
|
self._base = base
|
|
|
|
def serialize_entity(self, ctxt, entity):
|
|
if not self._base:
|
|
return entity
|
|
return self._base.serialize_entity(ctxt, entity)
|
|
|
|
def deserialize_entity(self, ctxt, entity):
|
|
if not self._base:
|
|
return entity
|
|
return self._base.deserialize_entity(ctxt, entity)
|
|
|
|
def serialize_context(self, ctxt):
|
|
return ctxt.to_dict()
|
|
|
|
def deserialize_context(self, ctxt):
|
|
rpc_ctxt_dict = ctxt.copy()
|
|
user_id = rpc_ctxt_dict.pop('user_id', None)
|
|
if not user_id:
|
|
user_id = rpc_ctxt_dict.pop('user', None)
|
|
tenant_id = rpc_ctxt_dict.pop('tenant_id', None)
|
|
if not tenant_id:
|
|
tenant_id = rpc_ctxt_dict.pop('project_id', None)
|
|
return context.Context(user_id, tenant_id,
|
|
load_admin_roles=False, **rpc_ctxt_dict)
|