Move JSON (de)serialization to socket boundaries

lib/tfw/networking/event_handler_connector.py

@@ -6,7 +6,7 @@ from .serialization import serialize_all
 from ..config import PUBLISHER_PORT, RECEIVER_PORT
 from ..config.logs import logging
 log = logging.getLogger(__name__)
-from ..util import parse_anchor_from_message, ZMQConnectorBase
+from ..util import ZMQConnectorBase
 
 
 ioloop.install()
@@ -38,7 +38,7 @@ class EventHandlerConnector(EventHandlerDownlinkConnector, EventHandlerUplinkCon
     def register_callback(self, callback):
         self._zmq_pull_stream.on_recv(callback)
 
-    def send_message(self, message: str, anchor: str = None):
+    def send_message(self, message: dict, anchor: str = None):
         if not anchor:
-            anchor = parse_anchor_from_message(message)
+            anchor = message['anchor']
         self._zmq_pub_socket.send_multipart(serialize_all(anchor, message))

lib/tfw/networking/serialization.py

@@ -1,3 +1,6 @@
+import json
+
+
 def encode_if_needed(value):
     if isinstance(value, str):
         value = value.encode('utf-8')
@@ -10,9 +13,9 @@ def decode_if_needed(value):
     return value
 
 
-def serialize_all(*args):
-    return [encode_if_needed(a) for a in args]
+def serialize_all(anchor, message):
+    return [encode_if_needed(a) for a in (anchor, json.dumps(message))]
 
 
-def deserialize_all(*args):
-    return [decode_if_needed(a) for a in args]
+def deserialize_all(anchor, message):
+    return decode_if_needed(anchor), json.loads(message)

lib/tfw/networking/zmq_websocket_handler.py

@@ -1,8 +1,8 @@
 import json
+
 from tornado.websocket import WebSocketHandler
 
 from .serialization import deserialize_all
-from ..util import parse_anchor_from_message
 from .event_handler_connector import EventHandlerConnector
 from ..config.logs import logging
 log = logging.getLogger(__name__)
@@ -27,12 +27,12 @@ class ZMQWebSocketHandler(WebSocketHandler):
 
     def on_message(self, message):
         log.debug('Received on WebSocket: {}'.format(message))
-        self.send_message(*self.make_response(message))
+        self.send_message(*self.make_response(json.loads(message)))
 
     def make_response(self, message):
         raise NotImplementedError
 
-    def send_message(self, message: str, anchor: str = None):
+    def send_message(self, message: dict, anchor: str = None):
         self._event_handler_connector.send_message(message, anchor)
 
     def on_close(self):
@@ -53,10 +53,10 @@ class FSMManagingSocketHandler(ZMQWebSocketHandler):
         self._event_handler_connector.send_message(message)
 
     def make_response(self, message):
-        self.fsm.trigger(parse_anchor_from_message(message), message=message)
+        self.fsm.trigger(message['anchor'], message=message)
         anchor = 'FSMUpdate'
         response = {'anchor': anchor,
                     'data': {'current_state': self.fsm.state,
                              'valid_transitions':
                                  [{'trigger': trigger} for trigger in self.fsm.machine.get_triggers()]}}
-        return json.dumps(response), anchor
+        return response, anchor