Update frontend

Dockerfile

@@ -1,9 +1,10 @@
-FROM avatao/frontend-tutorial-framework:chausie-20190930 as frontend
+FROM avatao/frontend-tutorial-framework:chausie-20191021 as frontend
 FROM avatao/debian:buster
 
 RUN apt-get update                           &&\
     apt-get install -y --no-install-recommends \
         supervisor                             \
+        ncat                                   \
         libzmq5                                \
         nginx                                  \
         jq                                     \
@@ -19,12 +20,13 @@ RUN curl -Ls https://github.com/krallin/tini/releases/download/v0.18.0/tini-amd6
     sha256sum --check --status                                                                       &&\
     chmod 755 /bin/init
 
-ENV TFW_PUBLIC_PORT=8888          \
-    TFW_WEB_PORT=4242             \
-    TFW_LOGIN_APP_PORT=6666       \
-    TFW_TERMINADO_PORT=7878       \
-    TFW_SUPERVISOR_HTTP_PORT=9001 \
-    TFW_PUB_PORT=7654             \
+ENV TFW_PUBLIC_PORT=8888            \
+    TFW_WEB_PORT=4242               \
+    TFW_LOGIN_APP_PORT=6666         \
+    TFW_TERMINAL_PORT=7878          \
+    TFW_TERMINAL_FALLBACK_PORT=7879 \
+    TFW_SUPERVISOR_HTTP_PORT=9001   \
+    TFW_PUB_PORT=7654               \
     TFW_PULL_PORT=8765
 
 EXPOSE ${TFW_PUBLIC_PORT}

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;

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

tfw/components/frontend/message_queue_handler/message_queue_handler.py

@@ -1,43 +1,90 @@
-from time import sleep
-from queue import Queue
+import logging
+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
-            cps = 5 * wpm / 60
-            sleep(len(message['message']) / cps)
+            message['typing'] = not self._queue.empty()
             self.connector.send_message(message)
+            self._sleep(self._get_sleep_time(message))
+
+    def _get_sleep_time(self, message):
+        words_per_min = message['wpm'] if 'wpm' in message else self.wpm
+        chars_per_min = self.avg_word_len * words_per_min / 60
+        return 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 sleep
+from time import time
 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()
-    handler = MessageQueueHandler(inf)
-    handler.connector = connector
-    connector.callback = handler.handle_event
+    class NoSleepMessageQueueHandler(MessageQueueHandler):
+        sleep_start_times = Queue()
+        sleep_seconds = Queue()
+        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 queue():
-    yield {
+def get_message_queue(*, size=None):
+    size = randint(5, 10) if not size else size
+    return {
         'key': 'message.queue',
         'messages': [
-            {'originator': urandom(4).hex(), 'message': urandom(16).hex()}
-            for _ in range(randint(5, 10))
+            {'originator': urandom(4).hex(), 'message': urandom(randint(10, 20)).hex()}
+            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']
-    new_list = handler.connector.messages
-    length = len(old_list)
-    assert len(new_list) == length
-    for i in range(length):
-        unpacked = new_list[i]
-        assert unpacked['key'] == 'message.send'
-        assert unpacked['originator'] == old_list[i]['originator']
-        assert unpacked['typing'] == (i < length-1)
+    expected_messages = queue['messages']
+
+    actual_messages = []
+    for _ in expected_messages:
+        actual_messages.append(handler.connector.messages.get())
+
+    assert len(actual_messages) == len(expected_messages)
+    for i in range(len(expected_messages)):  # pylint: disable=consider-using-enumerate
+        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):
-    handler.wpm = 10000
-    handler.connector.raise_event(queue)
-    assert not handler.connector.messages
-    handler.wpm = 100000000
-    handler.connector.raise_event(queue)
-    sleep(0.25)
-    assert len(handler.connector.messages) == 2*len(queue['messages'])
+def test_timing(handler):
+    q1 = get_message_queue(size=2)
+    q2 = get_message_queue(size=2)
+    handler.connector.raise_event(q1)
+    handler.connector.raise_event(q2)
+
+    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/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