baseimage-tutorial-framework/lib/tfw/networking/server/tfw_server.py

128 lines
4.0 KiB
Python

# Copyright (C) 2018 Avatao.com Innovative Learning Kft.
# All Rights Reserved. See LICENSE file for details.
from collections import defaultdict
from tornado.web import Application
from tfw.networking import MessageSender
from tfw.networking.event_handlers import ServerUplinkConnector
from tfw.networking.server import EventHandlerConnector
from tfw.config.logs import logging
from .zmq_websocket_proxy import ZMQWebSocketProxy
LOG = logging.getLogger(__name__)
class TFWServer:
"""
This class handles the proxying of messages between the frontend and event handers.
It proxies messages from the "/ws" route to all event handlers subscribed to a ZMQ
SUB socket. It also manages an FSM you can define as a constructor argument.
"""
def __init__(self, fsm_type):
"""
:param fsm_type: the type of FSM you want TFW to use
"""
self._fsm = fsm_type()
self._fsm_updater = FSMUpdater(self._fsm)
self._fsm_manager = FSMManager(self._fsm)
self._fsm.subscribe_callback(self._fsm_updater.update)
self._event_handler_connector = EventHandlerConnector()
self.application = Application([(
r'/ws', ZMQWebSocketProxy,{
'event_handler_connector': self._event_handler_connector,
'message_handlers': [self.append_fsm_data, self.handle_trigger]
})]
)
# self.controller_responder = ControllerResponder(self.fsm)
# TODO: add this once controller stuff is resolved
@property
def fsm(self):
return self._fsm
@property
def fsm_manager(self):
return self._fsm_manager
def append_fsm_data(self, message):
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:
self._fsm_manager.trigger(trigger, message)
except AttributeError:
LOG.debug('FSM failed to execute nonexistent trigger: "%s"', trigger)
def listen(self, port):
self.application.listen(port)
class FSMManager:
def __init__(self, fsm):
self._fsm = fsm
self.trigger_predicates = defaultdict(list)
self.messenge_sender = MessageSender()
@property
def fsm(self):
return self._fsm
def trigger(self, trigger, message):
predicate_results = []
for predicate in self.trigger_predicates[trigger]:
success, message = predicate(message)
predicate_results.append(success)
self.messenge_sender.send('FSM', message)
if all(predicate_results):
try:
self.fsm.trigger(trigger, message=message)
except AttributeError:
LOG.debug('FSM failed to execute nonexistent trigger: "%s"', trigger)
def subscribe_predicate(self, trigger, *predicates):
self.trigger_predicates[trigger].extend(predicates)
def unsubscribe_predicate(self, trigger, *predicates):
self.trigger_predicates[trigger] = [
predicate
for predicate in self.trigger_predicates[trigger]
not in predicates
]
class FSMUpdater:
def __init__(self, fsm):
self.fsm = fsm
self.uplink = ServerUplinkConnector()
def update(self, kwargs_dict):
# pylint: disable=unused-argument
self.uplink.send(self.generate_fsm_update())
def generate_fsm_update(self):
return {
'key': 'FSMUpdate',
'data': self.get_fsm_state_and_transitions()
}
def get_fsm_state_and_transitions(self):
state = self.fsm.state
valid_transitions = [
{'trigger': trigger}
for trigger in self.fsm.machine.get_triggers(self.fsm.state)
]
return {
'current_state': state,
'valid_transitions': valid_transitions
}