Make tfw.networking more structured, remove relative import hell

lib/tfw/networking/server/event_handler_connector.py

@@ -0,0 +1,44 @@
+import zmq
+from zmq.eventloop import ioloop
+from zmq.eventloop.zmqstream import ZMQStream
+
+from tfw.networking.serialization import serialize_all
+from tfw.config import PUBLISHER_PORT, RECEIVER_PORT
+from tfw.config.logs import logging
+log = logging.getLogger(__name__)
+from tfw.util import ZMQConnectorBase
+
+
+ioloop.install()
+
+
+class EventHandlerDownlinkConnector(ZMQConnectorBase):
+    def __init__(self, zmq_context=None):
+        super(EventHandlerDownlinkConnector, self).__init__(zmq_context)
+        self._zmq_pull_socket = self._zmq_context.socket(zmq.PULL)
+        self._zmq_pull_stream = ZMQStream(self._zmq_pull_socket)
+        address = 'tcp://*:{}'.format(RECEIVER_PORT)
+        self._zmq_pull_socket.bind(address)
+        log.debug('Pull socket bound to {}'.format(address))
+
+
+class EventHandlerUplinkConnector(ZMQConnectorBase):
+    def __init__(self, zmq_context=None):
+        super(EventHandlerUplinkConnector, self).__init__(zmq_context)
+        self._zmq_pub_socket = self._zmq_context.socket(zmq.PUB)
+        address = 'tcp://*:{}'.format(PUBLISHER_PORT)
+        self._zmq_pub_socket.bind(address)
+        log.debug('Pub socket bound to {}'.format(address))
+
+
+class EventHandlerConnector(EventHandlerDownlinkConnector, EventHandlerUplinkConnector):
+    def __init__(self, zmq_context=None):
+        super(EventHandlerConnector, self).__init__(zmq_context)
+
+    def register_callback(self, callback):
+        self._zmq_pull_stream.on_recv(callback)
+
+    def send_message(self, message: dict, anchor: str = None):
+        if not anchor:
+            anchor = message['anchor']
+        self._zmq_pub_socket.send_multipart(serialize_all(anchor, message))

lib/tfw/networking/server/zmq_websocket_handler.py

@@ -0,0 +1,62 @@
+import json
+from tornado.websocket import WebSocketHandler
+
+from tfw.networking.serialization import deserialize_all
+from tfw.networking.server.event_handler_connector import EventHandlerConnector
+from tfw.config.logs import logging
+log = logging.getLogger(__name__)
+
+
+class ZMQWebSocketHandler(WebSocketHandler):
+    _event_handler_connector = EventHandlerConnector()
+
+    def __init__(self, application, request, **kwargs):
+        super().__init__(application, request, **kwargs)
+
+    def open(self, *args, **kwargs):
+        log.debug('WebSocket connection initiated')
+        self._event_handler_connector.register_callback(self.zmq_callback)
+
+    def zmq_callback(self, msg_parts):
+        anchor, data = deserialize_all(*msg_parts)
+        log.debug('Received on pull socket: {}'.format(data))
+        self.write_message(data)
+
+    def on_message(self, message):
+        log.debug('Received on WebSocket: {}'.format(message))
+        self.send_message(*self.make_response(json.loads(message)))
+
+    def make_response(self, message):
+        raise NotImplementedError
+
+    def send_message(self, message: dict, anchor: str = None):
+        self._event_handler_connector.send_message(message, anchor)
+
+    def on_close(self):
+        pass
+
+    # much secure, very cors, wow
+    def check_origin(self, origin):
+        return True
+
+
+class FSMManagingSocketHandler(ZMQWebSocketHandler):
+    fsm = None
+
+    def __init__(self, FSM, application, request, **kwargs):
+        super().__init__(application, request, **kwargs)
+        if self.fsm is None:
+            FSMManagingSocketHandler.fsm = FSM()
+        self.fsm.subscribe_message_handler(self.handle_fsm_message)
+
+    def handle_fsm_message(self, message):
+        self._event_handler_connector.send_message(message)
+
+    def make_response(self, 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 response, anchor