Add MessagePack RPC Server

Author: repeatedly

msgpackrpc/__init__.py

@@ -10,4 +10,5 @@
 # shortcut for most-used symbols
 from msgpackrpc.loop import Loop
 from msgpackrpc.client import Client
+from msgpackrpc.server import Server
 from msgpackrpc.address import Address

msgpackrpc/error.py

@@ -6,3 +6,6 @@ class TimeoutError(RPCError):
 
 class TransportError(RPCError):
     pass
+
+class NoMethodError(RPCError):
+    pass

msgpackrpc/server.py

@@ -0,0 +1,74 @@
+import msgpack
+
+from msgpackrpc import error
+from msgpackrpc import loop
+from msgpackrpc import message
+from msgpackrpc import session
+from msgpackrpc.transport import tcp
+
+class Server(session.Session):
+    """\
+    Server is usaful for MessagePack RPC Server.
+    """
+
+    def __init__(self, dispatcher, loop=loop.Loop(), builder=tcp):
+        #session.Session.__init__(self, address, timeout, loop, builder)
+        self._loop = loop
+        self._builder = builder
+        self._listeners = []
+        self._dispatcher = dispatcher
+
+    def listen(self, address):
+        listener = self._builder.ServerTransport(address)
+        listener.listen(self)
+        self._listeners.append(listener)
+
+    def start(self):
+        self._loop.start()
+
+    def stop(self):
+        self._loop.stop()
+
+    def close(self):
+        for listener in self._listeners:
+            listener.close()
+        #session.Session.close(self)
+
+    def on_request(self, sendable, msgid, method, param):
+        self.dispatch(method, param, _Responder(sendable, msgid))
+
+    def on_notify(self, method, param):
+        self.dispatch(method, param, _NullResponder())
+
+    def dispatch(self, method, param, responder):
+        try:
+            if not hasattr(self._dispatcher, method):
+                raise error.NoMethodError("{0} not found".format(method))
+            responder.set_result(getattr(self._dispatcher, method)(*param))
+        except Exception as e:
+            responder.set_error(str(e))
+
+        # TODO: Support advanced and async return
+
+
+class _Responder:
+    def __init__(self, sendable, msgid):
+        self._sendable = sendable
+        self._msgid = msgid
+        self._sent = False
+
+    def set_result(self, value, error=None, packer=msgpack.Packer()):
+        if not self._sent:
+            self._sendable.send_message([message.RESPONSE, self._msgid, error, value])
+            self._sent = True
+
+    def set_error(self, error, value=None):
+        self.set_result(value, error)
+
+
+class _NullResponder:
+    def set_result(self, value, error=None):
+        pass
+
+    def set_error(self, error, value=None):
+        pass

msgpackrpc/transport/tcp.py

@@ -130,3 +130,54 @@ def on_close(self, sock):
         else:
             # Tornado does not have on_connect_failed event.
             self.on_connect_failed(sock)
+
+
+class ServerSocket(BaseSocket):
+    def __init__(self, stream, transport):
+        BaseSocket.__init__(self, stream)
+        self._transport = transport
+        self._stream.read_until_close(self.on_read, self.on_read)
+        #self._stream.set_close_callback(self.on_close)
+
+    #def connect(self):
+    #    self._stream.connect(self._transport._address.unpack(), self.on_connect)
+
+    #def on_connect(self):
+    #    self._stream.read_until_close(self.on_read, self.on_read)
+    #    self._transport.on_connect(self)
+
+    #def on_connect_failed(self):
+    #    self._transport.on_connect_failed(self)
+
+    def on_close(self):
+        self._transport.on_close(self)
+
+    def on_request(self, msgid, method, param):
+        self._transport._server.on_request(self, msgid, method, param)
+        self._stream.close()
+
+    def on_notify(self, method, param):
+        self._transport._server.on_request(method, param)
+        self._stream.close()
+
+
+class MessagePackServer(netutil.TCPServer):
+    def __init__(self, transport, io_loop=None):
+        self._transport = transport
+        netutil.TCPServer.__init__(self, io_loop=io_loop)
+
+    def handle_stream(self, stream, address):
+        ServerSocket(stream, self._transport)
+
+
+class ServerTransport(object):
+    def __init__(self, address):
+        self._address = address;
+
+    def listen(self, server):
+        self._server = server;
+        self._mp_server = MessagePackServer(self, io_loop=self._server._loop._ioloop)
+        self._mp_server.listen(self._address.port)
+
+    def close(self):
+        self._mp_server.stop()

test/add_client.py

@@ -0,0 +1,13 @@
+import os
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+def unused_port():
+    import socket
+
+    sock = socket.socket()
+    sock.bind(("localhost", 0))
+    port = sock.getsockname()[1]
+    sock.close()
+    return port

test/add_server.rb

@@ -0,0 +1,51 @@
+import threading
+import unittest
+
+import helper
+import msgpackrpc
+
+
+class TestMessagePackRPC(unittest.TestCase):
+    class TestServer(object):
+        def hello(self):
+            return "world"
+
+        def sum(self, x, y):
+            return x + y
+
+
+    def setUp(self):
+        self._address = msgpackrpc.Address('localhost', helper.unused_port())
+
+    def setup_env(self):
+        def _start_server(server):
+            server.start()
+            server.close()
+
+        self._server = msgpackrpc.Server(TestMessagePackRPC.TestServer())
+        self._server.listen(self._address)
+        self._thread = threading.Thread(target=_start_server, args=(self._server,))
+        self._thread.start()
+
+        import time
+        time.sleep(1)
+
+        self._client = msgpackrpc.Client(self._address)
+        return self._client;
+
+    def tearDown(self):
+        self._client.close();
+        self._server.stop();
+        self._thread.join();
+
+    def test_hello(self):
+        client = self.setup_env();
+        self.assertEqual(client.call('hello'), "world", "sum result is incorrect")
+
+    def test_add(self):
+        client = self.setup_env();
+        self.assertEqual(client.call('sum', 1, 2), 3, "sum result is incorrect")
+
+
+if __name__ == '__main__':
+    unittest.main()