mrtoth/baseimage-tutorial-framework
1
0
Fork 0
mirror of https://github.com/avatao-content/baseimage-tutorial-framework synced 2024-11-22 07:11:32 +00:00
Code Issues Releases Wiki Activity

Rework whole TFW networking model

Browse Source
This commit is contained in:
Kristóf Tóth 2019-05-26 18:26:33 +02:00
parent 613919a5b6
commit 01d9003501
19 changed files with 155 additions and 279 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
lib/tfw/components/__init__.py
View File

@ -14,3 +14,4 @@ from .pipe_io_event_handler import PipeIOEventHandlerBase, PipeIOEventHandler, P
from .pipe_io_event_handler import TransformerPipeIOEventHandler, CommandEventHandler
from .directory_snapshotting_event_handler import DirectorySnapshottingEventHandler
from .commands_equal import CommandsEqual
from .frontend_event_handler import FrontendEventHandler

6
lib/tfw/components/directory_monitor.py
View File

@ -5,7 +5,7 @@ from functools import wraps
from watchdog.events import FileSystemEventHandler as FileSystemWatchdogEventHandler
from tfw.networking.event_handlers.server_connector import ServerUplinkConnector
from tfw.networking.event_handlers.server_connector import ServerUplinkConnector, Scope
from tfw.decorators.rate_limiter import RateLimiter
from tfw.mixins.observer_mixin import ObserverMixin
@ -65,10 +65,10 @@ class IdeReloadWatchdogEventHandler(FileSystemWatchdogEventHandler):
self.ignore = self.ignore - 1
return
LOG.debug(event)
self.uplink.send({
self.uplink.send_message({
'key': self.ide_key,
'data': {'command': 'reload'}
})
}, Scope.WEBSOCKET)
def with_monitor_paused(fun):

4
lib/tfw/components/directory_monitoring_event_handler.py
View File

@ -3,7 +3,7 @@
from os.path import isdir, exists
from tfw.event_handler_base import EventHandlerBase
from tfw.event_handler_base import FrontendEventHandlerBase
from tfw.mixins.monitor_manager_mixin import MonitorManagerMixin
from tfw.components.directory_monitor import DirectoryMonitor
from tfw.config.logs import logging
@ -11,7 +11,7 @@ from tfw.config.logs import logging
LOG = logging.getLogger(__name__)
class DirectoryMonitoringEventHandler(EventHandlerBase, MonitorManagerMixin):
class DirectoryMonitoringEventHandler(FrontendEventHandlerBase, MonitorManagerMixin):
def __init__(self, key, directory):
super().__init__(key)
self._directory = directory

4
lib/tfw/components/directory_snapshotting_event_handler.py
View File

@ -8,7 +8,7 @@ from datetime import datetime
from dateutil import parser as dateparser
from tfw.event_handler_base import EventHandlerBase
from tfw.event_handler_base import FrontendEventHandlerBase
from tfw.components.snapshot_provider import SnapshotProvider
from tfw.config import TFWENV
from tfw.config.logs import logging
@ -16,7 +16,7 @@ from tfw.config.logs import logging
LOG = logging.getLogger(__name__)
class DirectorySnapshottingEventHandler(EventHandlerBase):
class DirectorySnapshottingEventHandler(FrontendEventHandlerBase):
def __init__(self, key, directories, exclude_unix_patterns=None):
super().__init__(key)
self.snapshot_providers = {}

62
lib/tfw/components/frontend_event_handler.py Normal file
View File

@ -0,0 +1,62 @@
from abc import ABC, abstractmethod
from contextlib import suppress
from tfw.networking.message_sender import MessageSender
from tfw.event_handler_base import FrontendEventHandlerBase
class FrontendEventHandler(FrontendEventHandlerBase):
def __init__(self):
frontend_keys = ('message', 'queueMessages', 'dashboard', 'console')
self._frontend_message_storage = FrontendMessageStorage(frontend_keys)
super().__init__((*frontend_keys, 'recover'))
def handle_event(self, message):
self._frontend_message_storage.save_message(message)
if message['key'] == 'recover':
self.recover_frontend()
return message
def recover_frontend(self):
for message in self._frontend_message_storage.messages:
self.send_message(message)
class MessageStorage(ABC):
def __init__(self):
self._messages = []
def save_message(self, message):
with suppress(KeyError, AttributeError):
if self._filter_message(message):
self._messages.extend(self._transform_message(message))
@abstractmethod
def _filter_message(self, message):
raise NotImplementedError
def _transform_message(self, message): # pylint: disable=no-self-use
yield message
def clear(self):
self._messages.clear()
@property
def messages(self):
yield from self._messages
class FrontendMessageStorage(MessageStorage):
def __init__(self, keys):
self._keys = keys
super().__init__()
def _filter_message(self, message):
key = message['key']
return key in self._keys
def _transform_message(self, message):
if message['key'] == 'queueMessages':
yield from MessageSender.generate_messages_from_queue(message)
else:
yield message

7
lib/tfw/components/fsm_managing_event_handler.py
View File

@ -1,14 +1,15 @@
# Copyright (C) 2018 Avatao.com Innovative Learning Kft.
# All Rights Reserved. See LICENSE file for details.
from tfw.event_handler_base import EventHandlerBase
from tfw.event_handler_base import FrontendEventHandlerBase
from tfw.crypto import KeyManager, sign_message, verify_message
from tfw.config.logs import logging
from tfw.networking.event_handlers.server_connector import Scope
LOG = logging.getLogger(__name__)
class FSMManagingEventHandler(EventHandlerBase):
class FSMManagingEventHandler(FrontendEventHandlerBase):
"""
EventHandler responsible for managing the state machine of
the framework (TFW FSM).
@ -42,7 +43,7 @@ class FSMManagingEventHandler(EventHandlerBase):
fsm_update_message = self._fsm_updater.fsm_update
sign_message(self.auth_key, message)
sign_message(self.auth_key, fsm_update_message)
self.server_connector.broadcast(fsm_update_message)
self.server_connector.send_message(fsm_update_message, Scope.BROADCAST)
return message
except KeyError:
LOG.error('IGNORING MESSAGE: Invalid message received: %s', message)

4
lib/tfw/components/ide_event_handler.py
View File

@ -6,7 +6,7 @@ from glob import glob
from fnmatch import fnmatchcase
from typing import Iterable
from tfw.event_handler_base import EventHandlerBase
from tfw.event_handler_base import FrontendEventHandlerBase
from tfw.mixins.monitor_manager_mixin import MonitorManagerMixin
from tfw.components.directory_monitor import DirectoryMonitor
from tfw.config.logs import logging
@ -102,7 +102,7 @@ class FileManager: # pylint: disable=too-many-instance-attributes
return relpath(self._filepath(filename), start=self._workdir)
class IdeEventHandler(EventHandlerBase, MonitorManagerMixin):
class IdeEventHandler(FrontendEventHandlerBase, MonitorManagerMixin):
# pylint: disable=too-many-arguments,anomalous-backslash-in-string
"""
Event handler implementing the backend of our browser based IDE.

6
lib/tfw/components/log_monitor.py
View File

@ -6,7 +6,7 @@ from os.path import dirname
from watchdog.events import PatternMatchingEventHandler as PatternMatchingWatchdogEventHandler
from tfw.networking.event_handlers.server_connector import ServerUplinkConnector
from tfw.networking.event_handlers.server_connector import ServerUplinkConnector, Scope
from tfw.decorators.rate_limiter import RateLimiter
from tfw.mixins.observer_mixin import ObserverMixin
from tfw.mixins.supervisor_mixin import SupervisorLogMixin
@ -47,11 +47,11 @@ class SendLogWatchdogEventHandler(PatternMatchingWatchdogEventHandler, Superviso
@RateLimiter(rate_per_second=5)
def on_modified(self, event):
self.uplink.send({
self.uplink.send_message({
'key': 'processlog',
'data': {
'command': 'new_log',
'stdout': self.read_stdout(self.process_name, tail=self.log_tail),
'stderr': self.read_stderr(self.process_name, tail=self.log_tail)
}
})
}, Scope.BROADCAST)

4
lib/tfw/components/log_monitoring_event_handler.py
View File

@ -1,7 +1,7 @@
# Copyright (C) 2018 Avatao.com Innovative Learning Kft.
# All Rights Reserved. See LICENSE file for details.
from tfw.event_handler_base import EventHandlerBase
from tfw.event_handler_base import FrontendEventHandlerBase
from tfw.mixins.monitor_manager_mixin import MonitorManagerMixin
from tfw.components.log_monitor import LogMonitor
from tfw.config.logs import logging
@ -9,7 +9,7 @@ from tfw.config.logs import logging
LOG = logging.getLogger(__name__)
class LogMonitoringEventHandler(EventHandlerBase, MonitorManagerMixin):
class LogMonitoringEventHandler(FrontendEventHandlerBase, MonitorManagerMixin):
"""
Monitors the output of a supervisor process (stdout, stderr) and
sends the results to the frontend.

4
lib/tfw/components/pipe_io_event_handler.py
View File

@ -56,7 +56,7 @@ class PipeIOEventHandler(PipeIOEventHandlerBase):
def handle_pipe_event(self, message_bytes):
json = loads(message_bytes)
self.server_connector.send(json)
self.send_message(json)
class TransformerPipeIOEventHandler(PipeIOEventHandlerBase):
@ -93,7 +93,7 @@ class TransformerPipeIOEventHandler(PipeIOEventHandlerBase):
transformed_bytes = self._transform_in(message_bytes)
if transformed_bytes:
json_message = loads(transformed_bytes)
self.server_connector.send(json_message)
self.send_message(json_message)
class CommandEventHandler(PipeIOEventHandler):

4
lib/tfw/components/process_managing_event_handler.py
View File

@ -3,7 +3,7 @@
from xmlrpc.client import Fault as SupervisorFault
from tfw.event_handler_base import EventHandlerBase
from tfw.event_handler_base import FrontendEventHandlerBase
from tfw.mixins.supervisor_mixin import SupervisorMixin, SupervisorLogMixin
from tfw.components.directory_monitor import with_monitor_paused
from tfw.config.logs import logging
@ -23,7 +23,7 @@ class ProcessManager(SupervisorMixin, SupervisorLogMixin):
return self.commands[command](process_name)
class ProcessManagingEventHandler(EventHandlerBase):
class ProcessManagingEventHandler(FrontendEventHandlerBase):
"""
Event handler that can manage processes managed by supervisor.

4
lib/tfw/components/terminal_event_handler.py
View File

@ -1,7 +1,7 @@
# Copyright (C) 2018 Avatao.com Innovative Learning Kft.
# All Rights Reserved. See LICENSE file for details.
from tfw.event_handler_base import EventHandlerBase
from tfw.event_handler_base import FrontendEventHandlerBase
from tfw.components.terminado_mini_server import TerminadoMiniServer
from tfw.config import TFWENV
from tfw.config.logs import logging
@ -10,7 +10,7 @@ from tao.config import TAOENV
LOG = logging.getLogger(__name__)
class TerminalEventHandler(EventHandlerBase):
class TerminalEventHandler(FrontendEventHandlerBase):
"""
Event handler responsible for managing terminal sessions for frontend xterm
sessions to connect to. You need to instanciate this in order for frontend

2
lib/tfw/event_handler_base/__init__.py
View File

@ -1,6 +1,6 @@
# Copyright (C) 2018 Avatao.com Innovative Learning Kft.
# All Rights Reserved. See LICENSE file for details.
from .event_handler_base import EventHandlerBase
from .event_handler_base import EventHandlerBase, FrontendEventHandlerBase
from .boradcasting_event_handler import BroadcastingEventHandler
from .fsm_aware_event_handler import FSMAwareEventHandler

3
lib/tfw/event_handler_base/boradcasting_event_handler.py
View File

@ -4,6 +4,7 @@
from abc import ABC
from tfw.event_handler_base.event_handler_base import EventHandlerBase
from tfw.networking.event_handlers.server_connector import Scope
from tfw.crypto import message_checksum
@ -27,4 +28,4 @@ class BroadcastingEventHandler(EventHandlerBase, ABC):
response = self.dispatch_handling(message)
if response:
self.own_message_hashes.append(message_checksum(response))
self.server_connector.broadcast(response)
self.server_connector.send_message(response, Scope.BROADCAST)

12
lib/tfw/event_handler_base/event_handler_base.py
View File

@ -5,7 +5,7 @@ from abc import ABC, abstractmethod
from inspect import currentframe
from typing import Iterable
from tfw.networking.event_handlers.server_connector import ServerConnector
from tfw.networking.event_handlers.server_connector import ServerConnector, Scope
from tfw.config.logs import logging
LOG = logging.getLogger(__name__)
@ -48,7 +48,10 @@ class EventHandlerBase(ABC):
response = self.dispatch_handling(message)
if response:
self.server_connector.send(response)
self.send_message(response)
def send_message(self, message):
self.server_connector.send_message(message)
def check_key(self, message):
"""
@ -128,3 +131,8 @@ class EventHandlerBase(ABC):
instance for instance in locals_values
if isinstance(instance, cls)
}
class FrontendEventHandlerBase(EventHandlerBase): # pylint: disable=abstract-method
def send_message(self, message):
self.server_connector.send_message(message, Scope.WEBSOCKET)

46
lib/tfw/networking/event_handlers/server_connector.py
View File

@ -2,6 +2,7 @@
# All Rights Reserved. See LICENSE file for details.
from functools import partial
from enum import Enum
import zmq
from zmq.eventloop.zmqstream import ZMQStream
@ -33,52 +34,23 @@ class ServerDownlinkConnector(ZMQConnectorBase):
self._zmq_sub_stream.close()
class Scope(Enum):
ZMQ = 'zmq'
WEBSOCKET = 'websocket'
BROADCAST = 'broadcast'
class ServerUplinkConnector(ZMQConnectorBase):
"""
Class capable of sending messages to the TFW server and event handlers.
"""
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(f'tcp://localhost:{TFWENV.RECEIVER_PORT}')
self._zmq_push_socket.setsockopt(zmq.SNDHWM, 0)
def send_to_eventhandler(self, message):
"""
Send a message to an event handler through the TFW server.
This envelopes the desired message in the 'data' field of the message to
TFWServer, which will mirror it to event handlers.
:param message: JSON message you want to send
"""
self.send({
'key': 'mirror',
'data': message
})
def send(self, message):
"""
Send a message to the frontend through the TFW server.
:param message: JSON message you want to send
"""
def send_message(self, message, scope=Scope.ZMQ):
message['scope'] = scope.value
self._zmq_push_socket.send_multipart(serialize_tfw_msg(message))
def broadcast(self, message):
"""
Broadast a message through the TFW server.
This envelopes the desired message in the 'data' field of the message to
TFWServer, which will broadast it.
:param message: JSON message you want to send
"""
self.send({
'key': 'broadcast',
'data': message
})
def close(self):
self._zmq_push_socket.close()

32
lib/tfw/networking/message_sender.py
View File

@ -19,14 +19,14 @@ class MessageSender:
:param originator: name of sender to be displayed on the frontend
:param message: message to send
"""
data = {
'originator': originator,
'message': message
}
self.server_connector.send({
message = {
'key': self.key,
'data': data
})
'data': {
'originator': originator,
'message': message
}
}
self.server_connector.send_message(message)
def queue_messages(self, originator, messages):
"""
@ -34,16 +34,16 @@ class MessageSender:
:param originator: name of sender to be displayed on the frontend
:param messages: list of messages to queue
"""
data = {
'messages': [
{'message': message, 'originator': originator}
for message in messages
]
}
self.server_connector.send({
message = {
'key': self.queue_key,
'data': data
})
'data': {
'messages': [
{'message': message, 'originator': originator}
for message in messages
]
}
}
self.server_connector.send_message(message)
@staticmethod
def generate_messages_from_queue(queue_message):

88
lib/tfw/networking/server/tfw_server.py
View File

@ -1,114 +1,28 @@
# Copyright (C) 2018 Avatao.com Innovative Learning Kft.
# All Rights Reserved. See LICENSE file for details.
from abc import ABC, abstractmethod
from contextlib import suppress
from tornado.web import Application
from tfw.networking.server.zmq_websocket_proxy import ZMQWebSocketProxy
from tfw.networking.event_handlers.server_connector import ServerUplinkConnector
from tfw.networking.server.event_handler_connector import EventHandlerConnector
from tfw.networking.message_sender import MessageSender
from tfw.networking.fsm_aware import FSMAware
from tfw.crypto import KeyManager, verify_message, sign_message
from tfw.config.logs import logging
LOG = logging.getLogger(__name__)
class TFWServer(FSMAware):
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.
"""
def __init__(self):
super().__init__()
self._event_handler_connector = EventHandlerConnector()
self._uplink_connector = ServerUplinkConnector()
self._auth_key = KeyManager().auth_key
self.application = Application([(
r'/ws', ZMQWebSocketProxy, {
'event_handler_connector': self._event_handler_connector,
'proxy_filters_and_callbacks': {
'message_handlers': [
self.handle_trigger,
self.handle_recover,
self.handle_fsm_update
],
'frontend_message_handlers': [self.save_frontend_messages]
}
}
)])
self._frontend_messages = FrontendMessageStorage()
def handle_trigger(self, message):
if 'trigger' in message:
LOG.debug('Executing handler for trigger "%s"', message.get('trigger', ''))
fsm_eh_command = {
'key': 'fsm',
'data': {
'command': 'trigger',
'value': message['trigger']
}
}
if verify_message(self._auth_key, message):
sign_message(self._auth_key, fsm_eh_command)
self._uplink_connector.send_to_eventhandler(fsm_eh_command)
def handle_recover(self, message):
if message['key'] == 'recover':
self._frontend_messages.replay_messages(self._uplink_connector)
self._frontend_messages.clear()
def handle_fsm_update(self, message):
self.update_fsm_data(message)
def save_frontend_messages(self, message):
self._frontend_messages.save_message(message)
def listen(self, port):
self.application.listen(port)
class MessageStorage(ABC):
def __init__(self):
self.saved_messages = []
def save_message(self, message):
with suppress(KeyError, AttributeError):
if self.filter_message(message):
self.saved_messages.extend(self.transform_message(message))
@abstractmethod
def filter_message(self, message):
raise NotImplementedError
def transform_message(self, message): # pylint: disable=no-self-use
yield message
def clear(self):
self.saved_messages.clear()
class FrontendMessageStorage(MessageStorage):
def filter_message(self, message):
key = message['key']
command = message.get('data', {}).get('command')
return (
key in ('message', 'dashboard', 'queueMessages')
or key == 'ide' and command in ('select', 'read')
)
def transform_message(self, message):
if message['key'] == 'queueMessages':
yield from MessageSender.generate_messages_from_queue(message)
else:
yield message
def replay_messages(self, connector):
for message in self.saved_messages:
connector.send(message)

141
lib/tfw/networking/server/zmq_websocket_proxy.py
View File

@ -5,7 +5,7 @@ import json
from tornado.websocket import WebSocketHandler
from tfw.mixins.callback_mixin import CallbackMixin
from tfw.networking.event_handlers.server_connector import Scope
from tfw.config.logs import logging
LOG = logging.getLogger(__name__)
@ -14,38 +14,18 @@ LOG = logging.getLogger(__name__)
class ZMQWebSocketProxy(WebSocketHandler):
# pylint: disable=abstract-method
instances = set()
sequence_number = 0
def initialize(self, **kwargs): # pylint: disable=arguments-differ
self._event_handler_connector = kwargs['event_handler_connector']
self._proxy_filters_and_callbacks = kwargs.get('proxy_filters_and_callbacks', {})
self.event_handler_connector = kwargs['event_handler_connector']
self.tfw_router = TFWRouter(self.send_to_zmq, self.send_to_websockets)
self.proxy_eventhandler_to_websocket = TFWProxy(
self.send_eventhandler_message,
self.send_websocket_message
)
self.proxy_websocket_to_eventhandler = TFWProxy(
self.send_websocket_message,
self.send_eventhandler_message
)
def send_to_zmq(self, message):
self.event_handler_connector.send_message(message)
self.subscribe_proxy_callbacks()
def subscribe_proxy_callbacks(self):
eventhandler_message_handlers = self._proxy_filters_and_callbacks.get('eventhandler_message_handlers', [])
frontend_message_handlers = self._proxy_filters_and_callbacks.get('frontend_message_handlers', [])
message_handlers = self._proxy_filters_and_callbacks.get('message_handlers', [])
proxy_filters = self._proxy_filters_and_callbacks.get('proxy_filters', [])
self.proxy_websocket_to_eventhandler.subscribe_proxy_callbacks_and_filters(
eventhandler_message_handlers + message_handlers,
proxy_filters
)
self.proxy_eventhandler_to_websocket.subscribe_proxy_callbacks_and_filters(
frontend_message_handlers + message_handlers,
proxy_filters
)
@staticmethod
def send_to_websockets(message):
for instance in ZMQWebSocketProxy.instances:
instance.write_message(message)
def prepare(self):
ZMQWebSocketProxy.instances.add(self)
@ -54,102 +34,39 @@ class ZMQWebSocketProxy(WebSocketHandler):
ZMQWebSocketProxy.instances.remove(self)
def open(self, *args, **kwargs):
LOG.debug('WebSocket connection initiated')
self._event_handler_connector.register_callback(self.eventhander_callback)
LOG.debug('WebSocket connection initiated!')
self.event_handler_connector.register_callback(self.zmq_callback)
def eventhander_callback(self, message):
"""
Invoked on ZMQ messages from event handlers.
"""
self.sequence_message(message)
LOG.debug('Received on pull socket: %s', message)
self.proxy_eventhandler_to_websocket(message)
@classmethod
def sequence_message(cls, message):
cls.sequence_number += 1
message['seq'] = cls.sequence_number
def zmq_callback(self, message):
LOG.debug('Received on ZMQ pull socket: %s', message)
self.tfw_router.route(message)
def on_message(self, message):
"""
Invoked on WS messages from frontend.
"""
message = json.loads(message)
self.sequence_message(message)
LOG.debug('Received on WebSocket: %s', message)
self.proxy_websocket_to_eventhandler(message)
def send_eventhandler_message(self, message):
self._event_handler_connector.send_message(message)
@staticmethod
def send_websocket_message(message):
for instance in ZMQWebSocketProxy.instances:
instance.write_message(message)
self.tfw_router.route(message)
# much secure, very cors, wow
def check_origin(self, origin):
return True
class TFWProxy:
# pylint: disable=protected-access
def __init__(self, to_source, to_destination):
self.to_source = to_source
self.to_destination = to_destination
class TFWRouter:
def __init__(self, send_to_zmq, send_to_websockets):
self.send_to_zmq = send_to_zmq
self.send_to_websockets = send_to_websockets
self.proxy_filters = CallbackMixin()
self.proxy_callbacks = CallbackMixin()
def route(self, message):
scope = Scope(message.get('scope', 'zmq'))
self.proxy_filters.subscribe_callback(self.validate_message)
self.keyhandlers = {
'mirror': self.mirror,
'broadcast': self.broadcast
routing_table = {
Scope.ZMQ: self.send_to_zmq,
Scope.WEBSOCKET: self.send_to_websockets,
Scope.BROADCAST: self.broadcast
}
@staticmethod
def validate_message(message):
if 'key' not in message:
raise ValueError('Invalid TFW message format!')
def __call__(self, message):
if not self.filter_and_execute_callbacks(message):
return
if message['key'] not in self.keyhandlers:
self.to_destination(message)
else:
handler = self.keyhandlers[message['key']]
try:
handler(message)
except KeyError:
LOG.error('Invalid "%s" message format! Ignoring.', handler.__name__)
def filter_and_execute_callbacks(self, message):
try:
self.proxy_filters._execute_callbacks(message)
self.proxy_callbacks._execute_callbacks(message)
return True
except ValueError:
LOG.exception('Invalid TFW message received!')
return False
def mirror(self, message):
message = message['data']
if not self.filter_and_execute_callbacks(message):
return
LOG.debug('Mirroring message: %s', message)
self.to_source(message)
action = routing_table[scope]
action(message)
def broadcast(self, message):
message = message['data']
if not self.filter_and_execute_callbacks(message):
return
LOG.debug('Broadcasting message: %s', message)
self.to_source(message)
self.to_destination(message)
def subscribe_proxy_callbacks_and_filters(self, proxy_callbacks, proxy_filters):
self.proxy_callbacks.subscribe_callbacks(*proxy_callbacks)
self.proxy_filters.subscribe_callbacks(*proxy_filters)
self.send_to_zmq(message)
self.send_to_websockets(message)