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

oslo/messaging/notify/notifier.py

@@ -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,

oslo/messaging/rpc/client.py

@@ -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.

oslo/messaging/rpc/dispatcher.py

@@ -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)

oslo/messaging/rpc/server.py

@@ -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__ = [

oslo/messaging/serializer.py

@@ -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

tests/test_notifier.py

@@ -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):

tests/test_rpc_client.py

@@ -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).\
+        transport._send(messaging.Target(),
-            AndReturn(self.retval)
+                        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

tests/test_rpc_dispatcher.py

@@ -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()

tests/test_rpc_server.py

@@ -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)