Отделение сервиса

В самом начале мы для примера реализовали три команды: get, set, update, которые напрямую управляли состоянием приложения. Состояние при этом хранилось в памяти и было доступно лишь для одного процесса. Чтобы использовать общее состояние разными процессами, запущенными на разных машинах, нужно хранить его в базе данных на выделенном сервере, и при каждом обращении к данным брать его оттуда.

В качестве примера возьмем ту же задачу, но данные будем хранить для простоты не в БД, а в файле:

class MyController:
    async def handle_command(self, storage, index, command, result):
        key = command.get("key")
        key = escape(key)  # Never trust data from users
        filename = "../data/save_" + key
        dirname = os.path.dirname(filename)
        if not os.path.exists(dirname):
            os.makedirs(dirname)
        code = command.get("code")
        all_indexes = storage.get("indexes")
        if code == "save" or code == "set":
            state = command.get("state")
            with open(filename, "w") as f:
                json.dump(state, f)
                f.flush()
            result.append((all_indexes, [{**command, "success": True}]))
        elif code == "load" or code == "get":
            if not os.path.exists(filename):
                return {"success": False, **command}, None
            with open(filename, "r") as f:
                state = json.load(f)
            result.append(([index], [{**command, "success": True, "state": state}]))
        elif code == "update":
            index = command.get("index")
            value = command.get("value")
            if not isinstance(index, int) or not isinstance(value, int):
                result.append(([index], [{**command, "success": False}]))
                return
            if not os.path.exists(filename):
                state = []
            else:
                with open(filename, "r") as f:
                    state = json.load(f)
            if index >= len(state):
                state += [0] * (index - len(state) + 1)
            state[index] = value
            with open(filename, "w") as f:
                json.dump(state, f)
                f.flush()
            result.append((all_indexes, [{**command, "success": success}]))

def escape(string):
    return re.sub(r"[^A-Za-z]+", "", string)

Как мы и говорили раньше — мешанина. Давайте вычленим отсюда отдельные примеси, которые можно объединить по некоему общему признаку и выделить в отдельный класс. Прежде всего можно вынести работу с файлами. Поскольку работа с файлами — функция служебная и чисто техническая, то есть не содержащая бизнес-логики, то назовем данный тип классов службами (Service), а нашу реализацию с файлами — FileService.

class StorageService:
    def save(self, key, data):
        pass

    def load(self, key):
        return None

class FileService(StorageService):
    filename_prefix = ""

    def __init__(self, filename_prefix=None):
        self.filename_prefix = filename_prefix or self.filename_prefix

    def save(self, key, data):
        key = escape(key)  # Never trust data from users
        filename = self.filename_prefix + key
        dirname = os.path.dirname(filename)
        if dirname and not os.path.exists(dirname):
            os.makedirs(dirname)
        with open(filename, "w") as f:
            json.dump(data, f)
            f.flush()
            print("save", data)

    def load(self, key):
        key = escape(key)  # Never trust data from users
        filename = self.filename_prefix + key
        if not os.path.exists(filename):
            return None
        with open(filename, "r") as f:
            data = json.load(f)
            print("load", data)
            return data

class MyController:
    # Settings
    service_factory = lambda self: FileService("../data/save_")
    model_factory = lambda self, *args: MyModel(*args)

    def __init__(self):
        self.service = self.service_factory()
        self.model = self.model_factory(self.service)

    async def handle_command(self, storage, index, command, result):
        code = command.get("code")
        key = command.get("key")
        all_indexes = storage.get("indexes")
        if code == "save" or code == "set":
            state = command.get("state")
            success = self.service.save(key, state)
            result.append((all_indexes, [{**command, "success": success}]))
        elif code == "load" or code == "get":
            state = self.service.load(key)
            result.append(([index], [{**command, "success": True, "state": state}]))
        elif code == "update":
            index = command.get("index")
            value = command.get("value")
            state = self.service.load(key)
            if not isinstance(index, int) or not isinstance(value, int):
                success = False
            else:
                if state is None:
                    state = []
                if index >= len(state):
                    state += [0] * (index - len(state) + 1)
                state[index] = value
                success = True
            self.service.save(key, state)
            result.append((all_indexes, [{**command, "success": success}]))

Только взгляните на handle_command() — насколько сразу стало легче дышать! Но выгода не только в этом. Теперь весь функционал по сохранению и загрузки из файла собран в одном месте. А это значит, что нам теперь не нужно рыскать по коду, если нужно сменить имя файла или сохраняемый формат с JSON на YML, например. Достаточно найти соответствующий класс.

Более того. Мы можем реализовать в другом классе тот же интерфейс, но уже не сохранять в файл, а, например, пересылать данные на удаленный сервер (например, через HTTP-запрос) или помещать их в базу данных (БД). Ведь контроллеру все равно, как сервис выполнит свою задачу — он лишь вызывает методы load() и save(). С точки зрения логики (контроллера), не нужно сохранять данные именно в файл. Нужно просто сохранить данные. И не нужно загружать данные обязательно только из файла. Данные просто нужны, и поэтому их нужно загрузить — все равно откуда.

Отделение модели

Вторая категория функций, которые реализуют контроллеры — это обработка данных. Этот функционал можно также успешно вынести из контроллеров в отдельные классы, которые называются моделями.

Почему модели? Всякое приложение отражает ту или иную область деятельности человека из реальной жизни, которая называется предметной областью приложения. Но реальность нам не нужна во всей ее полноте. Нам нужны только те существенные ее стороны, которые помогают решить поставленную задачу. Эти грани реальности формализуются, параметризуются, и в результате на их основе строится модель. Модель, в узком смысле этого слова — это класс, в который входят настройки, состояние и алгоритмы обработки этих данных. В широком смысле модель — это вообще все приложение.

Однако, само наличие данных еще не означает, что перед нами модель. Объект хранилища (storage) просто хранит абстрактные данные, и он абсолютно одинаковый для всех предметных областей. Поэтому хранилище не является моделью, хотя данные и объекты, которые относятся к модели, и хранятся в нем.

Данные модели — это обычные абстрактные JSON-объекты (словари и списки), и определенную осмысленность они обретают только в функциях обработки данных. То есть только во время обработки просто данные становятся данными модели. Получается, в этих функциях воплощены оба аспекта модели: и данные (точнее, их формат: имена и типы), и алгоритмы обработки данных. Вот почему моделями (Model) мы называем только те классы, которые обрабатывают данные, а не те, которые их хранят.

В данном примере обработки данных не много — хватило лишь на один метод. Но для иллюстрации принципа вполне достаточно и этого:

class MyModel:
    def update(self, state, index, value):
        if not isinstance(index, int) or not isinstance(value, int):
            return False
        if state is None:
            state = []
        if index >= len(state):
            state += [0] * (index - len(state) + 1)
        state[index] = value
        return True

class MyController:
    # Settings
    service_factory = lambda self: FileService("../data/save_")
    model_factory = lambda self, *args: MyModel(*args)

    def __init__(self):
        self.service = self.service_factory()
        self.model = self.model_factory(self.service)

    async def handle_command(self, storage, index, command, result):
        code = command.get("code")
        key = command.get("key")
        if code == "save" or code == "set":
            state = command.get("state")
            success = self.service.save(key, state)
            result.append((all_indexes, [{**command, "success": success}]))
        elif code == "load" or code == "get":
            state = self.service.load(key)
            result.append(([index], [{**command, "success": True, "state": state}]))
        elif code == "update":
            index = command.get("index")
            value = command.get("value")
            state = self.service.load(key)
            success = self.model.update(state, index, value)
            self.service.save(key, state)
            result.append((all_indexes, [{**command, "success": success}]))

Использование классов модели дает нам все те же преимущества, что мы описывали и раньше:

Изменившаяся роль контроллера

Если посмотреть внимательнее на получившийся контроллер, то можно увидеть, что в нем фактически не осталось логики в чистом виде — одни вызовы методов. Все свои функции контроллер делегировал двум большим категориям классов: моделям и сервисам. Себе же он оставил функции диспетчера между состоянием (storage), моделями (model) и сервисами (service). Диспетчера и обработчика команд (command).

Все контроллеры построены по одному шаблону. Контроллер смотрит на имя команды и переходит к соответствующему блоку кода для ее обработки. Далее из объекта команды достаются переменные, а из хранилища берутся нужные объекты состояния. Затем все вместе они передаются в виде аргументов моделям и сервисам.

Последние оперируют только с теми объектами состояния, которые им действительно необходимы. Всё состояние приложения обычно не передается в модель или сервис, чтобы они сами выбрали, что им нужно. Это не их дело — ни модель, ни сервис не должны знать о структуре объектов состояния. Об этом по возможности должен знать только контроллер. Таким образом, контроллер занимается не только координацией совместной работы моделей и сервисов, но и всей подготовкой и подбором данных для них.

Отделение протокола

Кроме того, он еще отвечает за обработку результатов и подготовку ответных команд клиентам. Тем самым на его долю выпадает вся работа с командами. Фактически контроллер — единственный, кто знает формат команд.

Сервер оперирует просто потоком байтов, парсер — преобразует его в абстрактные JSON-объекты и обратно, Application знает ровно столько, чтобы подобрать подходящий контроллер. Но тип команды, операция, которую она выполняет, параметры этой операции и их назначение — все это известно лишь контроллеру. Модели и сервисы получают их уже в виде параметров своих функций — они тоже не знают ничего о командах.

Таким образом, контроллер реализует еще и протокол приложения. То есть тот формат команд, который используется сервером и который должен возвращать парсер. Клиенты могут использовать разный формат данных — XML, JSON или YML, но парсер всегда должен возвращать питоновские объекты: dict и list, как после обыкновенного JSON. Клиенты могут использовать разные имена и значения параметров, разные коды команд, но парсер должен всегда возвращать те имена, значения и коды, которые приняты у нас на сервере, которые понимают контроллеры. Это то, что и называется протоколом приложения. И он используется нигде больше, кроме как в контроллерах.

Получается, что один класс контроллера реализует сразу две функции: протокола приложения и диспетчера. А у нас уже сформировалось такое правило: если в классе обнаруживается два функционала, мы делаем из них два класса. Или, другими словами: ein Reich — ein Fü... один класс — одна функция, одно назначение.

Устраним этот недостаток. Создадим отдельно:

1. протокольный контроллер, который отвечает только за разбор команд, и

2. функциональный контроллер, который работает с моделями, сервисами и состоянием и наследуется от протокольного.

При этом первый будет шаблоном контроллера, а второй — его конкретной реализацией:

class MyProtocolController:
    async def handle_command(self, storage, index, command, result):
        code = command.get("code")
        key = command.get("key")
        all_indexes = storage.get("indexes")
        if code == "save" or code == "set":
            state = command.get("state")
            success = self.save(key, state)
            result.append((all_indexes, [{**command, "success": success}]))
        elif code == "load" or code == "get":
            state = self.load(key)
            result.append(([index], [{**command, "success": True, "state": state}]))
        elif code == "update":
            index = command.get("index")
            value = command.get("value")
            self.update(key, index, value, all_indexes, command, result)

    def save(self, key, state):
        pass

    def load(self, key):
        pass

    def update(self, key, index, value, all_indexes, command, result):
        pass

class MyController(MyProtocolController):
    # Settings
    service_factory = lambda self: FileService("../data/save_")
    model_factory = lambda self, *args: MyModel(*args)

    def __init__(self):
        self.service = self.service_factory()
        self.model = self.model_factory(self.service)

    def save(self, key, state):
        success = self.service.save(key, state)
        return success

    def load(self, key):
        state = self.service.load(key)
        return state

    def update(self, key, index, value, all_indexes, command, result):
        state = await self.service.load(key)
        success = self.model.update(state, index, value)
        if success:
            success = self.service.save(key, state)
        result.append((all_indexes, [{**command, "success": success}]))

В некоторых случаях допускается, чтобы информация о командах проникала в реализацию контроллера. Это бывает нужно в некоторых особенно сложных случаях, например, если генерируется несколько ответных команд, каждую из которых нужно рассылать особому адресату. И хоть у нас пока случай далеко не такой, но мы все равно реализовали для примера одну функцию в этом ключе (update()).

Внутренние и отложенные команды

Жизненный цикл приложения на данный момент очень простой: получаются команды, они обрабатываются, и результат отсылается назад. Если команда не была получена, то нечего и отсылать. Ясно, что для серьезных сложных приложений этого мало. Практически в каждой игре должен быть внутренний цикл работы или хотя бы таймер, по которому игра сама могла бы в нужный момент времени посылать команды. По собственному почину, а не только по запросу пользователя, как сейчас. Должна существовать возможность приложению самому инициировать события.

Так как все события у нас — это команды, то и для реализации внутреннего цикла приложения нет смысла отступать от общей канвы. Команды для этого очень даже подходят. А раз они не должны быть отосланы клиентам, то и индексы им не нужны. Вместо индексов будем использовать None. Подобные служебные команды с индексом None будем называть внутренними. А те, которые отсылаются клиентам или полученные от них, то есть ассоциированные с определенными индексами соединений — внешними.

Бывают команды, выполнение которых должно происходить не сразу, а через какой-то интервал времени, который можно указать в поле after_ms. Например, во время перехода хода к другому игроку можно в объекте игры указать время окончания хода и добавить команду "try_end_turn" c полем after_ms, которая через заданный интервал это время сравнит с текущим. (Мы не можем добавить сразу "end_turn" — без проверки, потому что пока команда будет ожидать своего часа, ход может несколько раз смениться, и команды выполнится не тогда, когда нужно.) Если команда должна выполняться больше одного раза, то есть периодически, то на этот случай предусмотрено поле period_ms. Также можно указать не относительное, а абсолютное время — поле at_ms.

Такие команды, выполнение которых отложено во времени, будем называть отложенными:

class SocketApplication:
    @property
    def time_ms(self):
        # Current app time
        return int(time.time() - self.storage.get("start_app_time_ms") * 1000)

    def __init__(self, default_controller, controller_by_key=None, storage=None) -> None:
        super().__init__()
        self.default_controller = default_controller
        self.controller_by_key = controller_by_key or {}
        self.storage = storage if storage else Repository()  # App state
        self.storage.get("start_app_time_ms", time.time())
        self.command_queue = self.storage.get("command_queue", [])
    # ...
    async def handle_commands(self, index, commands):
        result = []
        # Handle
        for command in commands:
            key = escape(command.get("key"))
            controller = self.controller_by_key.get(key, self.default_controller)
            if controller:
                await controller.handle_command(self.storage, index, command, result)
        # Handle internal or enqueue if deferred
        for indexes, commands in result:
            if indexes is None:
                result += self.handle_internal(commands)
        return result

    def handle_internal(self, commands):
        # Enqueue commands with time
        handle_now_list = []
        for command in commands:
            # Push each to self.command_queue or handle_now_list
            ...
        # Handle other
        result = await self.handle_commands(None, commands)
        for indexes, c in result:
            if indexes is None:
                result += self.handle_internal(c)
        return result

    def handle_deferred(self):
        commands = []
        periodical_commands = []
        # Find commands which time has come
        for command in self.command_queue.copy():
            self.command_queue.remove(command)
            at_ms = command.get("at_ms")
            if at_ms is None or at_ms < 0:
                # Note: All commands in queue should have at_ms.
                # If not, it's same as being marked deleted.
                continue
            elif at_ms > self.time_ms:
                # No more commands to handle
                break
            else:
                # Handle
                commands.append(command)
                # Check periodical
                period_ms = command.get("period_ms")
                if period_ms is not None and period_ms > 0:
                    # command["at_ms"] = None  # Ok, but slower than following
                    command["at_ms"] = self.time_ms + period_ms
                    periodical_commands.append(command)
        # Handle
        result = self.handle_commands(None, commands)
        # Enqueue periodical back
        result += self.handle_internal(periodical_commands)
        return result

В командах используется внутреннее время приложения, в котором отсчитываются миллисекунды с момента его запуска. Чтобы игру можно было перезагружать и продолжать с того же места, очередь команд и начало отсчета времени сохраняется в хранилище. Для порядка тут не хватает еще вычитания времени, когда приложение не работало (pause), но это несложно додумать и самим;).

Поля period_ms и after_ms в конечном счете преобразуются в at_ms, по которому все команды и сортируются в очереди command_queue. Очередь проверяется несколько раз за секунду (handle_deferred()), поэтому она должна быть отсортированной. Иначе нам придется перебирать каждый раз все элементы очереди.

class SocketServer:
    # ...
    async def main(self):
        print(f"Start server: {self.host}:{self.port}")
        server = await asyncio.start_server(self.handle_connection, self.host, self.port)
        async with server:
            self.is_running = True
            while self.is_running:
                result = self.application.handle_deferred()
                self.send(result)
                await asyncio.sleep(.1)

Переход к DDD-разработке