lightning.py: parse multiple JSON RPC commands accurately.

We need to keep the remaining buffer, and we need to try to parse it
before we read the next.  I first tried keeping it in the object, but
its lifetime is that of the *socket*, which we actually reopen for
every command.

Signed-off-by: Rusty Russell <rusty@rustcorp.com.au>

CHANGELOG.md

@@ -20,6 +20,7 @@ changes.
 ### Fixed
 
 - JSON API: uppercase invoices now parsed correctly (broken in 0.6.2).
+- pylightning: handle multiple simultanous RPC replies reliably.
 
 ### Security
 

contrib/pylightning/lightning/lightning.py

@@ -25,26 +25,23 @@ class UnixDomainSocketRpc(object):
         s = json.dumps(obj)
         sock.sendall(bytearray(s, 'UTF-8'))
 
-    def _readobj(self, sock):
-        buff = b''
+    def _readobj(self, sock, buff=b''):
+        """Read a JSON object, starting with buff; returns object and any buffer left over"""
         while True:
             try:
-                b = sock.recv(1024)
-                buff += b
-                if len(b) == 0:
-                    return {'error': 'Connection to RPC server lost.'}
-
-                if buff[-3:] != b' }\n':
-                    continue
-
                 # Convert late to UTF-8 so glyphs split across recvs do not
                 # impact us
-                objs, _ = self.decoder.raw_decode(buff.decode("UTF-8"))
-                return objs
+                objs, len_used = self.decoder.raw_decode(buff.decode("UTF-8"))
+                return objs, buff[len_used:].lstrip()
             except ValueError:
                 # Probably didn't read enough
                 pass
 
+            b = sock.recv(1024)
+            buff += b
+            if len(b) == 0:
+                return {'error': 'Connection to RPC server lost.'}, buff.lstrip()
+
     def __getattr__(self, name):
         """Intercept any call that is not explicitly defined and call @call
 
@@ -65,6 +62,7 @@ class UnixDomainSocketRpc(object):
         # Filter out arguments that are None
         payload = {k: v for k, v in payload.items() if v is not None}
 
+        # FIXME: we open a new socket for every readobj call...
         sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
         sock.connect(self.socket_path)
         self._writeobj(sock, {
@@ -72,7 +70,7 @@ class UnixDomainSocketRpc(object):
             "params": payload,
             "id": 0
         })
-        resp = self._readobj(sock)
+        resp, _ = self._readobj(sock)
         sock.close()
 
         self.logger.debug("Received response for %s call: %r", method, resp)

tests/test_misc.py

@@ -611,8 +611,9 @@ def test_multirpc(node_factory):
 
     sock.sendall(b'\n'.join(commands))
 
+    buff = b''
     for i in commands:
-        l1.rpc._readobj(sock)
+        _, buff = l1.rpc._readobj(sock, buff)
     sock.close()