From 87fa86d3146e3e75462d28e9f8ca3ea51b5628e0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krist=C3=B3f=20T=C3=B3th?= Date: Tue, 10 Apr 2018 13:00:56 +0200 Subject: [PATCH 1/4] Refactor FSM triggering in TFWServer --- lib/tfw/networking/server/tfw_server.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/lib/tfw/networking/server/tfw_server.py b/lib/tfw/networking/server/tfw_server.py index d6e6889..b18dc06 100644 --- a/lib/tfw/networking/server/tfw_server.py +++ b/lib/tfw/networking/server/tfw_server.py @@ -38,13 +38,16 @@ class TFWServer: return self._fsm_manager def make_response(self, message): + self.trigger_fsm(message) + message['FSMUpdate'] = self._fsm_updater.get_fsm_state_and_transitions() + return message + + def trigger_fsm(self, message): trigger = message.get('trigger', '') try: self._fsm_manager.trigger(trigger, message) except AttributeError: LOG.debug('FSM failed to execute nonexistent trigger: "%s"', trigger) - message['FSMUpdate'] = self._fsm_updater.get_fsm_state_and_transitions() - return message def proxy_filter(self, message): # pylint: disable=unused-argument,no-self-use From 6113149c589d4aa7a4c81039f507d4bf8bb5b27c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krist=C3=B3f=20T=C3=B3th?= Date: Tue, 10 Apr 2018 13:42:51 +0200 Subject: [PATCH 2/4] Rework serialization module to work regardless of message format --- lib/tfw/networking/serialization.py | 29 +++++++++++++++++++++-------- 1 file changed, 21 insertions(+), 8 deletions(-) diff --git a/lib/tfw/networking/serialization.py b/lib/tfw/networking/serialization.py index 26e222e..d26257e 100644 --- a/lib/tfw/networking/serialization.py +++ b/lib/tfw/networking/serialization.py @@ -4,6 +4,27 @@ import json +def serialize_all(*args): + return tuple(_serialize_single(arg) for arg in args) + + +def deserialize_all(*args): + return tuple(_deserialize_single(arg) for arg in args) + + +def _serialize_single(data): + if not isinstance(data, str): + data = json.dumps(data) + return encode_if_needed(data) + + +def _deserialize_single(data): + try: + return json.loads(data) + except ValueError: + return decode_if_needed(data) + + def encode_if_needed(value): if isinstance(value, str): value = value.encode('utf-8') @@ -14,11 +35,3 @@ def decode_if_needed(value): if isinstance(value, (bytes, bytearray)): value = value.decode('utf-8') return value - - -def serialize_all(key, data): - return [encode_if_needed(frame) for frame in (key, json.dumps(data))] - - -def deserialize_all(key, data): - return decode_if_needed(key), json.loads(data) From 4842574cf3ed69a67ed0e84c7d201d6accb3139c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krist=C3=B3f=20T=C3=B3th?= Date: Wed, 11 Apr 2018 11:03:59 +0200 Subject: [PATCH 3/4] Refactor LinearFSM triggers to have different names --- lib/tfw/linear_fsm.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/tfw/linear_fsm.py b/lib/tfw/linear_fsm.py index 2725ff8..574f5c9 100644 --- a/lib/tfw/linear_fsm.py +++ b/lib/tfw/linear_fsm.py @@ -7,6 +7,6 @@ from .fsm_base import FSMBase class LinearFSM(FSMBase): def __init__(self, number_of_steps): self.states = list(map(str, range(number_of_steps))) - self.transitions = [{'trigger': 'step', 'source': index, 'dest': str(int(index)+1)} + self.transitions = [{'trigger': 'step_{}'.format(int(index)+1), 'source': index, 'dest': str(int(index)+1)} for index in self.states[:-1]] super(LinearFSM, self).__init__() From 73fa65c6e88669d4e34947820b930dce3dbce99d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krist=C3=B3f=20T=C3=B3th?= Date: Wed, 11 Apr 2018 15:22:54 +0200 Subject: [PATCH 4/4] Implement triggering of FSM from ServerUplinkConnector --- lib/tfw/networking/server/tfw_server.py | 5 +++++ .../networking/server/zmq_websocket_handler.py | 17 +++++++++++++---- 2 files changed, 18 insertions(+), 4 deletions(-) diff --git a/lib/tfw/networking/server/tfw_server.py b/lib/tfw/networking/server/tfw_server.py index b18dc06..63ddf72 100644 --- a/lib/tfw/networking/server/tfw_server.py +++ b/lib/tfw/networking/server/tfw_server.py @@ -25,6 +25,7 @@ class TFWServer: self.application = Application( [(r'/ws', ZMQWebSocketProxy, {'make_response': self.make_response, 'proxy_filter': self.proxy_filter, + 'handle_trigger': self.handle_trigger, 'event_handler_connector': self._event_handler_connector})] ) #self.controller_responder = ControllerResponder(self.fsm) TODO: add this once controller stuff is resolved @@ -42,6 +43,10 @@ class TFWServer: message['FSMUpdate'] = self._fsm_updater.get_fsm_state_and_transitions() return message + def handle_trigger(self, message): + LOG.debug('Executing handler for trigger "%s"', message.get('trigger', '')) + self.trigger_fsm(message) + def trigger_fsm(self, message): trigger = message.get('trigger', '') try: diff --git a/lib/tfw/networking/server/zmq_websocket_handler.py b/lib/tfw/networking/server/zmq_websocket_handler.py index 80fd95a..73a9754 100644 --- a/lib/tfw/networking/server/zmq_websocket_handler.py +++ b/lib/tfw/networking/server/zmq_websocket_handler.py @@ -31,14 +31,15 @@ class ZMQWebSocketHandler(WebSocketHandler, ABC): def zmq_callback(self, msg_parts): keyhandlers = {'mirror': self.mirror} - key, data = deserialize_all(*msg_parts) - LOG.debug('Received on pull socket: %s', data) + key, message = deserialize_all(*msg_parts) + LOG.debug('Received on pull socket: %s', message) + self.handle_trigger(message) if key not in keyhandlers: for instance in ZMQWebSocketHandler.instances: - instance.write_message(data) + instance.write_message(message) else: try: - keyhandlers[key](data['data']) + keyhandlers[key](message['data']) except KeyError: LOG.error('Invalid mirror message format! Ignoring.') @@ -57,6 +58,10 @@ class ZMQWebSocketHandler(WebSocketHandler, ABC): def send_message(self, message: dict, key: str = None): self._event_handler_connector.send_message(message, key) + @abstractmethod + def handle_trigger(self, message): + raise NotImplementedError + # much secure, very cors, wow def check_origin(self, origin): return True @@ -68,6 +73,7 @@ class ZMQWebSocketProxy(ZMQWebSocketHandler): super(ZMQWebSocketProxy, self).initialize(**kwargs) self._make_response = kwargs['make_response'] self._proxy_filter = kwargs['proxy_filter'] + self._handle_trigger = kwargs['handle_trigger'] def on_message(self, message): message = json.loads(message) @@ -76,3 +82,6 @@ class ZMQWebSocketProxy(ZMQWebSocketHandler): def make_response(self, message): return self._make_response(message) + + def handle_trigger(self, message): + self._handle_trigger(message)