openstack
/
oslo.messaging
Code Issues Proposed changes

Add a context serialization hook 

The client call() and cast() methods take a request context argument
which is expected to be a dictionary. The RPC endpoint methods on the
server are invoked with a dictionary context argument.

However, Nova passes a nova.context.RequestContext on the client side
and the oslo-incubator RPC code deserializes that as a amqp.RpcContext
which looks vaguely compatible with Nova's RequestContext.

Support the serialization and deserialization of RequestContext objects
with an additional (de)serialize_context() hook on our Serializer class.

Note: this is a backwards incompatible API change because Serializer
implementations which do not implement the new abstract methods would
no longer be possible to instantiate. Not a problem with this commit,
but shows the type of compat concerns we'll need to think about once the
API is locked down for good.

Change-Id: I20782bad77fa0b0e396d082df852ca355548f9b7
This commit is contained in:
Mark McLoughlin 2013-08-09 07:55:46 +01:00
parent bf171ede28
commit 2abb40f9e9
9 changed files with 68 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
oslo/messaging/notify/notifier.py
View File

@ -129,6 +129,7 @@ class Notifier(object):
def _notify(self, ctxt, event_type, payload, priority):
payload = self._serializer.serialize_entity(ctxt, payload)
ctxt = self._serializer.serialize_context(ctxt)
msg = dict(message_id=uuidutils.generate_uuid(),
publisher_id=self.publisher_id,

12
oslo/messaging/rpc/client.py
View File

@ -129,6 +129,8 @@ class _CallContext(object):
def cast(self, ctxt, method, **kwargs):
"""Invoke a method and return immediately. See RPCClient.cast()."""
msg = self._make_message(ctxt, method, kwargs)
ctxt = self.serializer.serialize_context(ctxt)
if self.version_cap:
self._check_version_cap(msg.get('version'))
try:
@ -149,6 +151,7 @@ class _CallContext(object):
def call(self, ctxt, method, **kwargs):
"""Invoke a method and wait for a reply. See RPCClient.call()."""
msg = self._make_message(ctxt, method, kwargs)
msg_ctxt = self.serializer.serialize_context(ctxt)
timeout = self.timeout
if self.timeout is None:
@ -160,7 +163,7 @@ class _CallContext(object):
self._check_version_cap(msg.get('version'))
try:
result = self.transport._send(self.target, ctxt, msg,
result = self.transport._send(self.target, msg_ctxt, msg,
wait_for_reply=True, timeout=timeout)
except driver_base.TransportDriverError as ex:
raise ClientSendError(self.target, ex)
@ -335,6 +338,9 @@ class RPCClient(object):
Method arguments must either be primitive types or types supported by
the client's serializer (if any).
Similarly, the request context must be a dict unless the client's
serializer supports serializing another type.
:param ctxt: a request context dict
:type ctxt: dict
:param method: the method name
@ -348,7 +354,9 @@ class RPCClient(object):
"""Invoke a method and wait for a reply.
Method arguments must either be primitive types or types supported by
the client's serializer (if any).
the client's serializer (if any). Similarly, the request context must
be a dict unless the client's serializer supports serializing another
type.
The semantics of how any errors raised by the remote RPC endpoint
method are handled are quite subtle.

1
oslo/messaging/rpc/dispatcher.py
View File

@ -86,6 +86,7 @@ class RPCDispatcher(object):
return utils.version_is_compatible(endpoint_version, version)
def _dispatch(self, endpoint, method, ctxt, args):
ctxt = self.serializer.deserialize_context(ctxt)
new_args = dict()
for argname, arg in args.iteritems():
new_args[argname] = self.serializer.deserialize_entity(ctxt, arg)

3
oslo/messaging/rpc/server.py
View File

@ -86,7 +86,8 @@ supplied by the client.
Parameters to the method invocation are primitive types and so must be the
return values from the methods. By supplying a serializer object, a server can
deserialize arguments from - serialize return values to - primitive types.
deserialize a request context and arguments from - and serialize return values
to - primitive types.
"""
__all__ = [

26
oslo/messaging/serializer.py
View File

@ -27,7 +27,7 @@ class Serializer(object):
def serialize_entity(self, ctxt, entity):
"""Serialize something to primitive form.
:param context: Request context
:param ctxt: Request context, in deserialized form
:param entity: Entity to be serialized
:returns: Serialized form of entity
"""
@ -36,11 +36,27 @@ class Serializer(object):
def deserialize_entity(self, ctxt, entity):
"""Deserialize something from primitive form.
:param context: Request context
:param ctxt: Request context, in deserialized form
:param entity: Primitive to be deserialized
:returns: Deserialized form of entity
"""
@abc.abstractmethod
def serialize_context(self, ctxt):
"""Serialize a request context into a dictionary.
:param ctxt: Request context
:returns: Serialized form of context
"""
@abc.abstractmethod
def deserialize_context(self, ctxt):
"""Deserialize a dictionary into a request context.
:param ctxt: Request context dictionary
:returns: Deserialized form of entity
"""
class NoOpSerializer(Serializer):
"""A serializer that does nothing."""
@ -50,3 +66,9 @@ class NoOpSerializer(Serializer):
def deserialize_entity(self, ctxt, entity):
return entity
def serialize_context(self, ctxt):
return ctxt
def deserialize_context(self, ctxt):
return ctxt

10
tests/test_notifier.py
View File

@ -201,12 +201,15 @@ class TestSerializer(test_utils.BaseTestCase):
timeutils.set_time_override()
self.mox.StubOutWithMock(serializer, 'serialize_context')
self.mox.StubOutWithMock(serializer, 'serialize_entity')
serializer.serialize_entity({}, 'bar').AndReturn('sbar')
serializer.serialize_context(dict(user='bob')).\
AndReturn(dict(user='alice'))
serializer.serialize_entity(dict(user='bob'), 'bar').AndReturn('sbar')
self.mox.ReplayAll()
notifier.info({}, 'test.notify', 'bar')
notifier.info(dict(user='bob'), 'test.notify', 'bar')
message = {
'message_id': str(message_id),
@ -217,7 +220,8 @@ class TestSerializer(test_utils.BaseTestCase):
'timestamp': str(timeutils.utcnow.override_time),
}
self.assertEquals(_impl_test.NOTIFICATIONS, [({}, message, 'INFO')])
self.assertEquals(_impl_test.NOTIFICATIONS,
[(dict(user='alice'), message, 'INFO')])
class TestLogNotifier(test_utils.BaseTestCase):

9
tests/test_rpc_client.py
View File

@ -295,11 +295,14 @@ class TestSerializer(test_utils.BaseTestCase):
msg = dict(method='foo',
args=dict([(k, 's' + v) for k, v in self.args.items()]))
kwargs = dict(wait_for_reply=True, timeout=None) if self.call else {}
transport._send(messaging.Target(), self.ctxt, msg, **kwargs).\
AndReturn(self.retval)
transport._send(messaging.Target(),
dict(user='alice'),
msg,
**kwargs).AndReturn(self.retval)
self.mox.StubOutWithMock(serializer, 'serialize_entity')
self.mox.StubOutWithMock(serializer, 'deserialize_entity')
self.mox.StubOutWithMock(serializer, 'serialize_context')
for arg in self.args:
serializer.serialize_entity(self.ctxt, arg).AndReturn('s' + arg)
@ -308,6 +311,8 @@ class TestSerializer(test_utils.BaseTestCase):
serializer.deserialize_entity(self.ctxt, self.retval).\
AndReturn('d' + self.retval)
serializer.serialize_context(self.ctxt).AndReturn(dict(user='alice'))
self.mox.ReplayAll()
method = client.call if self.call else client.cast

14
tests/test_rpc_dispatcher.py
View File

@ -128,29 +128,33 @@ class TestSerializer(test_utils.BaseTestCase):
scenarios = [
('no_args_or_retval',
dict(ctxt={}, args={}, retval=None)),
dict(ctxt={}, dctxt={}, args={}, retval=None)),
('args_and_retval',
dict(ctxt=dict(user='bob'),
dctxt=dict(user='alice'),
args=dict(a='a', b='b', c='c'),
retval='d')),
]
def test_serializer(self):
endpoint = _FakeEndpoint()
serializer = msg_serializer.NoOpSerializer
serializer = msg_serializer.NoOpSerializer()
dispatcher = messaging.RPCDispatcher([endpoint], serializer)
self.mox.StubOutWithMock(endpoint, 'foo')
args = dict([(k, 'd' + v) for k, v in self.args.items()])
endpoint.foo(self.ctxt, **args).AndReturn(self.retval)
endpoint.foo(self.dctxt, **args).AndReturn(self.retval)
self.mox.StubOutWithMock(serializer, 'serialize_entity')
self.mox.StubOutWithMock(serializer, 'deserialize_entity')
self.mox.StubOutWithMock(serializer, 'deserialize_context')
serializer.deserialize_context(self.ctxt).AndReturn(self.dctxt)
for arg in self.args:
serializer.deserialize_entity(self.ctxt, arg).AndReturn('d' + arg)
serializer.deserialize_entity(self.dctxt, arg).AndReturn('d' + arg)
serializer.serialize_entity(self.ctxt, self.retval).\
serializer.serialize_entity(self.dctxt, self.retval).\
AndReturn('s' + self.retval if self.retval else None)
self.mox.ReplayAll()

8
tests/test_rpc_server.py
View File

@ -51,6 +51,12 @@ class ServerSetupMixin(object):
def deserialize_entity(self, ctxt, entity):
return 'd' + (entity or '')
def serialize_context(self, ctxt):
return dict([(k, 's' + v) for k, v in ctxt.items()])
def deserialize_context(self, ctxt):
return dict([(k, 'd' + v) for k, v in ctxt.items()])
def __init__(self):
self.serializer = self.TestSerializer()
@ -254,7 +260,7 @@ class TestRPCServer(test_utils.BaseTestCase, ServerSetupMixin):
self.assertEqual(client.call({'dsa': 'b'},
'ctxt_check',
key='a'),
'dsb')
'dsdsb')
self._stop_server(client, server_thread)