Refactor project structure

lib/tfw/event_handler_base.py

@@ -0,0 +1,53 @@
+import json
+
+from .networking.server_connector import ServerConnector
+
+
+class EventHandlerBase:
+    def __init__(self, anchor):
+        self.server_connector = ServerConnector()
+        self.anchor = anchor
+        self.subscriptions = set()
+        self.subscribe(self.anchor)
+        self.subscribe('reset')
+        self.server_connector.register_callback(self.event_handler_callback)
+
+    def event_handler_callback(self, msg_parts):
+        anchor, message = msg_parts
+        data_json = json.loads(message)
+        response = self.handle_event(anchor, data_json) if anchor != b'reset' else self.handle_reset(data_json)
+        if response is None: return
+        encoded_response = json.dumps(response).encode('utf-8')
+        self.server_connector.send(anchor, encoded_response)
+
+    def handle_event(self, anchor, data_json):
+        raise NotImplementedError
+
+    def handle_reset(self, data_json):
+        return None
+
+    def message_other(self, anchor, data):
+        encoded_anchor = anchor.encode('utf-8')
+        message = {
+            'anchor': anchor,
+            'data': data
+        }
+        encoded_message = json.dumps(message).encode('utf-8')
+        self.server_connector.send(encoded_anchor, encoded_message)
+
+    def subscribe(self, anchor):
+        if anchor not in self.subscriptions:
+            self.subscriptions.add(anchor)
+            self.server_connector.subscribe(anchor)
+
+    def unsubscribe(self, anchor):
+        try:
+            self.subscriptions.remove(anchor)
+            self.server_connector.unsubscribe(anchor)
+        except KeyError:
+            pass
+
+    def unsubscribe_all(self):
+        for sub in self.subscriptions:
+            self.server_connector.unsubscribe(anchor=sub)
+        self.subscriptions.clear()

lib/tfw/fsm_base.py

@@ -0,0 +1,23 @@
+from transitions import Machine
+
+
+class FSMBase:
+    states, transitions = [], []
+
+    def __init__(self, initial: str = None):
+        self.message_handlers = []
+        self.machine = Machine(model=self,
+                               states=self.states,
+                               transitions=self.transitions,
+                               initial=initial or self.states[0],
+                               send_event=True,
+                               ignore_invalid_triggers=True,
+                               after_state_change='forward_message')
+
+    def forward_message(self, event_data):
+        message = event_data.kwargs.get('message')
+        for msghandler in self.message_handlers:
+            msghandler(message)
+
+    def subscribe_message_handler(self, msghandler):
+        self.message_handlers.append(msghandler)

lib/tfw/message_sender.py

@@ -0,0 +1,21 @@
+import json
+from datetime import datetime
+
+from .networking.server_connector import ServerUplinkConnector
+
+
+class MessageSender:
+    def __init__(self, custom_anchor=None):
+        self.server_connector = ServerUplinkConnector()
+        if isinstance(custom_anchor, str):
+            custom_anchor = custom_anchor.encode('utf-8')
+        self.anchor = custom_anchor or b'message'
+
+    def send(self, originator, message):
+        response = {
+            'originator': originator,
+            'timestamp': datetime.now().isoformat(),
+            'message': message
+        }
+        encoded_response = json.dumps(response).encode('utf-8')
+        self.server_connector.send(self.anchor, encoded_response)

lib/tfw/networking/event_handler_connector.py

@@ -0,0 +1,44 @@
+import zmq
+from zmq.eventloop import ioloop
+from zmq.eventloop.zmqstream import ZMQStream
+
+from ..config import PUBLISHER_PORT, RECEIVER_PORT
+from ..config.logs import logging
+log = logging.getLogger(__name__)
+from ..util import parse_anchor_from_message, 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: str, anchor: str = None):
+        if not anchor:
+            anchor = parse_anchor_from_message(message)
+        encoded_message = [part.encode('utf-8') for part in (anchor, message)]
+        self._zmq_pub_socket.send_multipart(encoded_message)

lib/tfw/networking/server_connector.py

@@ -0,0 +1,37 @@
+import zmq
+from functools import partial
+from zmq.eventloop import ioloop
+from zmq.eventloop.zmqstream import ZMQStream
+
+from ..config import PUBLISHER_PORT, RECEIVER_PORT
+from ..util import ZMQConnectorBase
+
+
+ioloop.install()
+
+
+class ServerDownlinkConnector(ZMQConnectorBase):
+    def __init__(self, zmq_context=None):
+        super(ServerDownlinkConnector, self).__init__(zmq_context)
+        self._zmq_sub_socket = self._zmq_context.socket(zmq.SUB)
+        self._zmq_sub_socket.connect('tcp://localhost:{}'.format(PUBLISHER_PORT))
+        self._zmq_sub_stream = ZMQStream(self._zmq_sub_socket)
+
+        self.subscribe = partial(self._zmq_sub_socket.setsockopt_string, zmq.SUBSCRIBE)
+        self.unsubscribe = partial(self._zmq_sub_socket.setsockopt_string, zmq.UNSUBSCRIBE)
+        self.register_callback = self._zmq_sub_stream.on_recv
+
+
+class ServerUplinkConnector(ZMQConnectorBase):
+    def __init__(self, zmq_context=None):
+        super(ServerUplinkConnector, self).__init__(zmq_context)
+        self._zmq_push_socket = self._zmq_context.socket(zmq.PUSH)
+        self._zmq_push_socket.connect('tcp://localhost:{}'.format(RECEIVER_PORT))
+
+    def send(self, anchor, response):
+        self._zmq_push_socket.send_multipart([anchor, response])
+
+
+class ServerConnector(ServerUplinkConnector, ServerDownlinkConnector):
+    def __init__(self, zmq_context=None):
+        super(ServerConnector, self).__init__(zmq_context)

lib/tfw/networking/zmq_websocket_handler.py

@@ -0,0 +1,61 @@
+import json
+from tornado.websocket import WebSocketHandler
+
+from ..util import parse_anchor_from_message
+from .event_handler_connector import EventHandlerConnector
+from ..config.logs import logging
+log = logging.getLogger(__name__)
+
+
+class ZMQWebSocketHandler(WebSocketHandler):
+    _event_handler_connector = EventHandlerConnector() #----.
+    # TODO: investigate thread safety!    <-----------------'
+    # ZMQ sockets shared between several WebSocketHandler instances
+    # http://pyzmq.readthedocs.io/en/latest/morethanbindings.html#thread-safety
+    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 = msg_parts
+        log.debug('Received on pull socket: {}'.format(data.decode()))
+        self.write_message(data.decode())
+
+    def on_message(self, message):
+        log.debug('Received on WebSocket: {}'.format(message))
+        self.send_message(*self.make_response(message))
+
+    def make_response(self, message):
+        raise NotImplementedError
+
+    def send_message(self, message: str, 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):
+    def __init__(self, FSM, application, request, **kwargs):
+        super().__init__(application, request, **kwargs)
+        self.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(parse_anchor_from_message(message), 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

lib/tfw/util.py

@@ -1,6 +1,6 @@
 import json, xmlrpc.client, zmq
 
-from config.envvars import SUPERVISOR_HTTP_URI
+from .config.envvars import SUPERVISOR_HTTP_URI
 
 
 def parse_anchor_from_message(message):