Support notify

Author: repeatedly

README.md

@@ -8,7 +8,7 @@ MessagePack RPC implementation based on Tornado.
 
     address = msgpackrpc.Address("localhost", 18800)
     client = msgpackrpc.Client(address)
-    result = client.call('add', 1, 2)  # = > 3  
+    result = client.call('sum', 1, 2)  # = > 3  
 
 # Installation
 
@@ -21,7 +21,7 @@ MessagePack RPC implementation based on Tornado.
 
 # TODO
 
-* Add Server
+* Add advanced and async return to Server.
 * UDP, UNIX Domain support
 * Utilities (MultiFuture, SessionPool)
 * Support pyev for performance if needed

msgpackrpc/session.py

@@ -53,6 +53,12 @@ def send_request(self, method, args):
 
         return future
 
+    def notify(self, method, *args):
+        def callback():
+            self._loop.stop()
+        self._transport.send_message([message.NOTIFY, method, args], callback=callback)
+        self._loop.start()
+
     def close(self):
         if self._transport:
             self._transport.close()

msgpackrpc/transport/tcp.py

@@ -21,17 +21,18 @@ def __init__(self, stream):
     def close(self):
         self._stream.close()
 
-    def send_message(self, message):
-        self._stream.write(self._packer.pack(message))
+    def send_message(self, message, callback=None):
+        self._stream.write(self._packer.pack(message), callback=callback)
 
     def on_read(self, data):
         self._unpacker.feed(data)
         for message in self._unpacker:
             self.on_message(message)
 
     def on_message(self, message, *args):
-        if len(message) != 4:
-            raise Exception("Invalid MessagePack-RPC protocol: message = {0}".format(message))
+        msgsize = len(message)
+        if msgsize != 4 and msgsize != 3:
+            raise RPCError("Invalid MessagePack-RPC protocol: message = {0}".format(message))
 
         msgtype = message[0]
         if msgtype == msgpackrpc.message.REQUEST:
@@ -86,15 +87,15 @@ def __init__(self, session, address, reconnect_limit):
         self._pending = []
         self._sockets = []
 
-    def send_message(self, message):
+    def send_message(self, message, callback=None):
         if len(self._sockets) == 0:
             if self._connecting == 0:
                 self.connect()
                 self._connecting = 1
-            self._pending.append(message)
+            self._pending.append((message, callback))
         else:
             sock = self._sockets[0]
-            sock.send_message(message)
+            sock.send_message(message, callback)
 
     def connect(self):
         stream = IOStream(self._address.socket(), io_loop=self._session._loop._ioloop)
@@ -111,8 +112,8 @@ def close(self):
 
     def on_connect(self, sock):
         self._sockets.append(sock)
-        for pending in self._pending:
-            sock.send_message(pending)
+        for pending, callback in self._pending:
+            sock.send_message(pending, callback)
         self._pending = []
 
     def on_connect_failed(self, sock):
@@ -145,7 +146,7 @@ def on_request(self, msgid, method, param):
         self._transport._server.on_request(self, msgid, method, param)
 
     def on_notify(self, method, param):
-        self._transport._server.on_request(method, param)
+        self._transport._server.on_notify(method, param)
 
 
 class MessagePackServer(netutil.TCPServer):