Add AsyncResult to server as delayed return

Author: repeatedly

README.md

@@ -84,7 +84,7 @@ Test code are available in example directory(bench_client.py and bench_server.py
 
 ## TODO
 
-* Add advanced and async return to Server.
+* Add advanced return to Server.
 * UDP, UNIX Domain support
 * Utilities (MultiFuture, SessionPool)
 * Support pyev for performance if needed

msgpackrpc/server.py

@@ -45,13 +45,39 @@ def dispatch(self, method, param, responder):
             method = force_str(method)
             if not hasattr(self._dispatcher, method):
                 raise error.NoMethodError("'{0}' method not found".format(method))
-            responder.set_result(getattr(self._dispatcher, method)(*param))
+
+            result = getattr(self._dispatcher, method)(*param)
+            if isinstance(result, AsyncResult):
+                result.set_responder(responder)
+            else:
+                responder.set_result(result)
         except Exception as e:
             responder.set_error(str(e))
 
         # TODO: Support advanced and async return
 
 
+class AsyncResult:
+    def __init__(self):
+        self._responder = None
+        self._result = None
+
+    def set_result(self, value, error=None):
+        if self._responder is not None:
+            self._responder.set_result(value, error)
+        else:
+            self._result = [value, error]
+
+    def set_error(self, error, value=None):
+        self.set_result(value, error)
+
+    def set_responder(self, responder):
+        self._responder = responder
+        if self._result is not None:
+            self._responder.set_result(*self._result)
+            self._result = None
+
+
 class _Responder:
     def __init__(self, sendable, msgid):
         self._sendable = sendable

test/test_msgpackrpc.py

@@ -58,6 +58,14 @@ def long_exec(self):
             sleep(3)
             return 'finish!'
 
+        def async_result(self):
+            ar = msgpackrpc.server.AsyncResult()
+            def do_async():
+                sleep(2)
+                ar.set_result("You are async!")
+            threading.Thread(target=do_async).start()
+            return ar
+
     def setUp(self):
         self._address = msgpackrpc.Address('localhost', helper.unused_port())
 
@@ -146,6 +154,10 @@ def test_unknown_method(self):
             message = e.args[0]
             self.assertEqual(message, "'unknown' method not found", "Error message mismatched")
 
+    def test_async_result(self):
+        client = self.setup_env();
+        self.assertEqual(client.call('async_result'), "You are async!")
+
     def test_connect_failed(self):
         client = self.setup_env();
         client = msgpackrpc.Client(msgpackrpc.Address('localhost', self._address.port - 10), unpack_encoding='utf-8')