Refactor MessageSender

Dockerfile

@@ -1,9 +1,14 @@
-FROM avatao/frontend-tutorial-framework:chausie-20191002 as frontend
+ARG FRONTEND_VERSION
+FROM avatao/frontend-tutorial-framework:${FRONTEND_VERSION} as frontend
+
 FROM avatao/debian:buster
+ARG FRONTEND_VERSION
+LABEL tfw.frontend.version=${FRONTEND_VERSION}
 
 RUN apt-get update                           &&\
     apt-get install -y --no-install-recommends \
         supervisor                             \
+        ncat                                   \
         libzmq5                                \
         nginx                                  \
         jq                                     \
@@ -22,7 +27,8 @@ RUN curl -Ls https://github.com/krallin/tini/releases/download/v0.18.0/tini-amd6
 ENV TFW_PUBLIC_PORT=8888            \
     TFW_WEB_PORT=4242               \
     TFW_LOGIN_APP_PORT=6666         \
-    TFW_TERMINADO_PORT=7878       \
+    TFW_TERMINAL_PORT=7878          \
+    TFW_TERMINAL_FALLBACK_PORT=7879 \
     TFW_SUPERVISOR_HTTP_PORT=9001   \
     TFW_PUB_PORT=7654               \
     TFW_PULL_PORT=8765
@@ -67,7 +73,7 @@ ONBUILD COPY ${BUILD_CONTEXT}/supervisor/ ${TFW_SUPERVISORD_COMPONENTS}
 ONBUILD RUN for f in "${TFW_NGINX_DEFAULT}" ${TFW_NGINX_COMPONENTS}/*.conf; do                               \
                 envsubst "$(printenv | cut -d= -f1 | grep TFW_ | sed -e 's/^/$/g')" < $f > $f~ && mv $f~ $f ;\
             done
-ONBUILD VOLUME ["/etc/nginx", "/var/lib/nginx", "/var/log/nginx", "${TFW_LIB_DIR}/tfw"]
+ONBUILD VOLUME ["/etc/nginx", "/etc/supervisor", "/var/lib/nginx", "/var/log/nginx", "${TFW_LIB_DIR}/tfw"]
 
 ENTRYPOINT ["/bin/init", "--"]
 CMD exec supervisord --nodaemon --configuration ${TFW_SUPERVISORD_CONF}

nginx/components/terminal.conf

@@ -1,5 +1,5 @@
 location = /terminal {
-        proxy_pass http://127.0.0.1:${TFW_TERMINADO_PORT};
+        proxy_pass http://terminal;
         proxy_http_version 1.1;
         proxy_set_header Upgrade $http_upgrade;
         proxy_set_header Connection "upgrade";

nginx/default.conf

@@ -1,3 +1,8 @@
+upstream terminal {
+    server 127.0.0.1:${TFW_TERMINAL_PORT};
+    server 127.0.0.1:${TFW_TERMINAL_FALLBACK_PORT} backup;
+}
+
 server {
     listen ${TFW_PUBLIC_PORT};
     server_name localhost;

setup.py

@@ -1,6 +1,6 @@
 from os.path import dirname, realpath, join
 
-from setuptools import setup
+from setuptools import setup, find_packages
 
 here = dirname(realpath(__file__))
 
@@ -17,7 +17,7 @@ setup(
     author='Avatao.com Innovative Learning Kft.',
     author_email='support@avatao.com',
     license='custom',
-    packages=['tfw'],
+    packages=find_packages(),
     package_dir={'tfw': 'tfw'},
     install_requires=requirements,
     extras_require={

supervisor/components/terminal_fallback.conf

@@ -0,0 +1,4 @@
+[program:terminal_fallback]
+command=ncat -klp %(ENV_TFW_TERMINAL_FALLBACK_PORT)s
+autostart=true
+autorestart=true

supervisor/components/tfw_init.conf

@@ -4,3 +4,4 @@ directory=/tmp
 command=bash tfw_init.sh
 autorestart=false
 startsecs=0
+priority=1

supervisor/components/tfw_server.conf

@@ -2,3 +2,4 @@
 user=root
 directory=%(ENV_TFW_SERVER_DIR)s
 command=python3 -u tfw_server.py
+priority=2

tfw/components/frontend/frontend_config_handler/frontend_config_handler.py

@@ -12,7 +12,7 @@ class FrontendConfigHandler:
     def handle_event(self, _, connector):
         # pylint: disable=no-self-use
         for message in self._config_messages:
-            connector.send_message(message)
+            connector.send_message(message, scope=Scope.WEBSOCKET)
         connector.send_message({'key': 'frontend.ready'}, scope=Scope.WEBSOCKET)
 
     @property

tfw/components/frontend/message_queue_handler/message_queue_handler.py

@@ -1,43 +1,91 @@
-from time import sleep
+import logging
-from queue import Queue
+from time import sleep, time
+from queue import Queue, Empty
 from threading import Thread
+from contextlib import suppress
+
+LOG = logging.getLogger(__name__)
 
 
 class MessageQueueHandler:
     keys = ['message.queue']
     type_id = 'ControlEventHandler'
+    avg_word_len = 5
+    drain_poll_freq = 0.2
 
     def __init__(self, wpm):
         self.connector = None
         self.wpm = wpm
         self._queue = Queue()
+        self._drain_queue = Queue()
         self._thread = Thread(target=self._dispatch_messages)
 
+        self._commands = {
+            'message.queue': self.handle_queue,
+            'message.queue.drain': self.handle_drain
+        }
+
     def _dispatch_messages(self):
         for message in iter(self._queue.get, None):
-            wpm = message['wpm'] if 'wpm' in message else self.wpm
+            is_queue_empty = self._queue.empty()
-            cps = 5 * wpm / 60
+            message['typing'] = not is_queue_empty
-            sleep(len(message['message']) / cps)
+            self.connector.send_message(message)
+            self._sleep(self._get_sleep_time(message, is_queue_empty))
+
+    def _get_sleep_time(self, message, is_queue_empty):
+        words_per_min = message['wpm'] if 'wpm' in message else self.wpm
+        chars_per_min = self.avg_word_len * words_per_min / 60
+        return 0 if is_queue_empty else len(message['message']) / chars_per_min
+
+    def _sleep(self, seconds):
+        poll_freq = self.drain_poll_freq
+        if seconds < poll_freq:
+            poll_freq = seconds
+
+        sleep_until = time() + seconds
+        while time() < sleep_until:
+            sleep(poll_freq)
+            with suppress(Empty):
+                self._drain_queue.get(block=False)
+                self._drain()
+                break
+
+    def _drain(self):
+        with suppress(Empty):
+            while True:
+                message = self._queue.get(block=False)
+                if message is None:
+                    break
+                message['typing'] = False
                 self.connector.send_message(message)
 
     def handle_event(self, message, _):
+        try:
+            self._commands[message['key']](message)
+        except KeyError:
+            LOG.error('IGNORING MESSAGE: Invalid message received: %s', message)
+
+    def handle_queue(self, message):
         for unpacked in self._generate_messages_from_queue(message):
             self._queue.put(unpacked)
 
     @staticmethod
     def _generate_messages_from_queue(queue_message):
-        last = queue_message['messages'][-1]
         for message in queue_message['messages']:
             yield {
                 'key': 'message.send',
-                'typing': message is not last,
                 **message
             }
 
+    def handle_drain(self, _):
+        self._drain_queue.put(True)
+
     def start(self):
         self._thread.start()
 
     def cleanup(self):
+        # clearing the queue forces the loop in
+        # _dispatch_messages to block on _queue.get
         self._queue.queue.clear()
         self._queue.put(None)
         self._thread.join()

tfw/components/frontend/message_queue_handler/test_message_queue.py

@@ -1,8 +1,8 @@
 # pylint: disable=redefined-outer-name
-from math import inf
+from time import time
-from time import sleep
 from os import urandom
 from random import randint
+from queue import Queue, Empty
 
 import pytest
 
@@ -12,56 +12,141 @@ from .message_queue_handler import MessageQueueHandler
 class MockConnector:
     def __init__(self):
         self.callback = None
-        self.messages = []
+        self.messages = Queue()
+        self.send_times = Queue()
 
     def raise_event(self, message):
         self.callback(message, self)
-        sleep(0.01)
 
     def send_message(self, message):
-        self.messages.append(message)
+        self.messages.put(message)
+        self.send_times.put(time())
 
 
 @pytest.fixture
 def handler():
-    connector = MockConnector()
+    class NoSleepMessageQueueHandler(MessageQueueHandler):
-    handler = MessageQueueHandler(inf)
+        sleep_start_times = Queue()
-    handler.connector = connector
+        sleep_seconds = Queue()
-    connector.callback = handler.handle_event
+        sleep_end_times = Queue()
+
+        def _sleep(self, seconds):
+            self.sleep_start_times.put(time())
+            self.sleep_seconds.put(seconds)
+            super()._sleep(seconds)
+            self.sleep_end_times.put(time())
+
+    handler = NoSleepMessageQueueHandler(100000)
+    handler.connector = MockConnector()
+    handler.connector.callback = handler.handle_event
+
     handler.start()
     yield handler
     handler.cleanup()
 
 
-@pytest.fixture
+def get_message_queue(*, size=None):
-def queue():
+    size = randint(5, 10) if not size else size
-    yield {
+    return {
         'key': 'message.queue',
         'messages': [
-            {'originator': urandom(4).hex(), 'message': urandom(16).hex()}
+            {'originator': urandom(4).hex(), 'message': urandom(randint(10, 20)).hex()}
-            for _ in range(randint(5, 10))
+            for _ in range(size)
         ]
     }
 
 
-def test_message_order(handler, queue):
+def test_order(handler):
+    queue = get_message_queue()
     handler.connector.raise_event(queue)
-    old_list = queue['messages']
+    expected_messages = queue['messages']
-    new_list = handler.connector.messages
+
-    length = len(old_list)
+    actual_messages = []
-    assert len(new_list) == length
+    for _ in expected_messages:
-    for i in range(length):
+        actual_messages.append(handler.connector.messages.get())
-        unpacked = new_list[i]
+
-        assert unpacked['key'] == 'message.send'
+    assert len(actual_messages) == len(expected_messages)
-        assert unpacked['originator'] == old_list[i]['originator']
+    for i in range(len(expected_messages)):  # pylint: disable=consider-using-enumerate
-        assert unpacked['typing'] == (i < length-1)
+        message = actual_messages[i]
+        assert message['key'] == 'message.send'
+        assert message['originator'] == expected_messages[i]['originator']
+        assert message['typing'] == (i < len(expected_messages)-1)
 
 
-def test_wpm(handler, queue):
+def test_timing(handler):
-    handler.wpm = 10000
+    q1 = get_message_queue(size=2)
-    handler.connector.raise_event(queue)
+    q2 = get_message_queue(size=2)
-    assert not handler.connector.messages
+    handler.connector.raise_event(q1)
-    handler.wpm = 100000000
+    handler.connector.raise_event(q2)
-    handler.connector.raise_event(queue)
+
-    sleep(0.25)
+    messages = []
-    assert len(handler.connector.messages) == 2*len(queue['messages'])
+    send_times = []
+    sleep_start_times = []
+    sleep_seconds = []
+    sleep_end_times = []
+    for _ in range(len(q1['messages']) + len(q2['messages'])):
+        messages.append(handler.connector.messages.get())
+        send_times.append(handler.connector.send_times.get())
+        sleep_start_times.append(handler.sleep_start_times.get())
+        sleep_seconds.append(handler.sleep_seconds.get())
+        sleep_end_times.append(handler.sleep_end_times.get())
+
+    # no sleep before first message
+    assert sleep_start_times[0] > send_times[0]
+    assert messages[0]['typing']
+
+    # at least 'seconds' sleep before sending next messages
+    assert (send_times[0] + sleep_seconds[0]) < sleep_end_times[0]
+    assert (send_times[0] + sleep_seconds[0]) < send_times[1]
+    assert messages[1]['typing']
+    assert (send_times[1] + sleep_seconds[1]) < sleep_end_times[1]
+    assert (send_times[1] + sleep_seconds[1]) < send_times[2]
+    assert messages[2]['typing']
+
+    # at least 'seconds' sleep after last message
+    assert (send_times[2] + sleep_seconds[2]) < sleep_end_times[3]
+    assert not messages[3]['typing']
+
+
+def test_drain(handler):
+    q1 = get_message_queue()
+    q2 = get_message_queue()
+    messages_count = len(q1['messages']) + len(q2['messages'])
+    handler.wpm = 0.01  # very slow, everything will just block
+    handler.connector.raise_event(q1)
+    handler.connector.raise_event(q2)
+
+    handler.connector.messages.get(timeout=0.1)  # no sleep before first message
+    with pytest.raises(Empty):
+        handler.connector.messages.get(timeout=0.1)
+
+    handler.connector.raise_event({'key': 'message.queue.drain'})
+
+    for _ in range(messages_count - 1):
+        handler.connector.messages.get()
+
+
+def test_queue_works_after_drain(handler):
+    handler.connector.raise_event({'key': 'message.queue.drain'})
+
+    q = get_message_queue()
+    expected_msg_count = len(q['messages'])
+    handler.connector.raise_event(q)
+
+    messages = []
+    for _ in range(expected_msg_count):
+        messages.append(handler.connector.messages.get())
+
+    assert len(messages) == expected_msg_count
+    assert handler._thread.is_alive()  # pylint: disable=protected-access
+
+
+def test_queue_can_be_stopped(handler):
+    handler.wpm = 0.01  # very slow, everything will just block
+
+    q = get_message_queue()
+    handler.connector.raise_event(q)
+    handler.cleanup()
+    handler.connector.raise_event({'key': 'message.queue.drain'})
+
+    assert not handler._thread.is_alive()  # pylint: disable=protected-access

tfw/components/frontend/message_sender.py

@@ -1,17 +1,22 @@
+DEFAULT_ORIGINATOR = 'avataobot'
+
+
 class MessageSender:
     def __init__(self, uplink):
         self.uplink = uplink
 
-    def send(self, message, originator=None):
+    def send(self, message, originator=DEFAULT_ORIGINATOR, buttons=None):
         message = {
             'key': 'message.send',
             'message': message
         }
         if originator:
             message['originator'] = originator
+        if buttons:
+            message['buttons'] = buttons
         self.uplink.send_message(message)
 
-    def queue_messages(self, messages, originator=None):
+    def queue_messages(self, messages, originator=DEFAULT_ORIGINATOR, buttons=None):
         message_queue = {
             'key': 'message.queue',
             'messages': []
@@ -21,10 +26,6 @@ class MessageSender:
             if originator:
                 next_message['originator'] = originator
             message_queue['messages'].append(next_message)
+        if buttons and message_queue['messages']:
+            message_queue['messages'][-1]['buttons'] = buttons
         self.uplink.send_message(message_queue)
-
-    def set_originator(self, originator):
-        self.uplink.send_message({
-            'key': 'message.config',
-            'originator': originator
-        })

tfw/components/ide/ide_handler.py

@@ -79,6 +79,7 @@ class IdeHandler:
     def read(self, message):
         if 'patterns' in message:
             self.filemanager.patterns = message['patterns']
+            self.monitor.paths = self.filemanager.parents
         try:
             message['filename'] = self.filemanager.find_file(
                 message.get('filename') or self.initial_file

tfw/components/process_management/process_log_handler.py

@@ -16,10 +16,13 @@ class ProcessLogHandler:
         self._initial_log_tail = log_tail
 
         self.command_handlers = {
-            'process.log.set': self.handle_set
+            'process.log.set': self.handle_set,
+            'process.log.start': self.handle_start,
+            'process.log.stop': self.handle_stop
         }
 
     def start(self):
+        if not self._monitor:
             self._monitor = LogInotifyObserver(
                 connector=self.connector,
                 process_name=self.process_name,
@@ -40,5 +43,16 @@ class ProcessLogHandler:
         if data.get('tail'):
             self._monitor.log_tail = data['tail']
 
-    def cleanup(self):
+    def handle_start(self, _):
+        self.start()
+
+    def handle_stop(self, _):
+        self._stop_monitor()
+
+    def _stop_monitor(self):
+        if self._monitor:
             self._monitor.stop()
+            self._monitor = None
+
+    def cleanup(self):
+        self._stop_monitor()

tfw/components/terminal/terminado_mini_server.py

@@ -17,7 +17,7 @@ class TerminadoMiniServer:
             url,
             TerminadoMiniServer.ResetterTermSocket,
             {'term_manager': self._term_manager}
-        )])
+        )], websocket_ping_interval=30)
 
     @property
     def term_manager(self):
@@ -33,10 +33,6 @@ class TerminadoMiniServer:
         def check_origin(self, origin):
             return True
 
-        def on_close(self):
-            self.term_manager.terminal = None
-            self.term_manager.get_terminal()
-
     def listen(self):
         self.application.listen(self.port)
 

tfw/components/terminal/terminal_handler.py

@@ -42,6 +42,9 @@ class TerminalHandler:
             LOG.error('IGNORING MESSAGE: Invalid message received: %s', message)
 
     def handle_write(self, message):
+        concat = message.get('concat', False)
+        if not concat:
+            self.terminado_server.pty.write('\x15')
         self.terminado_server.pty.write(message['command'])
 
     def cleanup(self):

tfw/fsm/fsm_base.py

@@ -1,6 +1,7 @@
 import logging
 from collections import defaultdict
 from datetime import datetime
+from contextlib import suppress
 
 from transitions import Machine, MachineError
 
@@ -62,14 +63,12 @@ class FSMBase(Machine, CallbackMixin):
         )
 
         if all(predicate_results):
-            try:
+            with suppress(AttributeError, MachineError):
                 from_state = self.state
-                self.trigger(trigger)
+                if self.trigger(trigger):
                     self.update_event_log(from_state, trigger)
                     return True
-            except (AttributeError, MachineError):
             LOG.debug('FSM failed to execute nonexistent trigger: "%s"', trigger)
-        return False
 
     def update_event_log(self, from_state, trigger):
         self.event_log.append({

tfw/internals/networking/serialization.py

@@ -42,7 +42,8 @@ def deserialize_tfw_msg(*args):
     """
     Return message from TFW multipart data
     """
-    return _deserialize_all(*args)[1]
+    envelope = _deserialize_all(*args)
+    return _repair_if_needed(envelope)
 
 
 def _serialize_all(*args):
@@ -84,6 +85,18 @@ def _deserialize_single(data):
         return _decode_if_needed(data)
 
 
+def _repair_if_needed(envelope):
+    """
+    Quick fix for broken messages received from separate processes.
+    """
+    if len(envelope) == 2:
+        return envelope[1]
+    for part in envelope:
+        if isinstance(part, dict):
+            return part
+    return {}
+
+
 def _encode_if_needed(value):
     """
     Return input as bytes

tfw/internals/networking/test_networking.py

@@ -39,9 +39,12 @@ def zmq_connector(_listener_and_connector):
 
 
 def run_ioloop_once():
-    # hack: we have to wait for the messages to get through
+    # Hack: we have to wait for the messages to get through
     # the network stack of the OS while the IOLoop is waiting
-    # for them via select/epoll/kqueue
+    # for them via select/epoll/kqueue.
+    # This is an inherent race condition, but solving this
+    # problem properly would make the test code difficult
+    # to understand, so we use this half measure.
     IOLoop.current().call_later(0.1, IOLoop.current().stop)
     IOLoop.current().start()
 
@@ -67,7 +70,7 @@ def test_messages():
 def wait_until_subscriber_connects(listener, connector):
     # Warning: you are better off without comprehending how this works
     # Reference: ZMQ PUB-SUB slow joiner problem
-
+    connector.subscribe('-', '_')
     # Wait until something can go through the connection
     dummy = {'key': '-'}
     while True:
@@ -82,6 +85,7 @@ def wait_until_subscriber_connects(listener, connector):
         with suppress(IOError):
             if connector.recv_message(block=False) == sentinel:
                 break
+    connector.unsubscribe('-', '_')
 
 
 def test_server_downlink(zmq_listener, zmq_connector, test_messages):
@@ -158,8 +162,10 @@ def test_connector_preserves_intent(zmq_listener, zmq_connector):
 
 
 def test_server_uplink(zmq_listener, zmq_connector, test_messages):
-    messages = []
     zmq_connector.subscribe('')
+    wait_until_subscriber_connects(zmq_listener, zmq_connector)
+
+    messages = []
     zmq_connector.register_callback(messages.append)
 
     for message in test_messages:
@@ -175,8 +181,10 @@ def test_connector_downlink_subscribe(zmq_listener, zmq_connector):
     key2_messages = [{'key': '2', 'data': i} for i in range(randint(128, 256))]
     all_messages = key1_messages + key2_messages
 
-    messages = []
     zmq_connector.subscribe('1')
+    wait_until_subscriber_connects(zmq_listener, zmq_connector)
+
+    messages = []
     zmq_connector.register_callback(messages.append)
 
     for message in all_messages:
@@ -197,6 +205,7 @@ def test_listener_sync_recv(zmq_listener, zmq_connector, test_messages):
 def test_connector_sync_recv(zmq_listener, zmq_connector, test_messages):
     zmq_connector.subscribe('')
     wait_until_subscriber_connects(zmq_listener, zmq_connector)
+
     for message in test_messages:
         zmq_listener.send_message(message)
         assert zmq_connector.recv_message() == message

tfw/logging.py

@@ -2,6 +2,7 @@
 from datetime import datetime
 from typing import TextIO, Union
 from dataclasses import dataclass
+from collections.abc import Mapping
 from traceback import format_exception
 from logging import DEBUG, getLogger, Handler, Formatter, Filter
 
@@ -81,7 +82,7 @@ class LogFormatter(Formatter):
     def format(self, record):
         time = datetime.utcfromtimestamp(record.created).strftime('%H:%M:%S')
         if record.args:
-            tuple_args = (record.args,) if isinstance(record.args, dict) else record.args
+            tuple_args = (record.args,) if isinstance(record.args, Mapping) else record.args
             clean_args = tuple((self.trim(arg) for arg in tuple_args))
             message = record.msg % clean_args
         else:

tfw/main/tfw_server.py

@@ -28,7 +28,7 @@ class TFWServer:
             r'/ws', ZMQWebSocketRouter, {
                 'listener': self._listener,
             }
-        )])
+        )], websocket_ping_interval=30)
 
     def listen(self):
         self.application.listen(TFWENV.WEB_PORT)