Merge lp:~free.ekanayaka/landscape-client/method-call-defer into lp:~landscape/landscape-client/amp-trunk

[Free Ekanayaka]

[lp:~free.ekanayaka/landscape-client/method-call-defer]

216. By Free Ekanayaka on 2010-01-25

Use maybeDeferred to simplify receive_method_call logic (Duncan [3])

Unmerged revisions

=== modified file 'landscape/lib/amp.py'
--- landscape/lib/amp.py	2010-01-20 08:33:12 +0000
+++ landscape/lib/amp.py	2010-01-20 09:25:23 +0000
@@ -1,8 +1,10 @@
 """Expose the methods of a remote object over AMP."""
 
+from uuid import uuid4
 from twisted.internet.defer import Deferred
 from twisted.internet.protocol import ServerFactory, ClientFactory
 from twisted.protocols.amp import Argument, String, Command, AMP
+from twisted.python.failure import Failure
 
 from landscape.lib.bpickle import loads, dumps, dumps_table
 
@@ -35,11 +37,21 @@
                  ("args", MethodCallArgument()),
                  ("kwargs", MethodCallArgument())]
 
-    response = [("result", MethodCallArgument())]
+    response = [("result", MethodCallArgument()),
+                ("deferred", String(optional=True))]
 
     errors = {MethodCallError: "METHOD_CALL_ERROR"}
 
 
+class DeferredResponse(Command):
+    """Fire a L{Deferred} associated with an outstanding method call result."""
+
+    arguments = [("uuid", String()),
+                 ("result", MethodCallArgument(optional=True)),
+                 ("failure", String(optional=True))]
+    requiresAnswer = False
+
+
 class MethodCallServerProtocol(AMP):
     """Expose methods of a local object over AMP.
 
@@ -72,13 +84,105 @@
             result = getattr(self.factory.object, method)(*args, **kwargs)
         except Exception, error:
             raise MethodCallError("Remote exception %s" % str(error))
+
+        # If the method returns a Deferred, register a callback that will
+        # eventually notify the remote peer of its success or failure.
+        if isinstance(result, Deferred):
+
+            # If the Deferred was already fired, we can return its result
+            if result.called:
+                if isinstance(result.result, Failure):
+                    failure = str(result.result.value)
+                    result.addErrback(lambda error: None) # Stop propagating
+                    raise MethodCallError(failure)
+                return {"result": result.result}
+
+            uuid = str(uuid4())
+            result.addBoth(self.send_deferred_response, uuid)
+            return {"result": None, "deferred": uuid}
+
         if not MethodCallArgument.check(result):
             raise MethodCallError("Non-serializable result")
         return {"result": result}
 
+    def send_deferred_response(self, result, uuid):
+        """Send a L{DeferredResponse} for the deferred with given C{uuid}.
+
+        This is called when the result of a L{Deferred} returned by an
+        object's method becomes available. A L{DeferredResponse} notifying
+        such result (either success or failure) is sent to the peer.
+        """
+        kwargs = {"uuid": uuid}
+        if isinstance(result, Failure):
+            kwargs["failure"] = str(result.value)
+        else:
+            kwargs["result"] = result
+        self.callRemote(DeferredResponse, **kwargs)
+
 
 class MethodCallClientProtocol(AMP):
-    """Calls methods of a remote object over L{AMP}."""
+    """Calls methods of a remote object over L{AMP}.
+
+    @note: If the remote method returns a deferred, the associated local
+        deferred returned by L{send_method_call} will result in the same
+        callback value of the remote deferred.
+    @cvar timeout: A timeout for remote methods returning L{Deferred}s, if a
+        response for the deferred is not received within this amount of
+        seconds, the remote method call will errback with a L{MethodCallError}.
+    """
+    timeout = 60
+
+    def __init__(self):
+        AMP.__init__(self)
+        self._pending_responses = {}
+
+    @DeferredResponse.responder
+    def receive_deferred_response(self, uuid, result, failure):
+        """Receive the deferred L{MethodCall} response.
+
+        @param uuid: The id of the L{MethodCall} we're getting the result of.
+        @param result: The result of the associated deferred if successful.
+        @param failure: The failure message of the deferred if it failed.
+        """
+        self.fire_pending_response(uuid, result, failure)
+        return {}
+
+    def fire_pending_response(self, uuid, result, failure):
+        """Fire the L{Deferred} associated with a pending response.
+
+        @param uuid: The id of the L{MethodCall} we're getting the result of.
+        @param result: The result of the associated deferred if successful.
+        @param failure: The failure message of the deferred if it failed.
+        """
+        try:
+            deferred, call = self._pending_responses.pop(uuid)
+        except KeyError:
+            # Late response for a request that has timeout, just ignore it
+            return
+        if not call.called:
+            call.cancel()
+        if failure is None:
+            deferred.callback({"result": result})
+        else:
+            deferred.errback(MethodCallError(failure))
+
+    def handle_response(self, response):
+        """Handle a L{MethodCall} response.
+
+        If the response is tagged as deferred, it will be queued as pending,
+        and a L{Deferred} is returned, which will be fired as soon as the
+        final response becomes available, or the timeout is reached.
+        """
+        if response["deferred"]:
+            uuid = response["deferred"]
+            deferred = Deferred()
+            call = self.factory.reactor.callLater(self.timeout,
+                                                  self.fire_pending_response,
+                                                  uuid, None, "timeout")
+            self._pending_responses[uuid] = (deferred, call)
+            return deferred
+
+        return response
 
     def send_method_call(self, method, args=[], kwargs={}):
         """Send a L{MethodCall} command with the given arguments.
@@ -87,8 +191,12 @@
         @param args: The positional arguments to pass to the remote method.
         @param kwargs: The keyword arguments to pass to the remote method.
         """
-        return self.callRemote(MethodCall,
-                               method=method, args=args, kwargs=kwargs)
+        result = self.callRemote(MethodCall,
+                                 method=method, args=args, kwargs=kwargs)
+        # The result can be C{None} only if the requested command is a
+        # DeferredResponse, which has requiresAnswer set to False
+        if result is not None:
+            return result.addCallback(self.handle_response)
 
 
 class MethodCallServerFactory(ServerFactory):
@@ -111,16 +219,18 @@
 
     def __init__(self, reactor, notifier):
         """
-        @param reactor: The reactor used to schedule connection callbacks.
+        @param reactor: The reactor used by the created protocols to schedule
+            notifications and timeouts.
         @param notifier: A function that will be called when the connection is
             established. It will be passed the protocol instance as argument.
         """
-        self._reactor = reactor
+        self.reactor = reactor
         self._notifier = notifier
 
     def buildProtocol(self, addr):
         protocol = self.protocol()
-        self._reactor.callLater(0, self._notifier, protocol)
+        protocol.factory = self
+        self.reactor.callLater(0, self._notifier, protocol)
         return protocol
 
 

=== modified file 'landscape/lib/tests/test_amp.py'
--- landscape/lib/tests/test_amp.py	2010-01-20 08:33:12 +0000
+++ landscape/lib/tests/test_amp.py	2010-01-20 09:25:23 +0000
@@ -1,9 +1,11 @@
 from twisted.internet import reactor
+from twisted.internet.defer import Deferred
 from twisted.internet.protocol import ClientCreator
 
 from landscape.lib.amp import (
-    MethodCallError, MethodCallServerProtocol, MethodCallClientProtocol,
-    MethodCallServerFactory, RemoteObject, RemoteObjectCreator)
+    MethodCallError, MethodCallServerProtocol,
+    MethodCallClientProtocol, MethodCallServerFactory,
+    MethodCallClientFactory, RemoteObject, RemoteObjectCreator)
 from landscape.tests.helpers import LandscapeTest
 
 
@@ -60,6 +62,24 @@
     def translate(self, word):
         raise WordsException("Unknown word")
 
+    def google(self, word):
+        deferred = Deferred()
+        if word == "Landscape":
+            reactor.callLater(0.01, lambda: deferred.callback("Cool!"))
+        elif word == "Easy query":
+            deferred.callback("Done!")
+        elif word == "Weird stuff":
+            reactor.callLater(0.01, lambda: deferred.errback(Exception("bad")))
+        elif word == "Censored":
+            deferred.errback(Exception("very bad"))
+        elif word == "Long query":
+            # Do nothing, the deferred won't be fired at all
+            pass
+        elif word == "Slowish query":
+            # Fire the result after a while.
+            reactor.callLater(0.05, lambda: deferred.callback("Done!"))
+        return deferred
+
 
 class WordsProtocol(MethodCallServerProtocol):
 
@@ -72,7 +92,8 @@
                "multiply_alphabetically",
                "translate",
                "meaning_of_life",
-               "guess"]
+               "guess",
+               "google"]
 
 
 class MethodCallProtocolTest(LandscapeTest):
@@ -114,7 +135,8 @@
         result = self.protocol.send_method_call(method="empty",
                                                 args=[],
                                                 kwargs={})
-        return self.assertSuccess(result, {"result": None})
+        return self.assertSuccess(result, {"result": None,
+                                           "deferred": None})
 
     def test_with_return_value(self):
         """
@@ -124,7 +146,8 @@
         result = self.protocol.send_method_call(method="motd",
                                                 args=[],
                                                 kwargs={})
-        return self.assertSuccess(result, {"result": "Words are cool"})
+        return self.assertSuccess(result, {"result": "Words are cool",
+                                           "deferred": None})
 
     def test_with_one_argument(self):
         """
@@ -134,7 +157,8 @@
         result = self.protocol.send_method_call(method="capitalize",
                                                 args=["john"],
                                                 kwargs={})
-        return self.assertSuccess(result, {"result": "John"})
+        return self.assertSuccess(result, {"result": "John",
+                                           "deferred": None})
 
     def test_with_boolean_return_value(self):
         """
@@ -143,7 +167,8 @@
         result = self.protocol.send_method_call(method="is_short",
                                                 args=["hi"],
                                                 kwargs={})
-        return self.assertSuccess(result, {"result": True})
+        return self.assertSuccess(result, {"result": True,
+                                           "deferred": None})
 
     def test_with_many_arguments(self):
         """
@@ -152,7 +177,8 @@
         result = self.protocol.send_method_call(method="concatenate",
                                                 args=["You ", "rock"],
                                                 kwargs={})
-        return self.assertSuccess(result, {"result": "You rock"})
+        return self.assertSuccess(result, {"result": "You rock",
+                                           "deferred": None})
 
     def test_with_default_arguments(self):
         """
@@ -162,7 +188,8 @@
         result = self.protocol.send_method_call(method="lower_case",
                                                 args=["OHH"],
                                                 kwargs={})
-        return self.assertSuccess(result, {"result": "ohh"})
+        return self.assertSuccess(result, {"result": "ohh",
+                                           "deferred": None})
 
     def test_with_overriden_default_arguments(self):
         """
@@ -173,7 +200,8 @@
         result = self.protocol.send_method_call(method="lower_case",
                                                 args=["OHH"],
                                                 kwargs={"index": 2})
-        return self.assertSuccess(result, {"result": "OHh"})
+        return self.assertSuccess(result, {"result": "OHh",
+                                           "deferred": None})
 
     def test_with_dictionary_arguments(self):
         """
@@ -183,7 +211,8 @@
                                                        "alphabetically",
                                                 args=[{"foo": 2, "bar": 3}],
                                                 kwargs={})
-        return self.assertSuccess(result, {"result": "barbarbarfoofoo"})
+        return self.assertSuccess(result, {"result": "barbarbarfoofoo",
+                                           "deferred": None})
 
     def test_with_non_serializable_return_value(self):
         """
@@ -211,17 +240,19 @@
     def setUp(self):
         super(RemoteObjectTest, self).setUp()
         socket = self.mktemp()
-        factory = MethodCallServerFactory(Words())
-        factory.protocol = WordsProtocol
-        self.port = reactor.listenUNIX(socket, factory)
+        server_factory = MethodCallServerFactory(Words())
+        server_factory.protocol = WordsProtocol
+        self.port = reactor.listenUNIX(socket, server_factory)
 
         def set_remote(protocol):
             self.protocol = protocol
             self.words = RemoteObject(protocol)
 
-        connector = ClientCreator(reactor, MethodCallClientProtocol)
-        connected = connector.connectUNIX(socket)
-        return connected.addCallback(set_remote)
+        connected = Deferred()
+        connected.addCallback(set_remote)
+        client_factory = MethodCallClientFactory(reactor, connected.callback)
+        reactor.connectUNIX(socket, client_factory)
+        return connected
 
     def tearDown(self):
         self.protocol.transport.loseConnection()
@@ -319,6 +350,62 @@
         result = self.words.guess("word", "cool", value=4)
         return self.assertSuccess(result, "Guessed!")
 
+    def test_with_success_full_deferred(self):
+        """
+        If the target object method returns a L{Deferred}, it is handled
+        transparently.
+        """
+        result = self.words.google("Landscape")
+        return self.assertSuccess(result, "Cool!")
+
+    def test_with_failing_deferred(self):
+        """
+        If the target object method returns a failing L{Deferred}, a
+        L{MethodCallError} is raised.
+        """
+        result = self.words.google("Weird stuff")
+        return self.assertFailure(result, MethodCallError)
+
+    def test_with_already_callback_deferred(self):
+        """
+        The target object method can return an already fired L{Deferred}.
+        """
+        result = self.words.google("Easy query")
+        return self.assertSuccess(result, "Done!")
+
+    def test_with_already_errback_deferred(self):
+        """
+        If the target object method can return an already failed L{Deferred}.
+        """
+        result = self.words.google("Censored")
+        return self.assertFailure(result, MethodCallError)
+
+    def test_with_deferred_timeout(self):
+        """
+        If the peer protocol doesn't send a response for a deferred within
+        the given timeout, the method call fails.
+        """
+        self.protocol.timeout = 0.1
+        result = self.words.google("Long query")
+        return self.assertFailure(result, MethodCallError)
+
+    def test_with_late_response(self):
+        """
+        If the peer protocol sends a late response for a request that has
+        already timeout, that response is ignored.
+        """
+        self.protocol.timeout = 0.01
+        result = self.words.google("Slowish query")
+        self.assertFailure(result, MethodCallError)
+
+        def assert_late_response_is_handled(ignored):
+            deferred = Deferred()
+            # We wait a bit to be sure that the late response gets delivered
+            reactor.callLater(0.1, lambda: deferred.callback(None))
+            return deferred
+
+        return result.addCallback(assert_late_response_is_handled)
+
 
 class RemoteObjectCreatorTest(LandscapeTest):