f1612f2895
Oslo's logging code has some useful support for including bits of the request context in log messages. While this isn't exclusively about the request context in a dispatching RPC method, it seems useful for oslo.messaging to support the concept for at least this use case simply by recording the context in a thread local store before dispatching an endpoint method and immediately clearing it when the method returns. Note, we don't need to store weak refs in our store because we will clear the reference in all cases rather than ever leaving a stale reference around in the store. Change-Id: I70ac06ed3a2a891a7a7b388b1823a0f3b08f2dd1
133 lines
4.6 KiB
Python
133 lines
4.6 KiB
Python
# Copyright 2010 United States Government as represented by the
|
|
# Administrator of the National Aeronautics and Space Administration.
|
|
# All Rights Reserved.
|
|
# Copyright 2013 Red Hat, Inc.
|
|
# Copyright 2013 New Dream Network, LLC (DreamHost)
|
|
#
|
|
# 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__ = [
|
|
'NoSuchMethod',
|
|
'RPCDispatcher',
|
|
'RPCDispatcherError',
|
|
'UnsupportedVersion',
|
|
]
|
|
|
|
import logging
|
|
|
|
from oslo.messaging import _utils as utils
|
|
from oslo.messaging import localcontext
|
|
from oslo.messaging import serializer as msg_serializer
|
|
from oslo.messaging import server as msg_server
|
|
from oslo.messaging import target
|
|
|
|
_LOG = logging.getLogger(__name__)
|
|
|
|
|
|
class RPCDispatcherError(msg_server.MessagingServerError):
|
|
"A base class for all RPC dispatcher exceptions."
|
|
|
|
|
|
class NoSuchMethod(RPCDispatcherError, AttributeError):
|
|
"Raised if there is no endpoint which exposes the requested method."
|
|
|
|
def __init__(self, method):
|
|
msg = "Endpoint does not support RPC method %s" % method
|
|
super(NoSuchMethod, self).__init__(msg)
|
|
self.method = method
|
|
|
|
|
|
class UnsupportedVersion(RPCDispatcherError):
|
|
"Raised if there is no endpoint which supports the requested version."
|
|
|
|
def __init__(self, version):
|
|
msg = "Endpoint does not support RPC version %s" % version
|
|
super(UnsupportedVersion, self).__init__(msg)
|
|
self.version = version
|
|
|
|
|
|
class RPCDispatcher(object):
|
|
"""A message dispatcher which understands RPC messages.
|
|
|
|
A MessageHandlingServer is constructed by passing a callable dispatcher
|
|
which is invoked with context and message dictionaries each time a message
|
|
is received.
|
|
|
|
RPCDispatcher is one such dispatcher which understands the format of RPC
|
|
messages. The dispatcher looks at the namespace, version and method values
|
|
in the message and matches those against a list of available endpoints.
|
|
|
|
Endpoints may have a target attribute describing the namespace and version
|
|
of the methods exposed by that object. All public methods on an endpoint
|
|
object are remotely invokable by clients.
|
|
"""
|
|
|
|
def __init__(self, endpoints, serializer):
|
|
self.endpoints = endpoints
|
|
self.serializer = serializer or msg_serializer.NoOpSerializer()
|
|
self._default_target = target.Target()
|
|
|
|
@staticmethod
|
|
def _is_namespace(target, namespace):
|
|
return namespace == target.namespace
|
|
|
|
@staticmethod
|
|
def _is_compatible(target, version):
|
|
endpoint_version = target.version or '1.0'
|
|
return utils.version_is_compatible(endpoint_version, version)
|
|
|
|
def _dispatch(self, endpoint, method, ctxt, args):
|
|
new_args = dict()
|
|
for argname, arg in args.iteritems():
|
|
new_args[argname] = self.serializer.deserialize_entity(ctxt, arg)
|
|
result = getattr(endpoint, method)(ctxt, **new_args)
|
|
return self.serializer.serialize_entity(ctxt, result)
|
|
|
|
def __call__(self, ctxt, message):
|
|
"""Dispatch an RPC message to the appropriate endpoint method.
|
|
|
|
:param ctxt: the request context
|
|
:type ctxt: dict
|
|
:param message: the message payload
|
|
:type message: dict
|
|
:raises: NoSuchMethod, UnsupportedVersion
|
|
"""
|
|
method = message.get('method')
|
|
args = message.get('args', {})
|
|
namespace = message.get('namespace')
|
|
version = message.get('version', '1.0')
|
|
|
|
found_compatible = False
|
|
for endpoint in self.endpoints:
|
|
target = getattr(endpoint, 'target', None)
|
|
if not target:
|
|
target = self._default_target
|
|
|
|
if not (self._is_namespace(target, namespace) and
|
|
self._is_compatible(target, version)):
|
|
continue
|
|
|
|
if hasattr(endpoint, method):
|
|
localcontext.set_local_context(ctxt)
|
|
try:
|
|
return self._dispatch(endpoint, method, ctxt, args)
|
|
finally:
|
|
localcontext.clear_local_context()
|
|
|
|
found_compatible = True
|
|
|
|
if found_compatible:
|
|
raise NoSuchMethod(method)
|
|
else:
|
|
raise UnsupportedVersion(version)
|