Control socket: overhaul finished for now

This unbreaks the commands broken by the last commit. Everything is
still better.

anonstream/__init__.py

@@ -51,6 +51,9 @@ def create_app(config_file):
     # Background tasks' asyncio.sleep tasks, cancelled on shutdown
     app.background_sleep = set()
 
+    # Queues for event socket clients
+    app.event_queues = set()
+
     @app.after_serving
     async def shutdown():
         # Force all background tasks to finish
@@ -60,10 +63,23 @@ def create_app(config_file):
     @app.before_serving
     async def startup():
         # Start control server
-        from anonstream.control.server import start_control_server_at
-        async def start_control_server():
-            return await start_control_server_at(app.config['CONTROL_ADDRESS'])
-        app.add_background_task(start_control_server)
+        if app.config['SOCKET_CONTROL_ENABLED']:
+            from anonstream.control.server import start_control_server_at
+            async def start_control_server():
+                return await start_control_server_at(
+                    app.config['SOCKET_CONTROL_ADDRESS']
+                )
+            app.add_background_task(start_control_server)
+
+        # Start event server
+        if app.config['SOCKET_EVENT_ENABLED']:
+            from anonstream.events import start_event_server_at
+            async def start_event_server():
+                return await start_event_server_at(
+                        app.config['SOCKET_EVENT_ADDRESS']
+                )
+            app.add_background_task(start_event_server)
+
 
         # Create routes and background tasks
         import anonstream.routes

anonstream/chat.py

@@ -7,6 +7,7 @@ from datetime import datetime
 from quart import current_app, escape
 
 from anonstream.broadcast import broadcast, broadcast_users_update
+from anonstream.events import notify_event_sockets
 from anonstream.helpers.chat import generate_nonce_hash, get_scrollback
 from anonstream.utils.chat import get_message_for_websocket, get_approx_linespan
 
@@ -102,6 +103,12 @@ def add_chat_message(user, nonce, comment, ignore_empty=False):
     while len(MESSAGES_BY_ID) > CONFIG['MAX_CHAT_MESSAGES']:
         MESSAGES_BY_ID.pop(last=False)
 
+    # Notify event sockets that a chat message was added
+    notify_event_sockets({
+        'type': 'message',
+        'event': message,
+    })
+
     # Broadcast a users update to all websockets,
     # in case this message is from a new user
     broadcast_users_update()

anonstream/config.py

@@ -13,7 +13,6 @@ def update_flask_from_toml(toml_config, flask_config):
     flask_config.update({
         'SECRET_KEY_STRING': toml_config['secret_key'],
         'SECRET_KEY': toml_config['secret_key'].encode(),
-        'CONTROL_ADDRESS': toml_config['control']['address'],
         'AUTH_USERNAME': toml_config['auth']['username'],
         'AUTH_PWHASH': auth_pwhash,
         'AUTH_TOKEN': generate_token(),
@@ -25,6 +24,7 @@ def update_flask_from_toml(toml_config, flask_config):
 
 def toml_to_flask_sections(config):
     TOML_TO_FLASK_SECTIONS = (
+        toml_to_flask_section_socket,
         toml_to_flask_section_segments,
         toml_to_flask_section_title,
         toml_to_flask_section_names,
@@ -38,6 +38,15 @@ def toml_to_flask_sections(config):
     for toml_to_flask_section in TOML_TO_FLASK_SECTIONS:
         yield toml_to_flask_section(config)
 
+def toml_to_flask_section_socket(config):
+    cfg = config['socket']
+    return {
+        'SOCKET_CONTROL_ENABLED': cfg['control']['enabled'],
+        'SOCKET_CONTROL_ADDRESS': cfg['control']['address'],
+        'SOCKET_EVENT_ENABLED': cfg['event']['enabled'],
+        'SOCKET_EVENT_ADDRESS': cfg['event']['address'],
+    }
+
 def toml_to_flask_section_segments(config):
     cfg = config['segments']
     return {

anonstream/control/commands/__init__.py

@@ -1,44 +0,0 @@
-from anonstream.control.commands.help import *
-from anonstream.control.commands.exit import *
-from anonstream.control.commands.title import *
-from anonstream.control.commands.chat import *
-from anonstream.control.commands.user import *
-
-METHOD_HELP = 'help'
-METHOD_EXIT = 'exit'
-METHOD_TITLE = 'title'
-METHOD_CHAT = 'chat'
-METHOD_USER = 'user'
-
-METHOD_COMMAND_FUNCTIONS = {
-    METHOD_HELP: {
-        None: command_help,
-        'help': command_help_help,
-    },
-    METHOD_EXIT: {
-        None: command_exit,
-        'help': command_exit_help,
-    },
-    METHOD_TITLE: {
-        None: command_title_show,
-        'help': command_title_help,
-        'show': command_title_show,
-        'set': command_title_set,
-    },
-    METHOD_CHAT: {
-        None: command_chat_help,
-        'help': command_chat_help,
-        'delete': command_chat_delete,
-    },
-    METHOD_USER: {
-        None: command_user_show,
-        'help': command_user_help,
-        'show': command_user_show,
-        'attr': command_user_attr,
-        'get': command_user_get,
-        'set': command_user_set,
-        'eyes': command_user_eyes,
-    },
-}
-
-

anonstream/control/commands/chat.py

@@ -1,34 +0,0 @@
-from anonstream.control.exceptions import BadArgument, Incomplete, Garbage
-from anonstream.chat import delete_chat_messages
-
-async def command_chat_help(args):
-    match args:
-        case []:
-            normal_options = ['help']
-            response = (
-                'Usage: chat {show [MESSAGES] | delete SEQS}\n'
-                'Commands:\n'
-                #' chat show [MESSAGES]......show chat messages\n'
-                ' chat delete SEQS..........delete chat messages\n'
-                'Definitions:\n'
-                #' MESSAGES..................undefined\n'
-                ' SEQS......................=SEQ [SEQ...]\n'
-                ' SEQ.......................a chat message\'s seq, base-10 integer\n'
-            )
-        case [*garbage]:
-            raise Garbage(garbage)
-    return normal_options, response
-
-async def command_chat_delete(args):
-    match args:
-        case []:
-            raise Incomplete
-        case _:
-            try:
-                seqs = list(map(int, args))
-            except ValueError as e:
-                raise BadArgument('SEQ must be a base-10 integer') from e
-            delete_chat_messages(seqs)
-            normal_options = ['delete', *map(str, seqs)]
-            response = ''
-    return normal_options, response

anonstream/control/commands/exit.py

@@ -1,20 +0,0 @@
-from anonstream.control.exceptions import Exit, Garbage
-
-async def command_exit(args):
-    match args:
-        case []:
-            raise Exit
-        case [*garbage]:
-            raise Garbage(garbage)
-
-async def command_exit_help(args):
-    match args:
-        case []:
-            normal_options = ['help']
-            response = (
-                'Usage: exit\n'
-                'close the connection\n'
-            )
-        case [*garbage]:
-            raise Garbage(garbage)
-    return normal_options, response

anonstream/control/commands/help.py

@@ -1,38 +0,0 @@
-from anonstream.control.exceptions import Garbage
-
-async def command_help(args):
-    match args:
-        case []:
-            normal_options = []
-            response = (
-                'Usage: METHOD [COMMAND | help]\n'
-                'Examples:\n'
-                ' help...........................show this help message\n'
-                ' exit...........................close the control connection\n'
-                ' title [show]...................show the stream title\n'
-                ' title set TITLE................set the stream title\n'
-                ' user [show]....................show a list of users\n'
-                ' user attr USER.................set an attribute of a user\n'
-                ' user get USER ATTR.............set an attribute of a user\n'
-                ' user set USER ATTR VALUE.......set an attribute of a user\n'
-                #' user kick USERS [FAREWELL].....kick users\n'
-                ' user eyes USER [show]..........show a list of active video responses\n'
-                ' user eyes USER delete EYES_ID..end a video response\n'
-                #' chat show MESSAGES.............show a list of messages\n'
-                ' chat delete SEQS...............delete a set of messages\n'
-            )
-        case [*garbage]:
-            raise Garbage(garbage)
-    return normal_options, response
-
-async def command_help_help(args):
-    match args:
-        case []:
-            normal_options = ['help']
-            response = (
-                'Usage: help\n'
-                'show usage syntax and examples\n'
-            )
-        case [*garbage]:
-            raise Garbage(garbage)
-    return normal_options, response

anonstream/control/commands/title.py

@@ -1,53 +0,0 @@
-import json
-
-from anonstream.control.exceptions import BadArgument, Incomplete, Garbage, Failed
-from anonstream.control.utils import json_dumps_contiguous
-from anonstream.stream import get_stream_title, set_stream_title
-
-async def command_title_help(args):
-    match args:
-        case []:
-            normal_options = ['help']
-            response = (
-                'Usage: title [show | set TITLE]\n'
-                'Commands:\n'
-                ' title [show].......show the stream title\n'
-                ' title set TITLE....set the stream title to TITLE\n'
-                'Definitions:\n'
-                ' TITLE..............a json-encoded string, whitespace must be \\uXXXX-escaped\n'
-            )
-        case [*garbage]:
-            raise Garbage(garbage)
-    return normal_options, response
-
-async def command_title_show(args):
-    match args:
-        case []:
-            normal_options = ['show']
-            response = json.dumps(await get_stream_title()) + '\n'
-        case [*garbage]:
-            raise Garbage(garbage)
-    return normal_options, response
-
-async def command_title_set(args):
-    match args:
-        case [title_json]:
-            try:
-                title = json.loads(title_json)
-            except json.JSONDecodeError as e:
-                raise BadArgument('could not decode json')
-            else:
-                if not isinstance(title, str):
-                    raise BadArgument('could not decode json as string')
-                else:
-                    try:
-                        await set_stream_title(title)
-                    except OSError as e:
-                        raise Failed(str(e)) from e
-                    normal_options = ['set', json_dumps_contiguous(title)]
-                    response = ''
-        case []:
-            raise Incomplete
-        case [_, *garbage]:
-            raise Garbage(garbage)
-    return normal_options, response

anonstream/control/commands/user.py

@@ -1,153 +0,0 @@
-import json
-
-from quart import current_app
-
-from anonstream.control.exceptions import BadArgument, Incomplete, Garbage, Failed
-from anonstream.control.utils import json_dumps_contiguous
-from anonstream.utils.user import USER_WEBSOCKET_ATTRS
-
-USERS_BY_TOKEN = current_app.users_by_token
-USERS = current_app.users
-USERS_UPDATE_BUFFER = current_app.users_update_buffer
-
-async def command_user_help(args):
-    match args:
-        case []:
-            normal_options = ['help']
-            response = (
-                    'Usage: user [show | attr USER | get USER ATTR | set USER ATTR VALUE]\n'
-                    'Commands:\n'
-                    ' user [show].......................show all users\' tokens\n'
-                    ' user attr USER....................show names of a user\'s attributes\n'
-                    ' user get USER ATTR................show an attribute of a user\n'
-                    ' user set USER ATTR................set an attribute of a user\n'
-                    ' user eyes USER [show].............show a user\'s active video responses\n'
-                    ' user eyes USER delete EYES_ID.....end a video response to a user\n'
-                    'Definitions:\n'
-                    #' USER..............................={token TOKEN | hash HASH}\n'
-                    ' USER..............................=token TOKEN\n'
-                    ' TOKEN..............................a token\n'
-                    #' HASH..............................a token hash\n'
-                    ' ATTR...............................a user attribute, re:[a-z0-9_]+\n'
-                    ' EYES_ID............................a user\'s eyes_id, base 10 integer\n'
-            )
-        case [*garbage]:
-            raise Garbage(garbage)
-    return normal_options, response
-
-async def command_user_show(args):
-    match args:
-        case []:
-            normal_options = ['show']
-            response = json.dumps(tuple(USERS_BY_TOKEN)) + '\n'
-        case [*garbage]:
-            raise Garbage(garbage)
-    return normal_options, response
-
-async def command_user_attr(args):
-    match args:
-        case []:
-            raise Incomplete
-        case ['token', token_json]:
-            try:
-                token = json.loads(token_json)
-            except json.JSONDecodeError:
-                raise BadArgument('could not decode TOKEN as json')
-            try:
-                user = USERS_BY_TOKEN[token]
-            except KeyError:
-                raise Failed(f"no user exists with token {token!r}, try 'user show'")
-            normal_options = ['attr', 'token', json_dumps_contiguous(token)]
-            response = json.dumps(tuple(user.keys())) + '\n'
-        case [*garbage]:
-            raise Garbage(garbage)
-    return normal_options, response
-
-async def command_user_get(args):
-    match args:
-        case ['token', token_json, attr]:
-            try:
-                token = json.loads(token_json)
-            except json.JSONDecodeError:
-                raise BadArgument('could not decode TOKEN as json')
-            try:
-                user = USERS_BY_TOKEN[token]
-            except KeyError:
-                raise Failed(f"no user exists with token {token!r}, try 'user show'")
-            try:
-                value = user[attr]
-            except KeyError:
-                raise Failed(f"user has no attribute {attr!r}, try 'user attr token {json_dumps_contiguous(token)}'")
-            try:
-                value_json = json.dumps(value)
-            except TypeError:
-                raise Failed(f'attribute {attr!r} is not JSON serializable')
-            normal_options = ['get', 'token', json_dumps_contiguous(token), attr]
-            response = value_json + '\n'
-        case []:
-            raise Incomplete
-        case [*garbage]:
-            raise Garbage(garbage)
-    return normal_options, response
-
-async def command_user_set(args):
-    match args:
-        case ['token', token_json, attr, value_json]:
-            try:
-                token = json.loads(token_json)
-            except json.JSONDecodeError:
-                raise BadArgument('could not decode TOKEN as json')
-            try:
-                user = USERS_BY_TOKEN[token]
-            except KeyError:
-                raise Failed(f"no user exists with token {token!r}, try 'user show'")
-            try:
-                value = user[attr]
-            except KeyError:
-                raise Failed(f"user has no attribute {attr!r}, try 'user attr token {json_dumps_contiguous(token)}")
-            try:
-                value = json.loads(value_json)
-            except json.JSONDecodeError:
-                raise Failed('could not decode json')
-            user[attr] = value
-            if attr in USER_WEBSOCKET_ATTRS:
-                USERS_UPDATE_BUFFER.add(token)
-            normal_options = ['set', 'token', json_dumps_contiguous(token), attr, json_dumps_contiguous(value)]
-            response = ''
-        case []:
-            raise Incomplete
-        case [*garbage]:
-            raise Garbage(garbage)
-    return normal_options, response
-
-async def command_user_eyes(args):
-    match args:
-        case ['token', token_json, *subargs]:
-            try:
-                token = json.loads(token_json)
-            except json.JSONDecodeError:
-                raise BadArgument('could not decode TOKEN as json')
-            try:
-                user = USERS_BY_TOKEN[token]
-            except KeyError:
-                raise Failed(f"no user exists with token {token!r}, try 'user show'")
-            match subargs:
-                case [] | ['show']:
-                    normal_options = ['eyes', 'token', json_dumps_contiguous(token), 'show']
-                    response = json.dumps(user['eyes']['current']) + '\n'
-                case ['delete', eyes_id_json]:
-                    try:
-                        eyes_id = json.loads(eyes_id_json)
-                    except json.JSONDecodeError:
-                        raise BadArgument('could not decode EYES_ID as json')
-                    try:
-                        user['eyes']['current'].pop(eyes_id)
-                    except KeyError:
-                        pass
-                    normal_options = ['eyes', 'token', json_dumps_contiguous(token), 'delete', json_dumps_contiguous(eyes_id)]
-                    response = ''
-        case []:
-            raise Incomplete
-        case [*garbage]:
-            raise Garbage(garbage)
-    return normal_options, response

anonstream/control/exceptions.py

@@ -1,23 +1,5 @@
 class Exit(Exception):
     pass
 
-class UnknownMethod(Exception):
-    pass
-
-class UnknownCommand(Exception):
-    pass
-
-class UnknownArgument(Exception):
-    pass
-
-class BadArgument(Exception):
-    pass
-
-class Incomplete(Exception):
-    pass
-
-class Garbage(Exception):
-    pass
-
-class Failed(Exception):
+class Fail(Exception):
     pass

anonstream/control/parse.py

@@ -1,61 +1,41 @@
-import json
+from anonstream.control.spec import NoParse, Ambiguous, Parsed
+from anonstream.control.spec.common import Str
+from anonstream.control.spec.methods.chat import SPEC as SPEC_CHAT
+from anonstream.control.spec.methods.exit import SPEC as SPEC_EXIT
+from anonstream.control.spec.methods.help import SPEC as SPEC_HELP
+from anonstream.control.spec.methods.title import SPEC as SPEC_TITLE
+from anonstream.control.spec.methods.user import SPEC as SPEC_USER
 
-from anonstream.control.exceptions import UnknownMethod, UnknownCommand, BadArgument, Incomplete, Garbage, Failed
-from anonstream.control.commands import METHOD_COMMAND_FUNCTIONS
+SPEC = Str({
+    'help': SPEC_HELP,
+    'exit': SPEC_EXIT,
+    'title': SPEC_TITLE,
+    'chat': SPEC_CHAT,
+    'user': SPEC_USER,
+})
 
-async def parse_request(request):
-    try:
-        method, *options = request.split()
-    except ValueError:
-        normal, response = (None, []), ''
+async def parse(request):
+    words = request.split()
+    if not words:
+        normal, response = None, ''
     else:
-        try:
-            normal, response = await parse(method, options)
-        except UnknownMethod as e:
-            unknown_method, *_ = e.args
-            normal = None
-            response = f"method {unknown_method!r} is unknown, try 'help'\n"
-        except UnknownCommand as e:
-            method, unknown_command, *_ = e.args
-            normal = None
-            response = f"command {unknown_command!r} is unknown, try {f'{method} help'!r}\n"
-        except BadArgument as e:
-            reason, *_ = e.args
-            normal = None
-            response = f"{reason}, try {f'{method} help'!r}\n"
-        except Incomplete as e:
-            method, *_ = e.args
-            normal = None
-            response = f"command is incomplete, try {f'{method} help'!r}\n"
-        except Garbage as e:
-            garbage, *_ = e.args
-            normal = None
-            response = f"command has trailing garbage {garbage!r}, try {f'{method} help'!r}\n"
-        except Failed as e:
-            reason, *_ = e.args
-            normal = None
-            response = reason + '\n'
-    return normal, response
-
-async def parse(method, options):
-    try:
-        command, *args = options
-    except ValueError:
-        command, args = None, []
-    try:
-        functions = METHOD_COMMAND_FUNCTIONS[method]
-    except KeyError:
-        raise UnknownMethod(method)
-    else:
-        normal_method = method
-        try:
-            fn = functions[command]
-        except KeyError:
-            raise UnknownCommand(method, command)
-        else:
+        spec = SPEC
+        index = 0
+        args = []
+        while True:
             try:
-                normal_options, response = await fn(args)
-            except Incomplete as e:
-                raise Incomplete(method) from e
-    normal = (normal_method, normal_options)
+                spec, n_consumed, more_args = spec.consume(words, index)
+            except NoParse as e:
+                normal, response = None, e.args[0] + '\n'
+                break
+            except Ambiguous as e:
+                normal, response = None, e.args[0] + '\n'
+                break
+            except Parsed as e:
+                fn, *_ = e.args
+                normal, response = await fn(*args)
+                break
+            else:
+                index += n_consumed
+                args.extend(more_args)
     return normal, response

anonstream/control/server.py

@@ -1,7 +1,7 @@
 import asyncio
 
-from anonstream.control.exceptions import Exit
-from anonstream.control.parse import parse_request
+from anonstream.control.exceptions import Exit, Fail
+from anonstream.control.parse import parse
 
 def start_control_server_at(address):
     return asyncio.start_unix_server(serve_control_client, address)
@@ -14,19 +14,19 @@ async def serve_control_client(reader, writer):
             normal, response = None, str(e)
         else:
             try:
-                normal, response = await parse_request(request)
+                normal, response = await parse(request)
+            except Fail as e:
+                normal, response = None, e.args[0] + '\n'
             except Exit:
                 writer.close()
                 break
 
         if normal is not None:
-            normal_method, normal_options = normal
-            if normal_method is not None:
-                writer.write(normal_method.encode())
-                for arg in normal_options:
+            for index, word in enumerate(normal):
+                if index > 0:
                     writer.write(b' ')
-                    writer.write(arg.encode())
-                writer.write(b'\n')
+                writer.write(word.encode())
+            writer.write(b'\n')
         elif response:
             writer.write(b'error: ')
 

anonstream/control/spec/__init__.py

@@ -0,0 +1,12 @@
+class NoParse(Exception):
+    pass
+
+class Ambiguous(Exception):
+    pass
+
+class Parsed(Exception):
+    pass
+
+class Spec:
+    def consume(self, words, index):
+        raise NotImplemented

anonstream/control/spec/common.py

@@ -0,0 +1,103 @@
+import json
+
+from anonstream.control.spec import Spec, NoParse, Ambiguous, Parsed
+from anonstream.control.spec.utils import get_item, startswith
+
+class Str(Spec):
+    def __init__(self, directives):
+        self.directives = directives
+
+    def consume(self, words, index):
+        word = get_item(index, words)
+        candidates = tuple(filter(
+            lambda directive: startswith(directive, word),
+            self.directives,
+        ))
+        try:
+            directive = candidates[0]
+        except IndexError as e:
+            if word is None:
+                reason = f'incomplete: expected one of {set(self.directives)}'
+            else:
+                reason = (
+                    f'bad word at position {index} {word!r}: '
+                    f'expected one of {set(self.directives)}'
+                )
+            raise NoParse(reason) from e
+        else:
+            if len(candidates) > 1:
+                raise Ambiguous(
+                    f'bad word at position {index} {word!r}: ambiguous '
+                    f'abbreviation: {set(candidates)}'
+                )
+        return self.directives[directive], 1, []
+
+class End(Spec):
+    def __init__(self, fn):
+        self.fn = fn
+
+    def consume(self, words, index):
+        if len(words) <= index:
+            raise Parsed(self.fn)
+        raise NoParse(f'garbage at position {index} {words[index:]!r}')
+
+class Args(Spec):
+    def __init__(self, spec):
+        self.spec = spec
+
+class ArgsInt(Args):
+    def consume(self, words, index):
+        try:
+            n_string = words[index]
+        except IndexError:
+            raise NoParse(f'incomplete: expected integer')
+        else:
+            try:
+                n = int(n_string)
+            except ValueError:
+                raise NoParse(
+                    f'bad argument at position {index} {n_string!r}: '
+                    f'could not decode base-10 integer'
+                )
+        return self.spec, 1, [n]
+
+class ArgsString(Args):
+    def consume(self, words, index):
+        try:
+            string = words[index]
+        except IndexError:
+            raise NoParse(f'incomplete: expected string')
+        return self.spec, 1, [string]
+
+class ArgsJson(Args):
+    def assert_obj(self, index, obj_json, obj):
+        pass
+
+    def transform_obj(self, obj):
+        return obj
+
+    def consume(self, words, index):
+        try:
+            obj_json = words[index]
+        except IndexError:
+            raise NoParse(f'incomplete: expected json')
+        else:
+            try:
+               obj = json.loads(obj_json)
+            except json.JSONDecodeError as e:
+                raise NoParse(
+                    f'bad argument at position {index} {obj_json!r}: '
+                    f'could not decode json'
+                )
+            else:
+                self.assert_obj(index, obj_json, obj)
+                obj = self.transform_obj(obj)
+        return self.spec, 1, [obj]
+
+class ArgsJsonString(ArgsJson):
+    def assert_obj(self, index, obj_json, obj):
+        if not isinstance(obj, str):
+            raise NoParse(
+                f'bad argument at position {index} {obj_json!r}: '
+                f'could not decode json string'
+            )

anonstream/control/spec/methods/chat.py

@@ -0,0 +1,52 @@
+import itertools
+
+from anonstream.chat import delete_chat_messages
+from anonstream.control.spec import NoParse
+from anonstream.control.spec.common import Str, End, Args
+from anonstream.control.spec.utils import get_item, json_dumps_contiguous
+
+class ArgsSeqs(Args):
+    def consume(self, words, index):
+        seqs = []
+        for i in itertools.count():
+            seq_string = get_item(index + i, words)
+            try:
+                seq = int(seq_string)
+            except TypeError as e:
+                if not seqs:
+                    raise NoParse('incomplete: expected SEQ') from e
+                else:
+                    break
+            except ValueError as e:
+                raise NoParse(
+                        'could not decode {word!r} as base-10 integer'
+                ) from e
+            else:
+                seqs.append(seq)
+        return self.spec, i + 1, seqs
+
+async def cmd_chat_help():
+    normal = ['chat', 'help']
+    response = (
+        'Usage: chat {show [MESSAGES] | delete SEQS}\n'
+        'Commands:\n'
+        #' chat show [MESSAGES]......show chat messages\n'
+        ' chat delete SEQS..........delete chat messages\n'
+        'Definitions:\n'
+        #' MESSAGES..................undefined\n'
+        ' SEQS......................=SEQ [SEQ...]\n'
+        ' SEQ.......................a chat message\'s seq, base-10 integer\n'
+    )
+    return normal, response
+
+async def cmd_chat_delete(*seqs):
+    delete_chat_messages(seqs)
+    normal = ['chat', 'delete', *map(str, seqs)]
+    response = ''
+    return normal, response
+
+SPEC = Str({
+    None: End(cmd_chat_help),
+    'help': End(cmd_chat_help),
+    'delete': ArgsSeqs(End(cmd_chat_delete)),
+})

anonstream/control/spec/methods/exit.py

@@ -0,0 +1,18 @@
+from anonstream.control.spec.common import Str, End
+from anonstream.control.exceptions import Exit
+
+async def cmd_exit():
+    raise Exit
+
+async def cmd_exit_help():
+    normal = ['exit', 'help']
+    response = (
+        'Usage: exit\n'
+        'close the connection\n'
+    )
+    return normal, response
+
+SPEC = Str({
+    None: End(cmd_exit),
+    'help': End(cmd_exit_help),
+})

anonstream/control/spec/methods/help.py

@@ -0,0 +1,35 @@
+from anonstream.control.spec.common import Str, End
+
+async def cmd_help():
+    normal = ['help']
+    response = (
+        'Usage: METHOD [COMMAND | help]\n'
+        'Examples:\n'
+        ' help...........................show this help message\n'
+        ' exit...........................close the control connection\n'
+        ' title [show]...................show the stream title\n'
+        ' title set TITLE................set the stream title\n'
+        ' user [show]....................show a list of users\n'
+        ' user attr USER.................set an attribute of a user\n'
+        ' user get USER ATTR.............set an attribute of a user\n'
+        ' user set USER ATTR VALUE.......set an attribute of a user\n'
+        #' user kick USERS [FAREWELL].....kick users\n'
+        ' user eyes USER [show]..........show a list of active video responses\n'
+        ' user eyes USER delete EYES_ID..end a video response\n'
+        #' chat show MESSAGES.............show a list of messages\n'
+        ' chat delete SEQS...............delete a set of messages\n'
+    )
+    return normal, response
+
+async def cmd_help_help():
+    normal = ['help', 'help']
+    response = (
+        'Usage: help\n'
+        'show usage syntax and examples\n'
+    )
+    return normal, response
+
+SPEC = Str({
+    None: End(cmd_help),
+    'help': End(cmd_help_help),
+})

anonstream/control/spec/methods/title.py

@@ -0,0 +1,40 @@
+import json
+
+from anonstream.control.exceptions import Fail
+from anonstream.control.spec import Spec, NoParse
+from anonstream.control.spec.common import Str, End, ArgsJsonString
+from anonstream.control.spec.utils import get_item, json_dumps_contiguous
+from anonstream.stream import get_stream_title, set_stream_title
+
+async def cmd_title_help():
+    normal = ['title', 'help']
+    response = (
+        'Usage: title [show | set TITLE]\n'
+        'Commands:\n'
+        ' title [show].......show the stream title\n'
+        ' title set TITLE....set the stream title to TITLE\n'
+        'Definitions:\n'
+        ' TITLE..............a json string, whitespace must be \\uXXXX-escaped\n'
+    )
+    return normal, response
+
+async def cmd_title_show():
+    normal = ['title', 'show']
+    response = json.dumps(await get_stream_title()) + '\n'
+    return normal, response
+
+async def cmd_title_set(title):
+    try:
+        await set_stream_title(title)
+    except OSError as e:
+        raise Fail(f'could not set title: {e}') from e
+    normal = ['title', 'set', json_dumps_contiguous(title)]
+    response = ''
+    return normal, response
+
+SPEC = Str({
+    None: End(cmd_title_show),
+    'help': End(cmd_title_help),
+    'show': End(cmd_title_show),
+    'set': ArgsJsonString(End(cmd_title_set)),
+})

anonstream/control/spec/methods/user.py

@@ -0,0 +1,135 @@
+import json
+
+from quart import current_app
+
+from anonstream.control.exceptions import Fail
+from anonstream.control.spec import NoParse
+from anonstream.control.spec.common import Str, End, ArgsInt, ArgsString, ArgsJson, ArgsJsonString
+from anonstream.control.spec.utils import get_item, json_dumps_contiguous
+from anonstream.utils.user import USER_WEBSOCKET_ATTRS
+
+USERS_BY_TOKEN = current_app.users_by_token
+USERS = current_app.users
+USERS_UPDATE_BUFFER = current_app.users_update_buffer
+
+class ArgsJsonTokenUser(ArgsJsonString):
+    def transform_obj(self, token):
+        try:
+            user = USERS_BY_TOKEN[token]
+        except KeyError:
+            raise NoParse(f'no user with token {token!r}')
+        return user
+
+def ArgsUser(spec):
+    return Str({
+        'token': ArgsJsonTokenUser(spec),
+        #'hash': ArgsJsonHashUser(spec),
+    })
+
+async def cmd_user_help():
+    normal = ['help']
+    response = (
+            'Usage: user [show | attr USER | get USER ATTR | set USER ATTR VALUE]\n'
+            'Commands:\n'
+            ' user [show].......................show all users\' tokens\n'
+            ' user attr USER....................show names of a user\'s attributes\n'
+            ' user get USER ATTR................show an attribute of a user\n'
+            ' user set USER ATTR................set an attribute of a user\n'
+            ' user eyes USER [show].............show a user\'s active video responses\n'
+            ' user eyes USER delete EYES_ID.....end a video response to a user\n'
+            'Definitions:\n'
+            #' USER..............................={token TOKEN | hash HASH}\n'
+            ' USER..............................=token TOKEN\n'
+            ' TOKEN..............................a token, json string\n'
+            #' HASH..............................a token hash\n'
+            ' ATTR...............................a user attribute, re:[a-z0-9_]+\n'
+            ' EYES_ID............................a user\'s eyes_id, base 10 integer\n'
+    )
+    return normal, response
+
+async def cmd_user_show():
+    normal = ['user', 'show']
+    response = json.dumps(tuple(USERS_BY_TOKEN)) + '\n'
+    return normal, response
+
+async def cmd_user_attr(user):
+    normal = ['user', 'attr', 'token', json_dumps_contiguous(user['token'])]
+    response = json.dumps(tuple(user.keys())) + '\n'
+    return normal, response
+
+async def cmd_user_get(user, attr):
+    try:
+        value = user[attr]
+    except KeyError as e:
+        raise Fail('user has no such attribute') from e
+    try:
+        value_json = json.dumps(value)
+    except (TypeError, ValueError) as e:
+        raise Fail('value is not representable in json') from e
+    normal = [
+        'user',
+        'get',
+        'token',
+        json_dumps_contiguous(user['token']),
+        attr,
+    ]
+    response = value_json + '\n'
+    return normal, response
+
+async def cmd_user_set(user, attr, value):
+    if attr not in user:
+        raise Fail(f'user has no attribute {attr!r}')
+    user[attr] = value
+    if attr in USER_WEBSOCKET_ATTRS:
+        USERS_UPDATE_BUFFER.add(user['token'])
+    normal = [
+        'user',
+        'set',
+        'token',
+        json_dumps_contiguous(user['token']),
+        attr,
+        json_dumps_contiguous(value),
+    ]
+    response = ''
+    return normal, response
+
+async def cmd_user_eyes_show(user):
+    normal = [
+        'user',
+        'eyes',
+        'token',
+        json_dumps_contiguous(user['token']),
+        'show'
+    ]
+    response = json.dumps(user['eyes']['current']) + '\n'
+    return normal, response
+
+async def cmd_user_eyes_delete(user, eyes_id):
+    try:
+        user['eyes']['current'].pop(eyes_id)
+    except KeyError:
+        pass
+    normal = [
+        'user',
+        'eyes',
+        'token',
+        json_dumps_contiguous(user['token']),
+        'delete',
+        str(eyes_id),
+    ]
+    response = ''
+    return normal, response
+
+SPEC = Str({
+    None: End(cmd_user_show),
+    'help': End(cmd_user_help),
+    'show': End(cmd_user_show),
+    'attr': ArgsUser(End(cmd_user_attr)),
+    'get': ArgsUser(ArgsString(End(cmd_user_get))),
+    'set': ArgsUser(ArgsString(ArgsJson(End(cmd_user_set)))),
+    'eyes': ArgsUser(Str({
+        None: End(cmd_user_eyes_show),
+        'show': End(cmd_user_eyes_show),
+        'delete': ArgsInt(End(cmd_user_eyes_delete)),
+    })),
+})

anonstream/control/spec/utils.py

@@ -0,0 +1,19 @@
+import json
+
+def get_item(index, words):
+    try:
+        word = words[index]
+    except IndexError:
+        word = None
+    else:
+        if not word:
+            raise NoParse(f'empty word at position {index}')
+    return word
+
+def json_dumps_contiguous(obj, **kwargs):
+    return json.dumps(obj, **kwargs).replace(' ', r'\u0020')
+
+def startswith(string, prefix):
+    if string is None or prefix is None:
+        return string is prefix
+    return string.startswith(prefix)

anonstream/control/utils.py

@@ -1,4 +0,0 @@
-import json
-
-def json_dumps_contiguous(obj, **kwargs):
-    return json.dumps(obj, **kwargs).replace(' ', r'\u0020')

anonstream/events.py

@@ -0,0 +1,30 @@
+import asyncio
+import json
+
+from quart import current_app
+
+async def start_event_server_at(address):
+    return await asyncio.start_unix_server(serve_event_client, address)
+
+async def serve_event_client(reader, writer):
+    reader.feed_eof()
+    queue = asyncio.Queue()
+    current_app.event_queues.add(queue)
+    try:
+        while True:
+            event = await queue.get()
+            event_json = json.dumps(event, separators=(',', ':'))
+            writer.write(event_json.encode())
+            writer.write(b'\n')
+            try:
+                await writer.drain()
+            # Because we used reader.feed_eof(), if the client sends anything
+            # an AsserionError will be raised
+            except (ConnectionError, AssertionError):
+                break
+    finally:
+        current_app.event_queues.remove(queue)
+
+def notify_event_sockets(event):
+    for queue in current_app.event_queues:
+        queue.put_nowait(event)

anonstream/helpers/chat.py

@@ -9,7 +9,7 @@ CONFIG = current_app.config
 
 def generate_nonce_hash(nonce):
     parts = CONFIG['SECRET_KEY'] + b'nonce-hash\0' + nonce.encode()
-    return hashlib.sha256(parts).digest()
+    return hashlib.sha256(parts).hexdigest()
 
 def get_scrollback(messages):
     n = CONFIG['MAX_CHAT_SCROLLBACK']

config.toml

@@ -1,8 +1,13 @@
 secret_key = "place secret key here"
 
-[control]
+[socket.control]
+enabled = true
 address = "control.sock"
 
+[socket.event]
+enabled = true
+address = "event.sock"
+
 [auth]
 username = "broadcaster"