Use msgpack for quote-packet serialization

piker/brokers/core.py

@@ -3,12 +3,12 @@ Core broker-daemon tasks and API.
 """
 import time
 import inspect
-import json
 from functools import partial
 import socket
 from types import ModuleType
 from typing import Coroutine
 
+import msgpack
 import trio
 
 from ..log import get_logger
@@ -88,6 +88,7 @@ class StreamQueue:
         """
         delim = self._delim
         buff = b''
+        unpacker = msgpack.Unpacker(raw=False)
         while True:
             packets = []
             try:
@@ -101,23 +102,13 @@ class StreamQueue:
                 log.debug("Stream connection was closed")
                 return
 
-            if buff:  # last received packet was segmented
+            unpacker.feed(data)
-                data = buff + data
+            for packet in unpacker:
-
+                yield packet
-            # if last packet has not fully arrived it will
-            # be a truncated byte-stream
-            packets = data.split(delim)
-            buff = packets.pop()
-
-            for packet in packets:
-                try:
-                    yield json.loads(packet)
-                except json.decoder.JSONDecodeError:
-                    log.exception(f"Failed to process JSON packet: {buff}")
-                    continue
 
     async def put(self, data):
-        return await self.stream.send_all(json.dumps(data).encode() + b'\n')
+        return await self.stream.send_all(
+            msgpack.dumps(data, use_bin_type=True))
 
     async def get(self):
         return await self._agen.asend(None)

setup.py

@@ -36,7 +36,7 @@ setup(
     },
     install_requires=[
         'click', 'colorlog', 'trio', 'attrs', 'async_generator',
-        'pygments', 'cython', 'asks', 'pandas',
+        'pygments', 'cython', 'asks', 'pandas', 'msgpack',
         #'kivy',  see requirement.txt; using a custom branch atm
     ],
     extras_require={